(十七) 整合spring cloud雲架構 -消息驅動 Spring Cloud Stream

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

簡介：

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

使用工具：

rabbit，具體的下載下傳和安裝細節我這裡不做太多講解，網上的執行個體太多了

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

<dependency>  
   <groupId>org.springframework.cloud</groupId>  
   <artifactId>spring-cloud-starter-stream-rabbit</artifactId>  
</dependency></span>

在yml檔案裡面配置rabbit mq

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>

定義接口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>

定義綁定

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

定義發送消息業務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>

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

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

在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

定義接口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>  
  
}

定義啟動類和消息消費

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>  
  
}

分别啟動commonservice-mq-producer、commonservice-mq-consumer1
通過postman來驗證消息的發送和接收

(十七) 整合spring cloud雲架構 -消息驅動 Spring Cloud Stream

可以看到接收到了消息，下一章我們介紹mq的叢集方案。

到此，整個消息中心方案內建完畢（企業架構源碼可以加求球：叁五三陸二肆柒二伍玖）！！

歡迎大家和我一起學習spring cloud建構微服務雲架構，我這邊會将近期研發的spring cloud微服務雲架構的搭建過程和精髓記錄下來，幫助更多有興趣研發spring cloud架構的朋友，大家來一起探讨spring cloud架構的搭建過程及如何運用于企業項目。

(十七) 整合spring cloud雲架構 -消息驅動 Spring Cloud Stream

繼續閱讀

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的簡單使用

neo4j之cypher使用文檔

GitHub連夜封殺！這份阿裡 10W 字内部 Java 字面試手冊到底有多強？

spark/scala關于【資源檔案】加載方法概述外部檔案加載方案測試資源檔案打包入jar包中小結

mybatis_入門程式Mybatis入門

vue-cli簡介（中文翻譯）

AOP程式設計_Android優雅權限架構(1)概念基礎，2021金三銀四前言正文大綱正文

Ajax發送和擷取json資料到Spring mvc 1.spring mvc後端2.web前段

Effective Java 8:通用程式設計

OOM三種類型

工廠模式-三種類型

【遞歸】高效率求2的n次幂

win10本地scala和spark安裝安裝scala安裝spark

scala (3) Function 和 Method

JSONObject包導入異常 java.lang.NoClassDefFoundErrorweb項目的導入包的問題