关键词:
上篇博客讲解了服务器集群部署RocketMQ 博客地址:RocketMQ(2)---Docker部署RocketMQ集群
这篇在上篇搭建好的基础上,将SpringBoot整合RocketMQ实现生产消费。
GitHub地址
: https://github.com/yudiandemingzi/spring-boot-study
一、搭建步骤
先说下技术大致架构
SpringBoot2.1.6 + Maven3.5.4 + rocketmq4.3.0 + JDK1.8 +Lombok(插件)
1、添加rocketmq包
<!--注意: 这里的版本,要和部署在服务器上的版本号一致--> <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-client</artifactId> <version>4.3.0</version> </dependency>
2、JmsConfig(配置类)
连接RocketMQ服务器配置类,这里为了方便直接写成常量。
/** * @Description: 安装实际开发这里的信息 都是应该写在配置里,来读取,这里为了方便所以写成常量 */ public class JmsConfig { /** * Name Server 地址,因为是集群部署 所以有多个用 分号 隔开 */ public static final String NAME_SERVER = "127.12.15.6:9876;127.12.15.6:9877"; /** * 主题名称 主题一般是服务器设置好 而不能在代码里去新建topic( 如果没有创建好,生产者往该主题发送消息 会报找不到topic错误) */ public static final String TOPIC = "topic_family"; }
3、Producer (生产者)
@Slf4j @Component public class Producer { private String producerGroup = "test_producer"; private DefaultMQProducer producer; public Producer(){ //示例生产者 producer = new DefaultMQProducer(producerGroup); //不开启vip通道 开通口端口会减2 producer.setVipChannelEnabled(false); //绑定name server producer.setNamesrvAddr(JmsConfig.NAME_SERVER); start(); } /** * 对象在使用之前必须要调用一次,只能初始化一次 */ public void start(){ try { this.producer.start(); } catch (MQClientException e) { e.printStackTrace(); } } public DefaultMQProducer getProducer(){ return this.producer; } /** * 一般在应用上下文,使用上下文监听器,进行关闭 */ public void shutdown(){ this.producer.shutdown(); } }
4、Consumer (消费者)
@Slf4j @Component public class Consumer { /** * 消费者实体对象 */ private DefaultMQPushConsumer consumer; /** * 消费者组 */ public static final String CONSUMER_GROUP = "test_consumer"; /** * 通过构造函数 实例化对象 */ public Consumer() throws MQClientException { consumer = new DefaultMQPushConsumer(CONSUMER_GROUP); consumer.setNamesrvAddr(JmsConfig.NAME_SERVER); //消费模式:一个新的订阅组第一次启动从队列的最后位置开始消费 后续再启动接着上次消费的进度开始消费 consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET); //订阅主题和 标签( * 代表所有标签)下信息 consumer.subscribe(JmsConfig.TOPIC, "*"); // //注册消费的监听 并在此监听中消费信息,并返回消费的状态信息 consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> { // msgs中只收集同一个topic,同一个tag,并且key相同的message // 会把不同的消息分别放置到不同的队列中 try { for (Message msg : msgs) { //消费者获取消息 这里只输出 不做后面逻辑处理 String body = new String(msg.getBody(), "utf-8"); log.info("Consumer-获取消息-主题topic为={}, 消费消息为={}", msg.getTopic(), body); } } catch (UnsupportedEncodingException e) { e.printStackTrace(); return ConsumeConcurrentlyStatus.RECONSUME_LATER; } return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; }); consumer.start(); System.out.println("消费者 启动成功======="); } }
大致就是这边简单,下面就是测试。
二、测试
先写个测试接口进行测试。
1、Controller
@Slf4j @RestController public class Controller { @Autowired private Producer producer; private List<String> mesList; /** * 初始化消息 */ public Controller() { mesList = new ArrayList<>(); mesList.add("小小"); mesList.add("爸爸"); mesList.add("妈妈"); mesList.add("爷爷"); mesList.add("奶奶"); } @RequestMapping("/text/rocketmq") public Object callback() throws Exception { //总共发送五次消息 for (String s : mesList) { //创建生产信息 Message message = new Message(JmsConfig.TOPIC, "testtag", ("小小一家人的称谓:" + s).getBytes()); //发送 SendResult sendResult = producer.getProducer().send(message); log.info("输出生产者信息={}",sendResult); } return "成功"; } }
2、测试结果
很明显生产发送消息已经成功,二消费者也成功接收了消息!
另外我们再来看下RocketMQ控制台是否也有消费记录
很明显在控制台这边也会有消费记录!
总结
这边只是简单的整合,后面会通过RocketMQ实现分布式事务,可以用于线上实际环境中,到时候会深入讲解下源码。
转载于:https://www.cnblogs.com/qdhxhz/p/11109696.html
rocketmq(20)——整合springboot(代码片段)
整合SpringBootApacheRocketMQ提供了对SpringBoot的支持,目前最新的整合版本使用的是RocketMQ4.5.1版本,使用SpringBoot是2.0.5版本。整合SpringBoot需要引入rocketmq-spring-boot-starter依赖,目前最新版本是2.0.3。<dependency><groupId>org.apache.rock... 查看详情
rocketmq(20)——整合springboot(代码片段)
整合SpringBootApacheRocketMQ提供了对SpringBoot的支持,目前最新的整合版本使用的是RocketMQ4.5.1版本,使用SpringBoot是2.0.5版本。整合SpringBoot需要引入rocketmq-spring-boot-starter依赖,目前最新版本是2.0.3。<dependency><groupId>org.apache.ro 查看详情
rocketmq(二十四)整合springboot(代码片段)
1、官网RocketMQ为SpringBoot提供了整合方案,官网地址如下,上面提供了详细的整合步骤及案例。https://github.com/apache/rocketmq-spring官方详细文档(可以切换不同的版本)https://github.com/apache/rocketmq-spring/blob/release-2.0.1 查看详情
springboot整合rocketmq实现入门案例
...习了Spring整合RocketMQ的第一个案例!现在我们来学习SpringBoot如何整合RocketMQ实现更加简单的使用!文章目录1创建maven项目2配置文件3生产者4消费者5测试1创建maven项目创建一个maven项目。引入sp 查看详情
springboot整合rocketmq(代码片段)
项目环境jdk:1.8rocketmq:4.5.1springboot:2.6.3引入依赖<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId>& 查看详情
4springboot整合rocketmq实现消息发送和接收(代码片段)
我们使用主流的SpringBoot框架整合RocketMQ来讲解,使用方便快捷;最终项目结构如下:具体步骤如下:第一步:新建SpringBoot项目rocketmq-test,引入rocketmq依赖,以及项目配置<dependency> <groupId>org.apac... 查看详情
springboot2.x整合rocketmq4.x
开发生产者代码第一步:创建很普通的SpringBoot项目第二步:加入相关依赖<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>4.3.0</version></ 查看详情
springboot整合rocketmq的各种消息类型,生产者,消费者(代码片段)
文章目录Springboot整合使用pom依赖yml整合使用同步消息消费者异步消息消费者单向消息消费者延时消息消费者顺序消息消费者sql92过滤消息消费者事物消息消费者@RocketMQMessageListener参数解释我的rocketmq各种集群方案安装Springboot... 查看详情
[原创]springboot整合rocketmq消息队列
参考技术A什么情况下的异步操作需要使用消息队列而不是多线程?StartNameServerStartBrokerWindows下安装RocketMq:https://www.jianshu.com/p/4a275e779afaRocketMq名词解释:https://my.oschina.net/javamaster/blog/2051703解释Push与Pull区别:https://www.jianshu.com/p/f0... 查看详情
11springboot整合rocketmq实现事务消息(代码片段)
事务消息是RocketMQ提供的非常重要的一个特性,在4.x版本之后开源,可以利用事务消息轻松地实现分布式事务。RocketMQ在其消息定义的基础上,对事务消息扩展了两个相关的概念:Half(Prepare)Message——半消息(预处理... 查看详情
springboot实战项目整合阿里云rocketmq消息队列实现发送普通消息,延时消息(代码片段)
原文地址:Springboot实战项目整合阿里云RocketMQ消息队列实现发送普通消息,延时消息--附代码-学不会丶-博客园一.为什么选择RocketMQ消息队列?(可跳过看三的整合代码实例)首先RocketMQ是阿里巴巴自研出来的... 查看详情
9springboot整合rocketmq实现顺序消息(代码片段)
rocketmq默认发送的消息是进入多个消息队列,然后消费端多线程并发消费,所以默认情况,不是顺序消费消息的;有时候,我们需要实现顺序消费一批消息,比如电商系统,订单创建,支付,完... 查看详情
八.rocketmq极简入门-在springboot中使用rocketmq(代码片段)
前言现在开发项目都是基于SpringBoot,新项目很少使用Spring,所以我们学习一门技术除了要会原生API,还不得不考虑和SpringBoot集成,本篇文章为SpirngBoot整合RocketMQ案例SpringBoot集成RocketMQ导入依赖这里使用整合RocketMQ... 查看详情
6springboot整合rocketmq发送异步消息(代码片段)
发送异步消息是指producer向broker发送消息时指定消息发送成功及发送异常的回调方法,调用API后立即返回,producer发送消息线程不阻塞,消息发送成功或失败的回调任务在一个新的线程中执行。相对发送同步消息,... 查看详情
7springboot整合rocketmq发送单向消息(代码片段)
发送单向消息是指producer向broker发送消息,执行API时直接返回,不等待broker服务器的结果。这种方式主要用在不特别关心发送结果的场景,举例:日志发送;RocketMQTemplate给我们提供了sendOneWay方法(有多个重载)... 查看详情
10springboot整合rocketmq实现延迟消息(代码片段)
延迟消息对于消息中间件来说,producer将消息发送到mq的服务器上,但并不希望这条消息马上被消费,而是推迟到当前时间节点之后的某个时间点,再将消息投递到queue中让consumer进行消费。延迟消息的使用场景... 查看详情
5springboot整合rocketmq发送同步消息(代码片段)
发送同步消息是指producer向broker发送消息,执行API时同步等待,直到broker服务器返回发送结果;相对异步发送消息,同步会阻塞线程,性能相对差点,但是可靠性高,这种方式得到广泛使用,比如... 查看详情
12springboot整合rocketmq实现过滤消息-根据tag方式过滤消息(代码片段)
消息发送端只能设置一个tag,消息接收端可以设置多个tag。接收消息端通过‘||’设置多个tag,如下:tag1||tag2||tag3||…上实例,生产端发送三个消息,TAG分别是TAG1,TAG2,TAG3/***发送带Tag消息,测试根据Tag过滤消息*/... 查看详情