本篇內(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ā)送和接收
到此,相信大家對(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://aaarwkj.com/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)站立場,如需處理請(qǐng)聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時(shí)需注明來源: 創(chuàng)新互聯(lián)