SpringBoot集成RocketMQ实现消息发送的三种方式
RocketMQ 支持3 种消息发送方式: 同步 (sync)、异步(async)、单向(oneway)。
- 同步:发送者向 MQ 执行发送消息API 时,同步等待,直到消息服务器返回发送结果。
- 异步:发送者向MQ 执行发送消息API 时,指定消息发送成功后的回调函数,然后调用消息发送API 后,立即返回,消息发送者线程不阻塞,直到运行结束,消息发送成功或失败的回调任务在一个新的线程中返回。
- 单向:消息发送者向MQ 执行发送消息API 时,直接返回,不等待消息服务器的结果,也不注册回调函数,只管发,不管是否成功存储在消息服务器上。
前提:
运行项目需要具备RocketMQ环境
一、pom文件引入依赖
<dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-spring-boot-starter</artifactId> <version>2.2.2</version> </dependency>
二、application.yml文件添加内容
rocketmq: name-server: IP:9876 #IP为rocketmq访问的地址 producer: group: first1-group #事务消息才会用到
三、创建producer生产者
package com.tlxy.lhn.controller.rocketmq; 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.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestMethod; import org.springframework.web.bind.annotation.RestController; @RestController public class RocketController { @Autowired private RocketMQTemplate rocketMQTemplate; @RequestMapping(value = "/rocket", method = RequestMethod.GET) public void noTag() { // convertAndSend() 发送普通字符串消息 rocketMQTemplate.convertAndSend("sendMessage_topic", "Hello Word"); } @RequestMapping(value = "/tagA", method = RequestMethod.GET) public void tagA() { rocketMQTemplate.convertAndSend("sendMessage_topic:tagA", "hello world tagA"); } @RequestMapping(value = "/tagB", method = RequestMethod.GET) public void tagB() { rocketMQTemplate.convertAndSend("sendMessage_topic:tagB", "hello world tagB"); } @RequestMapping(value = "/syncSend", method = RequestMethod.GET) public void syncSend() { String json = "发送同步消息"; SendResult sendResult = rocketMQTemplate.syncSend("sendMessage_topic:1", json); System.out.println(sendResult); } @RequestMapping(value = "/aSyncSend", method = RequestMethod.GET) public void aSyncSend() { String json = "发送异步消息"; SendCallback callback = new SendCallback() { @Override public void onSuccess(SendResult sendResult) { System.out.println("发送消息成功"); } @Override public void onException(Throwable throwable) { System.out.println("发送消息失败"); } }; rocketMQTemplate.asyncSend("sendMessage_topic", json, callback); } @RequestMapping(value = "/sendOneWay", method = RequestMethod.GET) public void sendOneWay() { rocketMQTemplate.sendOneWay("sendMessage_topic", "发送单向消息"); } }
四、创建Consumer消费者(创建两个消费者,所属一个Topic)
Consumer1:
package com.tlxy.lhn.controller.rocketmq; import org.apache.rocketmq.spring.annotation.MessageModel; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.springframework.stereotype.Component; /** * topic 是主题 * consumerGroup 是消费者组,一条消息只能被同一个消费者组里的一个消费者消费。 * selectorExpression 是用于消息过滤的,以 TAG 方式为例: * 默认为 "*",表示不过滤,消费此 topic 下所有消息 * 配置为 "tagA",表示只消费此 topic 下 TAG = tagA 的消息 * 配置为 "tagA || tagB",表示消费此 topic 下 TAG = tagA 或 TAG = tagB 的消息,以此类推 * 消费模式:默认 CLUSTERING ( CLUSTERING:负载均衡 )( BROADCASTING:广播机制 ) */ @RocketMQMessageListener(topic = "sendMessage_topic", consumerGroup = "consumer-group-test1", // selectorExpression = "tagA || tagB", messageModel = MessageModel.CLUSTERING) @Component public class DemoConsumer1 implements RocketMQListener<String> { @Override public void onMessage(String s) { System.out.println("receive message1:" + s); try { Thread.sleep(5000); } catch (InterruptedException e) { throw new RuntimeException(e); } System.out.println("处理完成"); } }
Consumer2:
package com.tlxy.lhn.controller.rocketmq; import org.apache.rocketmq.spring.annotation.MessageModel; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.springframework.stereotype.Component; /** * topic 是主题 * consumerGroup 是消费者组,一条消息只能被同一个消费者组里的一个消费者消费。 * selectorExpression 是用于消息过滤的,以 TAG 方式为例: * 默认为 "*",表示不过滤,消费此 topic 下所有消息 * 配置为 "tagA",表示只消费此 topic 下 TAG = tagA 的消息 * 配置为 "tagA || tagB",表示消费此 topic 下 TAG = tagA 或 TAG = tagB 的消息,以此类推 * 消费模式:默认 CLUSTERING ( CLUSTERING:负载均衡 )( BROADCASTING:广播机制 ) */ @RocketMQMessageListener(topic = "sendMessage_topic", consumerGroup = "consumer-group-test1", messageModel = MessageModel.CLUSTERING) @Component public class DemoConsumer2 implements RocketMQListener<String> { @Override public void onMessage(String s) { System.out.println("receive message2:" + s); try { Thread.sleep(8000); } catch (InterruptedException e) { throw new RuntimeException(e); } System.out.println("处理完成"); } }
五、启动项目测试
1、发送同步消息:
以上输出可以看到:同步消息发送后,消息发送到broker后就返回结果了,消费端还未处理完,两者互互不影响。
2、发送异步消息:
以上输出:发送者向MQ 执行发送消息API 时,指定消息发送成功后的回调函数,然后调用消息发送API 后,立即返回,消息发送者线程不阻塞,直到运行结束,消息发送成功或失败的回调任务在一个新的线程中返回。
3、发送单向消息:
以上输出:消息发送者向MQ 执行发送消息API 时,直接返回,不等待消息服务器的结果,也不注册回调函数,只管发,不管是否成功存储在消息服务器上。
以上就是SpringBoot集成RocketMQ实现消息发送的三种方式的详细内容,更多关于SpringBoot RocketMQ消息发送的资料请关注脚本之家其它相关文章!
相关文章
Mybatis一级缓存和结合Spring Framework后失效的源码探究
这篇文章主要介绍了Mybatis一级缓存和结合Spring Framework后失效的源码探究,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下2021-04-04使用MockMvc进行controller层单元测试 事务自动回滚的完整案例
这篇文章主要介绍了使用MockMvc进行controller层单元测试 事务自动回滚的完整案例,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教2021-06-06springboot3.2整合mybatis-plus详细代码示例
这篇文章主要给大家介绍了关于springboot3.2整合mybatis-plus的相关资料,Spring Boot是一个非常流行的Java Web框架,可以快速地搭建Web应用程序,需要的朋友可以参考下2023-12-12Java常用正则表达式验证类完整实例【邮箱、URL、IP、电话、身份证等】
这篇文章主要介绍了Java常用正则表达式验证类,结合完整实例形式分析了Java针对邮箱、网址URL、IP地址、电话、身份证等正则验证相关操作技巧,需要的朋友可以参考下2018-12-12基于Java+SpringBoot+Vue前后端分离实现仓库管理系统
这篇文章主要介绍了一个完整的仓库管理系统是基于Java+Springboot + Vue前后端分离编写的,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下2023-06-06浅谈Java自定义类加载器及JVM自带的类加载器之间的交互关系
这篇文章主要介绍了浅谈Java自定义类加载器及JVM自带的类加载器之间的交互关系,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧2021-02-02
最新评论