一、简介
rocketmq 是一款开源的分布式消息中间件,由阿里巴巴开源。由阿里巴巴集团开发并开源,目前被捐赠给apache基金会,并入选孵化器项目,2017年从apache基金会毕业后,rocketmq被指定为顶级项目(tlp)。它具有高可用性、高性能、低延迟等特点,广泛应用于阿里巴巴集团内部以及众多外部企业的业务系统中。
1.1、rocketmq 主要特点
- 分布式架构:rocketmq 是基于分布式架构设计的,支持水平扩展,可以灵活地部署和扩展。
- 高可用性:rocketmq 支持主从架构,消息存储采用主从复制的方式,保证了消息的高可用性和可靠性。
- 高性能:rocketmq 在消息存储、消息传输等方面进行了优化,具有较高的吞吐量和较低的延迟。
- 丰富的特性:rocketmq 提供了丰富的特性,包括顺序消息、延迟消息、事务消息、消息过滤、消息轨迹等,满足了不同场景下的需求。
- 监控和管理:rocketmq 提供了丰富的监控和管理功能,可以通过控制台或者监控工具实时监控消息的状态和性能指标。
- 开源社区支持:rocketmq 是一款开源的消息中间件,拥有活跃的开源社区,提供了丰富的文档和示例,为用户提供了便利的支持和帮助。
1.2、rocketmq 核心组件
rocketmq 的架构主要包括以下核心组件:
- nameserver:nameserver 是 rocketmq 的路由管理组件,负责管理 broker 的路由信息。客户端在发送消息或者消费消息时,需要先从 nameserver 获取相应的 broker 地址,然后再与 broker 建立连接。
- broker:broker 是 rocketmq 的消息存储和传输组件,负责存储消息以及向消费者传递消息。一个 rocketmq 系统可以包含多个 broker,每个 broker 负责存储一部分消息数据,并提供相应的消息服务。
- producer:producer 是消息的生产者,负责产生消息并将消息发送到 broker 中。producer 将消息发送到指定的 topic,然后由 broker 存储并传递给相应的消费者。
- consumer:consumer 是消息的消费者,负责订阅并消费 broker 中的消息。consumer 通过订阅指定的 topic 来接收消息,并进行相应的业务处理。
- topic:topic 是消息的主题,用于对消息进行分类和管理。producer 将消息发送到指定的 topic,而 consumer 则订阅相应的 topic 来接收消息。
- message queue:message queue 是消息队列,用于存储消息。每个 topic 可以分为多个 message queue,每个 message queue 保存了一部分消息,多个 message queue 组成了一个 topic 的完整消息存储。
总的来说,rocketmq是阿里推出的优秀开源分布式消息中间件,具有高性能、高可靠、高并发等优点,是构建分布式系统不可或缺的基础组件之一。
1.3、概念
同步发送指的是生产者在发送消息后,会阻塞当前线程,直到收到broker的发送响应后才返回,响应中包含消息是否发送成功的状态。同步发送的优缺点如下:
优点:
- 发送可靠性高,能够立即得知发送结果
- 发送反馈直接返回,不需要通过专门的监听通道接收结果
缺点:
- 当broker发生故障或者网络延迟时,生产者线程会被阻塞,影响发送效率
- 对于发送端吞吐量较高的场景,同步发送存在性能瓶颈
1.4、场景
同步发送对消息可靠性传输有较高要求的场景,如通知消息等,发送端对发送吞吐量要求不是特别高的场景
二、父工程
因为这个系统会有很多rocketmq的知识,我准备拆开写,避免频繁的导入依赖和包,我这里采用分模块开发。
2.1、父工程依赖
pom.xml
<parent> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-starter-parent</artifactid> <version>2.6.0</version> <relativepath/> <!-- lookup parent from repository --> </parent> <properties> <maven.compiler.source>8</maven.compiler.source> <maven.compiler.target>8</maven.compiler.target> </properties> <dependencies> <dependency> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-starter-web</artifactid> </dependency> <dependency> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-starter-test</artifactid> <scope>test</scope> </dependency> <dependency> <groupid>org.apache.rocketmq</groupid> <artifactid>rocketmq-spring-boot-starter</artifactid> <version>2.2.3</version> </dependency> <dependency> <groupid>org.apache.commons</groupid> <artifactid>commons-lang3</artifactid> <version>3.12.0</version> </dependency> <dependency> <groupid>org.projectlombok</groupid> <artifactid>lombok</artifactid> <version>1.18.26</version> </dependency> </dependencies> <dependencymanagement> <dependencies> <dependency> <groupid>com.alian</groupid> <artifactid>common-rocketmq-dto</artifactid> <version>1.0.0-snapshot</version> </dependency> </dependencies> </dependencymanagement>
2.2、公共模块
这里需要注意的是下面这个依赖,其实就是一个会员类。如果你传输的不是对象也可以不要,我这里演示就创建了。
<dependency> <groupid>com.alian</groupid> <artifactid>common-rocketmq-dto</artifactid> <version>1.0.0-snapshot</version> </dependency>
member.java
package com.alian.common; import lombok.data; import java.util.date; @data public class member { private long id; private string membername; private int age; private date birthday; }
后续的项目都会用到这个父工程和公共模块,后面就不再过多说明了。
三、生产者
我们在父工程下新建一个模块用于发送同步消息。
3.1 maven依赖
pom.xml
<?xml version="1.0" encoding="utf-8"?> <project xmlns="http://maven.apache.org/pom/4.0.0" xmlns:xsi="http://www.w3.org/2001/xmlschema-instance" xsi:schemalocation="http://maven.apache.org/pom/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <parent> <artifactid>rocketmq</artifactid> <groupid>com.alian</groupid> <version>1.0.0-snapshot</version> </parent> <modelversion>4.0.0</modelversion> <artifactid>01-send-sync-message</artifactid> <properties> <maven.compiler.source>8</maven.compiler.source> <maven.compiler.target>8</maven.compiler.target> </properties> <dependencies> <dependency> <groupid>com.alian</groupid> <artifactid>common-rocketmq-dto</artifactid> <version>1.0.0-snapshot</version> </dependency> </dependencies> </project>
3.2 application配置
application.properties
server.port=8001 # rocketmq地址 rocketmq.name-server=192.168.0.234:9876 # 默认的生产者组 rocketmq.producer.group=sync_group # 发送同步消息超时时间 rocketmq.producer.send-message-timeout=3000 # 用于设置在消息发送失败后,生产者是否尝试切换到下一个服务器。设置为 true 表示启用,在发送失败时尝试切换到下一个服务器 rocketmq.producer.retry-next-server=true # 用于指定消息发送失败时的重试次数 rocketmq.producer.retry-times-when-send-failed=3 # 设置消息压缩的阈值,为0表示禁用消息体的压缩 rocketmq.producer.compress-message-body-threshold=0
3.3 发送字符串消息
消息的发送比较简单,我们直接引用
@autowired private rocketmqtemplate rocketmqtemplate;
此对象封装了一系列的消息发送方法。
sendstrmessagetest.java
package com.alian.sync; import lombok.extern.slf4j.slf4j; import org.apache.rocketmq.client.producer.sendresult; import org.apache.rocketmq.spring.core.rocketmqtemplate; import org.junit.jupiter.api.aftereach; import org.junit.jupiter.api.test; import org.springframework.beans.factory.annotation.autowired; import org.springframework.boot.test.context.springboottest; import org.springframework.messaging.message; import org.springframework.messaging.messageheaders; import org.springframework.messaging.support.messagebuilder; @slf4j @springboottest public class sendstrmessagetest { @autowired private rocketmqtemplate rocketmqtemplate; @test public void syncsendstringmessage() { string topic = "string_message_topic"; string message = "我是一条同步文本消息:syncsendstringmessage"; sendresult sendresult = rocketmqtemplate.syncsend(topic, message); log.info("同步发送返回的结果:{}", sendresult); } @test public void syncsendstringmessagewithbuilder() { string topic = "string_message_topic"; string message = "我是一条同步的文本消息:syncsendstringmessagewithbuilder"; message<string> msg = messagebuilder.withpayload(message) // 消息类型 .setheader(messageheaders.content_type, "text/plain") .build(); sendresult sendresult = rocketmqtemplate.syncsend(topic, msg); log.info("同步发送返回的结果:{}", sendresult); } @test public void syncsendstringmessagewithbuildertimeout() { string topic = "string_message_topic"; string message = "我是一条同步的文本消息:syncsendstringmessagewithbuildertimeout"; message<string> msg = messagebuilder.withpayload(message) // 消息类型 .setheader(messageheaders.content_type, "text/plain") .build(); // 3秒发送超时 sendresult sendresult = rocketmqtemplate.syncsend(topic, msg, 3000); log.info("同步发送返回的结果:{}", sendresult); } @aftereach public void waiting() { try { thread.sleep(3000l); } catch (interruptedexception e) { e.printstacktrace(); } } }
这里需要注意的是,我发送字符串消息的时候,topic都是 string_message_topic,因为我这里是本地开发环境,并且我在配置文件中配置了
autocreatetopicenable=true
当该参数设置为true时,如果生产者在发送消息时使用了一个在broker端不存在的topic,则broker会自动创建该topic,允许消息正常发送和存储。
当该参数设置为false时,如果生产者使用了不存在的topic,则broker会直接拒绝发送请求,不会自动创建topic。
官方对该参数的解释是:自动创建topic的特性主要是为了方便,但也可能带来一些风险,比如有的应用程序由于编码上的低级错误导致无意中创建了大量的topic。因此,生产环境建议将该参数设置为false,只有手工创建所需的topic。
3.4 发送json消息
sendjsonmessagetest.java
package com.alian.sync; import com.alibaba.fastjson.jsonobject; import lombok.extern.slf4j.slf4j; import org.apache.rocketmq.client.producer.sendresult; import org.apache.rocketmq.spring.core.rocketmqtemplate; import org.junit.jupiter.api.aftereach; import org.junit.jupiter.api.test; import org.springframework.beans.factory.annotation.autowired; import org.springframework.boot.test.context.springboottest; import org.springframework.messaging.message; import org.springframework.messaging.messageheaders; import org.springframework.messaging.support.messagebuilder; import java.util.hashmap; import java.util.map; @slf4j @springboottest public class sendjsonmessagetest { @autowired private rocketmqtemplate rocketmqtemplate; @test public void sendjsonmessage() { string topic = "json_message_topic"; jsonobject json = new jsonobject(); json.put("name", "alian"); json.put("age", "28"); json.put("hobby", "java"); sendresult sendresult = rocketmqtemplate.syncsend(topic, json); log.info("同步发送返回的结果:{}", sendresult); } @test public void sendjsonmessagewithbuilder() { string topic = "json_message_topic"; jsonobject json = new jsonobject(); json.put("name", "alian"); json.put("age", "28"); json.put("hobby", "java"); message<jsonobject> msg = messagebuilder.withpayload(json) // 消息类型 .setheader(messageheaders.content_type, "application/json") .build(); sendresult sendresult = rocketmqtemplate.syncsend(topic, msg); log.info("同步发送返回的结果:{}", sendresult); } @test public void sendmapmessage() { string topic = "json_message_topic"; map<string, string> map = new hashmap<>(); map.put("1", "java"); map.put("2", "go"); map.put("3", "c"); map.put("4", "vue"); map.put("5", "react"); sendresult sendresult = rocketmqtemplate.syncsend(topic, map); log.info("同步发送返回的结果:{}", sendresult); } @aftereach public void waiting() { try { thread.sleep(3000l); } catch (interruptedexception e) { e.printstacktrace(); } } }
这里需要注意的是,我发送json消息的时候,topic都是 json_message_topic,这里map消息也能被jsonobject消费。
3.5 发送java对象消息
sendjavaobjectmessagetest.java
package com.alian.sync; import com.alian.common.member; import com.alibaba.fastjson.jsonobject; import lombok.extern.slf4j.slf4j; import org.apache.rocketmq.client.producer.sendresult; import org.apache.rocketmq.spring.core.rocketmqtemplate; import org.junit.jupiter.api.aftereach; import org.junit.jupiter.api.test; import org.springframework.beans.factory.annotation.autowired; import org.springframework.boot.test.context.springboottest; import org.springframework.messaging.message; import org.springframework.messaging.messageheaders; import org.springframework.messaging.support.messagebuilder; import java.util.date; import java.util.hashmap; import java.util.map; @slf4j @springboottest public class sendjavaobjectmessagetest { @autowired private rocketmqtemplate rocketmqtemplate; @test public void sendjavaobjectmessage() { string topic = "java_object_message_topic"; member member = new member(); member.setid(10086l); member.setmembername("alian"); member.setage(28); member.setbirthday(new date()); sendresult sendresult = rocketmqtemplate.syncsend(topic, member); log.info("同步发送返回的结果:{}", sendresult); } @test public void sendjavaobjectmessagewithbuilder() { string topic = "java_object_message_topic"; member member = new member(); member.setid(10086l); member.setmembername("alian"); member.setage(28); member.setbirthday(new date()); message<member> msg = messagebuilder.withpayload(member) // 设置消息类型 .setheader(messageheaders.content_type, "application/json") .build(); sendresult sendresult = rocketmqtemplate.syncsend(topic, msg); log.info("同步发送返回的结果:{}", sendresult); } @aftereach public void waiting() { try { thread.sleep(3000l); } catch (interruptedexception e) { e.printstacktrace(); } } }
这里需要注意的是,我发送java对象消息的时候,topic都是 java_object_message_topic
四、消费者
我们在父工程下新建一个模块用于发送同步消息。
4.1 maven依赖
pom.xml
<?xml version="1.0" encoding="utf-8"?> <project xmlns="http://maven.apache.org/pom/4.0.0" xmlns:xsi="http://www.w3.org/2001/xmlschema-instance" xsi:schemalocation="http://maven.apache.org/pom/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <parent> <artifactid>rocketmq</artifactid> <groupid>com.alian</groupid> <version>1.0.0-snapshot</version> </parent> <modelversion>4.0.0</modelversion> <artifactid>08-comsume-concurrent</artifactid> <properties> <maven.compiler.source>8</maven.compiler.source> <maven.compiler.target>8</maven.compiler.target> </properties> <dependencies> <dependency> <groupid>com.alian</groupid> <artifactid>common-rocketmq-dto</artifactid> <version>1.0.0-snapshot</version> </dependency> </dependencies> </project>
4.2 application配置
application.properties
server.port=8008 # rocketmq地址 rocketmq.name-server=192.168.0.234:9876 # 默认的消费者组 rocketmq.consumer.group=concurrent_consumer_group # 批量拉取消息的数量 rocketmq.consumer.pull-batch-size=10 # 集群消费模式 rocketmq.consumer.message-model=clustering
实际上对于本文来说,下面两个配置不用配置,也不会生效。
# 默认的消费者组 rocketmq.consumer.group=concurrent_consumer_group # 集群消费模式 rocketmq.consumer.message-model=clustering
因为优先的是@rocketmqmessagelistener 注解中设置 consumergroup 和messagemodel 参数。
4.3 消费字符串消息
stringmessageconsumer.java
package com.alian.concurrent; import lombok.extern.slf4j.slf4j; import org.apache.rocketmq.spring.annotation.rocketmqmessagelistener; import org.apache.rocketmq.spring.core.rocketmqlistener; import org.springframework.stereotype.service; @slf4j @service @rocketmqmessagelistener(topic = "string_message_topic", consumergroup = "concurrent_group_string", consumethreadnumber = 1) public class stringmessageconsumer implements rocketmqlistener<string> { @override public void onmessage(string message) { log.info("字符串消费者接收到的消息: {}", message); // 处理消息的业务逻辑 } }
如果要消费消息,我们需要实现rocketmqlistener<t>,实现onmessage方法。发送的是什么对象,我们接收就是什么对象,也就是t是什么对象。生产者发送的字符串消息,我们这里就用string接收,也就是rocketmqlistener<string>。
package org.apache.rocketmq.spring.core; public interface rocketmqlistener<t> { void onmessage(t var1); }
同时加上@rocketmqmessagelistener注解,主要用到三个注解
- topic:指定该消费者订阅的topic名称,可以是单个topic,也可以是用||分隔的多个topic
- consumergroup:指定该消费者所属的消费组名称,消费组用于组织同一类消费实例,相同消费组的消费实例可以消费完全相同的消息,rocketmq 通过消费者组(consumer group)来维护不同消费者的消费进度。 每个消费者组都有一个消费进度(offset),用于标记该组下的消费者在某个主题(topic)和队列(queue)上已经消费到的位置。
需要注意的是:如果topic不存在,只有在生产者发送消息时,并且autocreatetopicenable设置为true的情况下,broker端才会自动创建该topic。消费者启动时,即使autocreatetopicenable=true,也不会自动创建不存在的topic。
具体来说:
- 生产者启动并发送消息到一个不存在的topic时:
- 如果autocreatetopicenable=true,broker会自动创建该topic,允许消息发送成功
- 如果autocreatetopicenable=false,broker会拒绝消息发送,报错该topic不存在
- 消费者启动订阅一个不存在的topic时:
- 无论autocreatetopicenable的值是true还是false,broker都不会自动创建该topic
- 消费者会启动成功,但获取不到消息,处于持续等待状态
所以生产者发送消息时才可能自动创建topic,而消费者启动时是不会自动创建topic的。
4.4 消费json消息
jsonmessageconsumer.java
package com.alian.concurrent; import com.alibaba.fastjson.jsonobject; import lombok.extern.slf4j.slf4j; import org.apache.rocketmq.spring.annotation.rocketmqmessagelistener; import org.apache.rocketmq.spring.core.rocketmqlistener; import org.springframework.stereotype.service; @slf4j @service @rocketmqmessagelistener(topic = "json_message_topic", consumergroup = "concurrent_group_json") public class jsonmessageconsumer implements rocketmqlistener<jsonobject> { @override public void onmessage(jsonobject json) { log.info("json消费者接收到的消息: {}", json); // 处理消息的业务逻辑 } }
生产者发送的jsonobject消息,我们这里就用jsonobject接收,也就是rocketmqlistener<jsonobject>
4.5 消费java对象消息
javaobjectmessageconsumer.java
package com.alian.concurrent; import com.alian.common.member; import lombok.extern.slf4j.slf4j; import org.apache.rocketmq.spring.annotation.rocketmqmessagelistener; import org.apache.rocketmq.spring.core.rocketmqlistener; import org.springframework.stereotype.service; @slf4j @service @rocketmqmessagelistener(topic = "java_object_message_topic", consumergroup = "concurrent_group_java_object") public class javaobjectmessageconsumer implements rocketmqlistener<member> { @override public void onmessage(member member) { // 因为发序列化的原因member必须是同一个包 log.info("java对象消费者接收到的消息: {}", member); // 处理消息的业务逻辑 } }
生产者发送的member消息,我们这里就用member接收,也就是rocketmqlistener<member>。
这里需要再次说明下发送java对象消息时,因为反序列的原因,所以生产者和消费者使用的是公共包里同一个对象,也就是发送和接收的对象的包路径要一致。
五、部分运行结果
5.1、字符串消息
同步发送返回的结果:sendresult [sendstatus=send_ok, msgid=7f000001339418b4aac22db848c70000, offsetmsgid=c0a800ea00002a9f0000000000000000, messagequeue=messagequeue [topic=string_message_topic, brokername=broker-a, queueid=1], queueoffset=0] 字符串消费者接收到的消息: 我是一条同步文本消息:syncsendstringmessage
5.2、json消息
同步发送返回的结果:sendresult [sendstatus=send_ok, msgid=7f00000137fc18b4aac22db9d42f0000, offsetmsgid=c0a800ea00002a9f0000000000000146, messagequeue=messagequeue [topic=json_message_topic, brokername=broker-a, queueid=1], queueoffset=0] json消费者接收到的消息: {"name":"alian","age":"28","hobby":"java"}
5.3、java对象消息
同步发送返回的结果:sendresult [sendstatus=send_ok, msgid=7f000001098c18b4aac22dbaceb50000, offsetmsgid=c0a800ea00002a9f0000000000000270, messagequeue=messagequeue [topic=java_object_message_topic, brokername=broker-a, queueid=0], queueoffset=0] java对象消费者接收到的消息: member(id=10086, membername=alian, age=28, birthday=sat mar 09 21:06:57 cst 2024)
5.4、现象说明
为什么在rocketmq中当autocreatetopicenable=true,先启动消费者,然后生产者第一次向一个未创建的topic中发送消息时,消息发送成功了(马上返回成功了),但是消费者要等一段时间才能收到?这种现象的原因是rocketmq在创建topic时,存在一个延迟同步的过程。具体来说:
- 当生产者发送消息到一个不存在的topic时,broker会自动创建该topic。
- 但是broker创建topic后,并不会立即将其元数据信息同步给所有的消费者,而是异步延迟同步。
- 生产者发送消息时,broker会立即返回发送成功响应,因为它只需将消息持久化存储即可。
- 消费者在启动时已订阅了该topic,但由于元数据未同步,暂时无法获知该topic的路由信息,所以无法立即开始消费。
- 过了一段时间后,broker将新创建的topic元数据同步给了消费者,消费者才开始从该topic拉取并消费消息。
这种延迟同步机制是rocketmq的一个设计,目的是为了减小创建topic时对消费者的影响,避免大量消费者同时更新元数据造成系统抖动。
我们可以通过调整broker的配置项来控制这个延迟同步时间,比如:
- brokertopicputenable=true开启自动创建topic功能
- topicdelayenable=true开启延时创建topic功能
- topicdelayoffsetenabled=true开启延时创建topic偏移量同步功能
通过调小topicdelayoffsetinterval可以缩短元数据同步延迟时间,但也会增加系统开销。
所以这种现象实际上是rocketmq的一个正常设计行为,目的是为了系统整体的健壮性和可用性。如果应用对延迟时间不太敏感,保持默认配置即可;如果对延迟敏感,可以适当调小延迟同步时间。
以上就是springboot整合rocketmq实现发送同步消息的详细内容,更多关于springboot rocketmq发送同步消息的资料请关注代码网其它相关文章!
发表评论