Springboot 整合 RocketMQ 收發消息的配置過程
Springboot 整合 RocketMQ 收發消息
創建springboot項目
pom.xml添加rocketmq-spring-boot-starter
依賴。
<dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-spring-boot-starter</artifactId> <version>2.1.0</version> </dependency>
yml 配置
application.yml
rocketmq: name-server: 192.168.64.141:9876
application-demo1.yml
使用 demo1 profile 指定生產者組組名
rocketmq: producer: group: producer-demo1
application-demo2.yml
使用 demo2 profile 指定生產者組組名
rocketmq: producer: group: producer-demo2
測試
demo 1
- 發送普通消息
- 發送 Spring 的通用 Message 對象
- 發送異步消息
- 發送順序消息
生產者
package cn.tedu.demo2.m1; import org.apache.rocketmq.client.producer.SendCallback; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.messaging.Message; import org.springframework.messaging.support.MessageBuilder; import org.springframework.stereotype.Component; @Component public class Producer { @Autowired private RocketMQTemplate t ; public void send(){ //發送同步消息 t.convertAndSend("Topic1:TagA", "Hello world! "); //發送spring的Message Message<String> message = MessageBuilder.withPayload("Hello Spring message! ").build(); t.send("Topic1:TagA",message); //發送異步消息 t.asyncSend("Topic1:TagA", "hello world asyn", new SendCallback() { @Override public void onSuccess(SendResult sendResult) { System.out.println("發送成功"); } @Override public void onException(Throwable throwable) { System.out.println("發送失敗"); } }); //發送順序消息 t.syncSendOrderly("Topic1", "98456237,創建", "98456237"); t.syncSendOrderly("Topic1", "98456237,支付", "98456237"); t.syncSendOrderly("Topic1", "98456237,完成", "98456237"); } }
消費者
package cn.tedu.demo2.m1; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.springframework.stereotype.Component; @Component @RocketMQMessageListener(topic = "Topic1",consumerGroup = "consumer-demo1") public class Consumer implements RocketMQListener<String> { @Override public void onMessage(String s) { System.out.println("收到"+s); } }
主類
package cn.tedu.demo2.m1; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @SpringBootApplication public class Main { public static void main(String[] args) { SpringApplication.run(Main.class, args); } }
測試類
需要放在 test 文件夾
激活 demo1 profile @ActiveProfiles("demo1")
package cn.tedu.demo2.m1; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.ActiveProfiles; @SpringBootTest @ActiveProfiles("demo1") public class Test1 { @Autowired private Producer producer; @Test public void test1(){ producer.send(); try { Thread.sleep(5000); } catch (InterruptedException e) { e.printStackTrace(); } } }
demo 2
發送事務消息
生產者
package cn.tedu.demo2.m2; import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener; import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener; import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState; import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.messaging.Message; import org.springframework.messaging.support.MessageBuilder; import org.springframework.stereotype.Component; @Component public class Producer { @Autowired private RocketMQTemplate t; public void send(){ Message<String> message = MessageBuilder.withPayload("Hello world").build(); //一旦發送消息,則執行監聽器 t.sendMessageInTransaction("Topic2",message,null); } @RocketMQTransactionListener class Lis implements RocketMQLocalTransactionListener { @Override public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) { System.out.println("執行本地事務"); return RocketMQLocalTransactionState.UNKNOWN; } @Override public RocketMQLocalTransactionState checkLocalTransaction(Message message) { System.out.println("執行事務回查"); return RocketMQLocalTransactionState.COMMIT; } } }
消費者
package cn.tedu.demo2.m2; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.springframework.stereotype.Component; @Component @RocketMQMessageListener(topic = "Topic2",consumerGroup = "consumer-demo2") public class Consumer implements RocketMQListener<String> { @Override public void onMessage(String s) { System.out.println("收到"+s); } }
主類
package cn.tedu.demo2.m2; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @SpringBootApplication public class Main { public static void main(String[] args) { SpringApplication.run(Main.class, args); } }
測試類
package cn.tedu.demo2.m2; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.ActiveProfiles; @SpringBootTest @ActiveProfiles("demo2") public class Test2 { @Autowired private Producer producer; @Test public void test1(){ producer.send(); //為瞭能夠收到消費者消費的數據,這裡通過休眠模擬等待時間 try { Thread.sleep(30000); } catch (InterruptedException e) { e.printStackTrace(); } } }
到此這篇關於Springboot 整合 RocketMQ 收發消息的文章就介紹到這瞭,更多相關Springboot 整合 RocketMQ 收發消息內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet!
推薦閱讀:
- Springboot詳解RocketMQ實現廣播消息流程
- SpringBoot整合RocketMQ實現消息發送和接收的詳細步驟
- RocketMQTemplate 註入失敗的解決
- springboot整合rocketmq實現分佈式事務
- Spring框架學習之Spring @Autowired實現自動裝配的代碼