消息驅(qū)動(dòng)SpringCloudStream怎么配置

本篇內(nèi)容主要講解“消息驅(qū)動(dòng)Spring Cloud Stream怎么配置”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實(shí)用性強(qiáng)。下面就讓小編來帶大家學(xué)習(xí)“消息驅(qū)動(dòng)Spring Cloud Stream怎么配置”吧!

專注于為中小企業(yè)提供網(wǎng)站設(shè)計(jì)、成都做網(wǎng)站服務(wù),電腦端+手機(jī)端+微信端的三站合一,更高效的管理,為中小企業(yè)從江免費(fèi)做網(wǎng)站提供優(yōu)質(zhì)的服務(wù)。我們立足成都，凝聚了一批互聯(lián)網(wǎng)行業(yè)人才，有力地推動(dòng)了上1000+企業(yè)的穩(wěn)健成長，幫助中小企業(yè)通過網(wǎng)站建設(shè)實(shí)現(xiàn)規(guī)模擴(kuò)充和轉(zhuǎn)變。

在使用spring cloud云架構(gòu)的時(shí)候，我們不得不使用Spring cloud Stream，因?yàn)橄⒅虚g件的使用在項(xiàng)目中無處不在，我們公司后面做了娛樂方面的APP，在使用spring cloud做架構(gòu)的時(shí)候，其中消息的異步通知，業(yè)務(wù)的異步處理都需要使用消息中間件機(jī)制。spring cloud的官方給出的集成建議（使用rabbit mq和kafka），我看了一下源碼和配置，只要把rabbit mq集成，kafka只是換了一個(gè)pom配置jar包而已，閑話少說，我們就直接進(jìn)入配置實(shí)施：

1. 簡介：

Spring cloud Stream 數(shù)據(jù)流操作開發(fā)包，封裝了與redis,Rabbit、Kafka等發(fā)送接收消息。

2. 使用工具：

rabbit，具體的下載和安裝細(xì)節(jié)我這里不做太多講解，網(wǎng)上的實(shí)例太多了

3. 創(chuàng)建commonservice-mq-producer消息的發(fā)送者項(xiàng)目，在pom里面配置stream-rabbit的依賴

<span style="font-size: 16px;"><!-- 引入MQ消息驅(qū)動(dòng)的微服務(wù)包，引入stream只需要進(jìn)行配置化即可，是對(duì)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都有相關(guān)配置的默認(rèn)值，如果修改，可以再次進(jìn)行配置  
    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. 定義發(fā)送消息業(yè)務(wù)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方式發(fā)送</span>對(duì)象<span style="font-size: 16px;"> 
     * @param name 路徑參數(shù) 
     * @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方式發(fā)送字符串消息 
     * @param name 路徑參數(shù) 
     * @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方式發(fā)送</span>json對(duì)象<span style="font-size: 16px;"> 
     * @param name 路徑參數(shù) 
     * @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. 創(chuàng)建commonservice-mq-consumer1消息的消費(fèi)者項(xiàng)目，在pom里面配置stream-rabbit的依賴

<!-- 引入MQ消息驅(qū)動(dòng)的微服務(wù)包，引入stream只需要進(jìn)行配置化即可，是對(duì)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. 定義啟動(dòng)類和消息消費(fèi)

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) //可以綁定多個(gè)接口  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("消費(fèi)者1，接收到的消息：" + obj);  
    }</span>  
  
}

12. 分別啟動(dòng)commonservice-mq-producer、commonservice-mq-consumer1

13. 通過postman來驗(yàn)證消息的發(fā)送和接收

 消息驅(qū)動(dòng)Spring Cloud Stream怎么配置

消息驅(qū)動(dòng)Spring Cloud Stream怎么配置

消息驅(qū)動(dòng)Spring Cloud Stream怎么配置

消息驅(qū)動(dòng)Spring Cloud Stream怎么配置

消息驅(qū)動(dòng)Spring Cloud Stream怎么配置

消息驅(qū)動(dòng)Spring Cloud Stream怎么配置

到此，相信大家對(duì)“消息驅(qū)動(dòng)Spring Cloud Stream怎么配置”有了更深的了解，不妨來實(shí)際操作一番吧！這里是創(chuàng)新互聯(lián)網(wǎng)站，更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢，關(guān)注我們，繼續(xù)學(xué)習(xí)！

文章名稱：消息驅(qū)動(dòng)SpringCloudStream怎么配置
網(wǎng)頁網(wǎng)址：http://www.muchs.cn/article0/gjjoio.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供自適應(yīng)網(wǎng)站、域名注冊(cè)、網(wǎng)站設(shè)計(jì)、定制開發(fā)、響應(yīng)式網(wǎng)站、網(wǎng)站收錄

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來源： 創(chuàng)新互聯(lián)

營銷型網(wǎng)站建設(shè)