Spring Cloud Stream整合RocketMQ的搭建方法
更新時間：2024年11月25日 10:23:19 作者：別團等shy哥發(fā)育
本文介紹了如何使用SpringCloudStream整合RocketMQ進行消息傳遞,SpringCloudStream是一個用于構(gòu)建與共享消息系統(tǒng)連接的框架,支持持久pub/sub語義和消費者組,感興趣的朋友跟隨小編一起看看吧
Spring Cloud Stream整合RocketMQ
這里書接上回，默認你已經(jīng)搭建好了RocketMQ主從異步集群，前面文章已經(jīng)介紹過搭建方法。
1、Spring Cloud Stream介紹
Spring Cloud Stream是一個框架，用于構(gòu)建與共享消息系統(tǒng)連接的高度可擴展的事件驅(qū)動微服務(wù)。
官網(wǎng)：https://spring.io/projects/spring-cloud-stream
該框架提供了一個靈活的編程模型，該模型基于已經(jīng)建立和熟悉的Spring習(xí)慣用法和最佳實踐，包括對持久pub/sub語義、消費者組和有狀態(tài)分區(qū)的支持。
Spring Cloud Stream的核心構(gòu)建塊是：
Destination Binders:負責(zé)提供與外部消息傳遞系統(tǒng)集成的組件。Destination Bindings:外部消息系統(tǒng)和最終用戶提供的應(yīng)用程序代碼(生產(chǎn)者/消費者)之間的橋梁。Message:生產(chǎn)者和消費者用來與目標(biāo)綁定器(以及通過外部消息系統(tǒng)的其他應(yīng)用程序)進行通信的規(guī)范數(shù)據(jù)結(jié)構(gòu)。
2、生產(chǎn)者
2.1 引入依賴
<dependencies>
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-alibaba-dependencies</artifactId>
<version>${spring-cloud-alibaba.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
<version>2.2.2.RELEASE</version>
<exclusions>
<exclusion>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
</exclusion>
<exclusion>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-acl</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.7.1</version>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-acl</artifactId>
<version>4.7.1</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>注意，RocketMQ官方維護的Spring-Cloud-Stream依賴中的rocketmq版本為4.4，需要排出后加入4.7.1的依。
2.2 編寫配置文件
spring:
application:
name: my-spring-cloud-rocketmq-producer
cloud:
stream:
bindings:
output:
destination: TopicTest
rocketmq:
binder:
name-server: 192.168.159.34:9876
server:
port: 80802.3 啟動類打上注解
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
@EnableBinding(Source.class)
@SpringBootApplication
public class MySpringCloudRocketmqProducerApplication {
public static void main(String[] args) {
SpringApplication.run(MySpringCloudRocketmqProducerApplication.class, args);
}
}其中，@EnableBinding(Source.class)指向配置文件的output參數(shù)。
2.4 編寫生產(chǎn)者程序
import org.apache.rocketmq.common.message.MessageConst;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
@Component
public class MyProducer {
@Resource
private Source source;
public void sendMessage(String msg){
//封裝消息頭
Map<String,Object> headers=new HashMap<>();
headers.put(MessageConst.PROPERTY_TAGS,"TagA");
MessageHeaders messageHeaders=new MessageHeaders(headers);
//創(chuàng)建消息對象
Message<String> message = MessageBuilder.createMessage(msg, messageHeaders);
//發(fā)送消息
source.output().send(message);
}
}2.5 編寫單元測試發(fā)送消息
@SpringBootTest
class MySpringCloudRocketmqProducerApplicationTests {
@Autowired
private MyProducer producer;
@Test
void contextLoads() {
producer.sendMessage("hello,spring cloud stream message");
}
}3、消費者
3.1 引入依賴
與生產(chǎn)者相同。
3.2 編寫配置文件
spring:
application:
name: my-spring-cloud-rocketmq-consumer
cloud:
stream:
bindings:
# input消費者
input:
destination: TopicTest
group: spring-cloud-stream-consumer-group
# 配置RocketMQ
rocketmq:
binder:
name-server: 192.168.159.34:9876
server:
port: 80813.3 啟動類打上注解
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Sink;
@EnableBinding(Sink.class)
@SpringBootApplication
public class MySpringCloudRocketmqConsumerApplication {
public static void main(String[] args) {
SpringApplication.run(MySpringCloudRocketmqConsumerApplication.class, args);
}
}其中@EnableBinding(Sink.class)指向配置文件的input參數(shù)。
3.4 編寫消費者程序
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.stereotype.Component;
@Component
public class MyConsumer {
@StreamListener(Sink.INPUT)
public void processMessage(String message){
System.out.println("收到的消息："+message);
}
}先啟動消費者，使用單元測試發(fā)送消息。
到此這篇關(guān)于Spring Cloud Stream整合RocketMQ的文章就介紹到這了,更多相關(guān)Spring Cloud Stream整合RocketMQ內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
您可能感興趣的文章:
- SpringCloudStream原理和深入使用小結(jié)
- SpringCloud中的Stream服務(wù)間消息傳遞詳解
- 使用Spring?Cloud?Stream處理事件的示例詳解
- spring-cloud-stream的手動消息確認問題
- SpringCloudStream中的消息分區(qū)數(shù)詳解
- 關(guān)于SpringCloudStream配置問題
- Spring Cloud Stream 高級特性使用詳解
- SpringCloud微服務(wù)開發(fā)基于RocketMQ實現(xiàn)分布式事務(wù)管理詳解
- SpringCloud+RocketMQ實現(xiàn)分布式事務(wù)的實踐
相關(guān)文章
本文主要介紹了java無鎖hashmap原理與實現(xiàn)，大家參考使用吧2014-01-01
Java數(shù)據(jù)結(jié)構(gòu)和算法之鏈表詳解
鏈表是一種物理存儲單元上非連續(xù)、非順序的存儲結(jié)構(gòu),java代碼實現(xiàn)單鏈表,插入,刪除和遍歷等功能,這篇文章主要給大家介紹了關(guān)于Java數(shù)據(jù)結(jié)構(gòu)和算法之鏈表的相關(guān)資料,需要的朋友可以參考下2024-01-01
SpringBoot+MyBatis-Plus實現(xiàn)數(shù)據(jù)庫讀寫分離的代碼示例
在當(dāng)今互聯(lián)網(wǎng)應(yīng)用中,數(shù)據(jù)庫讀寫分離是提高系統(tǒng)性能和穩(wěn)定性的重要手段之一,通過將讀操作和寫操作分別路由到不同的數(shù)據(jù)庫節(jié)點,可以有效減輕數(shù)據(jù)庫服務(wù)器的負擔(dān),本文將介紹如何利用SpringBoot和MyBatis-Plus框架實現(xiàn)數(shù)據(jù)庫讀寫分離,需要的朋友可以參考下2023-11-11
這篇文章主要介紹了Java?LocalDateTime實用方法,Java8提供了新的時間接口LocalDateTime，本文通過實例代碼給大家介紹的非常詳細，需要的朋友可以參考下2022-01-01
Springboot集成SpringState狀態(tài)機的實現(xiàn)
本文主要介紹了Springboot集成SpringState狀態(tài)機的實現(xiàn),文中通過示例代碼介紹的非常詳細,對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2025-06-06
這篇文章主要介紹了mybatis 一對一、一對多和多對多查詢的實例代碼，非常不錯，具有一定的參考借鑒價值,需要的朋友可以參考下2018-06-06
這篇文章主要為大家詳細介紹了java實現(xiàn)打印正三角的方法，具有一定的參考價值，感興趣的小伙伴們可以參考一下2018-01-01
java8 List<Object>去掉重復(fù)對象的幾種方法
本文主要介紹了java8 List<Object>去掉重復(fù)對象的幾種方法，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2022-04-04
這篇文章主要介紹了將對象轉(zhuǎn)化為字符串的java實例，有需要的朋友可以參考一下2013-12-12
Java語言作為第一種支持國際化的語言，在Internet從一開始就具有其他語言無與倫比的國際化的本質(zhì)特性，查看Java所支持的語言及相應(yīng)的版本信息可以采用以下代碼進行查詢2014-01-01