(十七) Java版本多用戶B2B2C商城源碼 -消息驅動 Spring Cloud Stream

在使用spring cloud雲架構的時候，咱們不得不使用Spring cloud Stream，由於消息中間件的使用在項目中無處不在，咱們公司後面作了娛樂方面的APP，在使用spring cloud作架構的時候，其中消息的異步通知，業務的異步處理都須要使用消息中間件機制。spring cloud的官方給出的集成建議（使用rabbit mq和kafka），我看了一下源碼和配置，只要把rabbit mq集成，kafka只是換了一個pom配置jar包而已，閒話少說，咱們就直接進入配置實施：

1. 簡介：

Spring cloud Stream 數據流操做開發包，封裝了與Redis,Rabbit、Kafka等發送接收消息。

2. 使用工具：

rabbit，具體的下載和安裝細節我這裏不作太多講解，網上的實例太多了

3. 建立commonservice-mq-producer消息的發送者項目，在pom裏面配置stream-rabbit的依賴

<span style="font-size: 16px;"><!-- 引入MQ消息驅動的微服務包，引入stream只須要進行配置化便可，是對rabbit、kafka很好的封裝 -->  <dependency>      <groupId>org.springframework.cloud</groupId>      <artifactId>spring-cloud-starter-stream-rabbit</artifactId>  </dependency></span>  複製代碼

 4. 在yml文件裏面配置rabbit mq

<span style="font-size: 16px;">server:    port: 5666  spring:    application:      name: commonservice-mq-producer    profiles:       active: dev    cloud:      config:        discovery:           enabled: true          service-id: commonservice-config-server    <span style="color: #ff0000;"># rabbitmq和kafka都有相關配置的默認值，若是修改，能夠再次進行配置 stream: bindings: mqScoreOutput: destination: honghu_exchange contentType: application/json rabbitmq: host: localhost port: 5672 username: honghu password: honghu</span> eureka: client: service-url: defaultZone: http://honghu:123456@localhost:8761/eureka instance: prefer-ip-address: true</span> 複製代碼

 5. 定義接口ProducerService

<span style="font-size: 16px;">package com.honghu.cloud.producer;    import org.springframework.cloud.stream.annotation.Output;  import org.springframework.messaging.SubscribableChannel;    public interface ProducerService {            String SCORE_OUPUT = "mqScoreOutput";            @Output(ProducerService.SCORE_OUPUT)      SubscribableChannel sendMessage();  }</span>  複製代碼

 6. 定義綁定

<span style="font-size: 16px;">package com.honghu.cloud.producer;    import org.springframework.cloud.stream.annotation.EnableBinding;    @EnableBinding(ProducerService.class)  public class SendServerConfig {    }</span>  複製代碼

 7. 定義發送消息業務ProducerController

<span style="font-size: 16px;">package com.honghu.cloud.controller;      import org.springframework.beans.factory.annotation.Autowired;  import org.springframework.integration.support.MessageBuilder;  import org.springframework.messaging.Message;  import org.springframework.web.bind.annotation.PathVariable;  import org.springframework.web.bind.annotation.RequestBody;  import org.springframework.web.bind.annotation.RequestMapping;  import org.springframework.web.bind.annotation.RequestMethod;  import org.springframework.web.bind.annotation.RestController;    import com.honghu.cloud.common.code.ResponseCode;  import com.honghu.cloud.common.code.ResponseVO;  import com.honghu.cloud.entity.User;  import com.honghu.cloud.producer.ProducerService;    import net.sf.json.JSONObject;    @RestController  @RequestMapping(value = "producer")  public class ProducerController {            @Autowired      private ProducerService producerService;                  /**      * 經過get方式發送</span>對象<span style="font-size: 16px;">      * @param name 路徑參數      * @return 成功|失敗      */      @RequestMapping(value = "/sendObj", method = RequestMethod.GET)      public ResponseVO sendObj() {          User user = new User(1, "hello User");          <span style="color: #ff0000;">Message<User> msg = MessageBuilder.withPayload(user).build();</span>          boolean result = producerService.sendMessage().send(msg);          if(result){              return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_SUCCESS, false);          }          return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_FAILURE, false);      }                  /**      * 經過get方式發送字符串消息      * @param name 路徑參數      * @return 成功|失敗      */      @RequestMapping(value = "/send/{name}", method = RequestMethod.GET)      public ResponseVO send(@PathVariable(value = "name", required = true) String name) {          Message msg = MessageBuilder.withPayload(name.getBytes()).build();          boolean result = producerService.sendMessage().send(msg);          if(result){              return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_SUCCESS, false);          }          return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_FAILURE, false);      }            /**      * 經過post方式發送</span>json對象<span style="font-size: 16px;">      * @param name 路徑參數      * @return 成功|失敗      */      @RequestMapping(value = "/sendJsonObj", method = RequestMethod.POST)      public ResponseVO sendJsonObj(@RequestBody JSONObject jsonObj) {          Message<JSONObject> msg = MessageBuilder.withPayload(jsonObj).build();          boolean result = producerService.sendMessage().send(msg);          if(result){              return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_SUCCESS, false);          }          return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_FAILURE, false);      }  }  </span>  複製代碼

8. 建立commonservice-mq-consumer1消息的消費者項目，在pom裏面配置stream-rabbit的依賴

<!-- 引入MQ消息驅動的微服務包，引入stream只須要進行配置化便可，是對rabbit、kafka很好的封裝 -->  <dependency>      <groupId>org.springframework.cloud</groupId>      <artifactId>spring-cloud-starter-stream-rabbit</artifactId>  </dependency>  複製代碼

 9. 在yml文件中配置：

server:    port: 5111  spring:    application:      name: commonservice-mq-consumer1    profiles:       active: dev    cloud:      config:        discovery:           enabled: true          service-id: commonservice-config-server                <span style="color: #ff0000;">stream:        bindings:          mqScoreInput:            group: honghu_queue            destination: honghu_exchange            contentType: application/json                rabbitmq:       host: localhost       port: 5672       username: honghu       password: honghu</span>  eureka:     client:      service-url:        defaultZone: http://honghu:123456@localhost:8761/eureka    instance:      prefer-ip-address: true  複製代碼

10. 定義接口ConsumerService

package com.honghu.cloud.consumer;    import org.springframework.cloud.stream.annotation.Input;  import org.springframework.messaging.SubscribableChannel;    public interface ConsumerService {            <span style="color: #ff0000;">String SCORE_INPUT = "mqScoreInput";        @Input(ConsumerService.SCORE_INPUT)      SubscribableChannel sendMessage();</span>    }  複製代碼

11. 定義啓動類和消息消費

package com.honghu.cloud;    import org.springframework.boot.SpringApplication;  import org.springframework.boot.autoconfigure.SpringBootApplication;  import org.springframework.cloud.netflix.eureka.EnableEurekaClient;  import org.springframework.cloud.stream.annotation.EnableBinding;  import org.springframework.cloud.stream.annotation.StreamListener;    import com.honghu.cloud.consumer.ConsumerService;  import com.honghu.cloud.entity.User;    @EnableEurekaClient  @SpringBootApplication  @EnableBinding(ConsumerService.class) //能夠綁定多個接口  public class ConsumerApplication {            public static void main(String[] args) {          SpringApplication.run(ConsumerApplication.class, args);      }            <span style="color: #ff0000;">@StreamListener(ConsumerService.SCORE_INPUT)      public void onMessage(Object obj) {          System.out.println("消費者1，接收到的消息：" + obj);      }</span>    }  複製代碼

12. 分別啓動commonservice-mq-producer、commonservice-mq-consumer1

13. 經過postman來驗證消息的發送和接收

能夠看到接收到了消息，下一章咱們介紹mq的集羣方案。

到此，整個消息中心方案集成完畢

歡迎你們和我一塊兒學習spring cloud構建微服務雲架構，我這邊會將近期研發的spring cloud微服務雲架構的搭建過程和精髓記錄下來，幫助更多有興趣研發spring cloud框架的朋友，你們來一塊兒探討spring cloud架構的搭建過程及如何運用於企業項目。

須要JAVASpring Cloud大型企業分佈式微服務雲構建的B2B2C電子商務平臺源碼請加企鵝求求：一零三八七七四六二六