Springcloud整合stream,rabbitmq實(shí)現(xiàn)消息驅(qū)動(dòng)功能
springcloud整合stream,rabbitmq實(shí)現(xiàn)消息驅(qū)動(dòng)功能
1.代碼實(shí)現(xiàn):
創(chuàng)建項(xiàng)目stream
添加依賴
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.6.2</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.cxh</groupId>
<artifactId>stream</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>stream</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>8</java.version>
<spring-cloud-alibaba-dependencies.version>2021.1</spring-cloud-alibaba-dependencies.version>
<spring-cloud-dependencies.version>2021.0.0</spring-cloud-dependencies.version>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dependencies</artifactId>
<version>${spring-cloud-dependencies.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-alibaba-dependencies</artifactId>
<version>${spring-cloud-alibaba-dependencies.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
</dependencies>監(jiān)聽類
@EnableBinding(Sink.class)
public class SinkReceiver {
private static Logger logger = LoggerFactory.getLogger(SinkReceiver.class);
@StreamListener(Sink.INPUT)
public void receive(String payload) {
logger.info("Received: " + payload);
}
}2.實(shí)現(xiàn)效果:
啟動(dòng)rabbitmq, 項(xiàng)目stream
打開瀏覽器http://localhost:15672/,使用賬號(hào)密碼guest登錄rabbitmq, 在隊(duì)列中發(fā)現(xiàn)消息:
測(cè)試send

查看控制臺(tái)消息:
com.cxh.stream.SinkReceiver : Received: 測(cè)試send
補(bǔ)充
Spring Cloud Stream專門用于事件驅(qū)動(dòng)的微服務(wù)系統(tǒng),使用消息中間件來(lái)收發(fā)信息。使用Spring Cloud Stream可專注于業(yè)務(wù)開發(fā),而不用花太多心思在應(yīng)用與MQ之間的交互上。而且,在切換MQ后,也無(wú)須做太多的代碼改動(dòng)。
所以Spring Cloud Stream和RabbitMQ還可以整合實(shí)現(xiàn)消息的收發(fā)

整合過(guò)程
添加依賴
<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-stream-binder-rabbit</artifactId> </dependency>
不同的MQ使用不同的依賴,非常容易切換。
定義處理收發(fā)的方法
隊(duì)列無(wú)非就是收和發(fā),所以我們要先定義好,怎么樣發(fā),怎么樣收。
發(fā)送消息:
@Bean
public Supplier<String> pkslowSource() {
return () -> {
String message = "www.pkslow.com";
log.info("Sending value: " + message);
return message;
};
}只發(fā)送一個(gè)String,一般業(yè)務(wù)通常為Entity類。這里發(fā)送的內(nèi)容也固定不變,實(shí)際業(yè)務(wù)可以通過(guò)查數(shù)據(jù)庫(kù),讀文件等方式獲取數(shù)據(jù)源。
接收消息:
@Bean
public Consumer<String> pkslowSink() {
return message -> {
log.info("Received message " + message);
};
}直接打印消息即可,項(xiàng)目中的邏輯可按具體業(yè)務(wù)實(shí)現(xiàn)。
配置屬性
配置RabbitMQ:
spring:
rabbitmq:
host: localhost
port: 5672
username: pkslow
password: 123456
配置Spring Cloud Stream的相關(guān)項(xiàng):
spring:
cloud:
stream:
function:
definition: pkslowSource;pkslowSink
bindings:
pkslowSource-out-0:
destination: pkslow-topic
pkslowSink-in-0:
destination: pkslow-topic
poller:
fixed-delay: 500
spring.cloud.stream.function.definition會(huì)定義處理方法,如本文的收發(fā)消息的方法;
bindings配置對(duì)應(yīng)的function;destination指向MQ的主題;
這里配了一個(gè)poller,每隔500ms就會(huì)發(fā)送一次消息。
運(yùn)行
先啟動(dòng)個(gè)RabbitMQ:
docker run \ -e RABBITMQ_DEFAULT_USER=pkslow \ -e RABBITMQ_DEFAULT_PASS=123456 \ -p 5672:5672 -p 15672:15672 \ rabbitmq:3.8-management
運(yùn)行程序后,會(huì)自己創(chuàng)建主題、發(fā)送信息、接收信息:

運(yùn)行日志如下:

可以看到每一次發(fā)/收大概是間隔了500ms,當(dāng)然不可能是精確的500ms。
以上就是Springcloud整合stream,rabbitmq實(shí)現(xiàn)消息驅(qū)動(dòng)功能的詳細(xì)內(nèi)容,更多關(guān)于Springcloud stream rabbitmq消息驅(qū)動(dòng)的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
spring如何集成cxf實(shí)現(xiàn)webservice接口功能詳解
這篇文章主要給大家介紹了關(guān)于spring如何集成cxf實(shí)現(xiàn)webservice接口功能的相關(guān)資料,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家 的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面來(lái)一起看看吧2018-07-07
繼承JpaRepository后,找不到findOne()方法的解決
這篇文章主要介紹了繼承JpaRepository后,找不到findOne()方法的解決方案,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2021-08-08
Java 利用binarySearch實(shí)現(xiàn)抽獎(jiǎng)計(jì)算邏輯
這篇文章主要介紹了Java 利用binarySearch實(shí)現(xiàn)抽獎(jiǎng)計(jì)算邏輯,幫助大家更好的理解和使用Java,感興趣的朋友可以了解下2020-12-12
基于Spring Boot不同的環(huán)境使用不同的配置方法
下面小編就為大家分享一篇基于Spring Boot不同的環(huán)境使用不同的配置方法,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2018-01-01

