关于消息队列这部分内容，之前自己简单了解和学习过Kafka，但是目前国内公司对RocketMq使用的也比较多，并且我所在的项目组已经引入了RocketMq，但是因为项目进度问题，在开发中还没有正式的使用。所以自己准别先提前了解一下，关于RocketMq我想不用我过多介绍，而且性能据说也是特别的好，并且也支持专门的事务消息，可以说非常的有吸引力。学习的资料无非也就是官方文档。这次学习使用的是4.4.0版本。
下面就正式开始今天的RocketMq学习，当然首先还是创建demo项目
一、创建项目
今天的主要目的就是学习如何在Spring Boot中使用RocketMq，包括消息的生产、消费（同步、异步）。当然如果时间允许的话我也会尝试使用消息广播和事务消息的发送、消费。
首先创建项目，引入相关依赖，这里说一下因为是和Spring Boot整合使用，我并没有使用官方的依赖，而是使用了starter其对rocketmq-client又做了一层的封装，个人感觉使用起来更方便，Spring Boot版本使用的是：2.3.1.RELEASE，项目的pom文件如下：
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.postgresql</groupId>
<artifactId>postgresql</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>org.junit.vintage</groupId>
<artifactId>junit-vintage-engine</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<!--rocketMQ-->
<!-- <dependency>-->
<!-- <groupId>org.apache.rocketmq</groupId>-->
<!-- <artifactId>rocketmq-client</artifactId>-->
<!-- <version>4.4.0</version>-->
<!-- </dependency>-->
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.0.4</version>
</dependency>
</dependencies>
关于普通消息的生产、消费我准备用一个异步写库的场景（这个在上家公司遇到过，不过当时使用的是阿里云上的消息队列），首先创建相应的Service和其实现类，为了同时模拟同步和异步消息，我决定在新增数据时发送同步消息，更新时发送异步消息，代码如下：
@Slf4j
@Service
public class UserServiceImpl implements UserService {
private static final String USER_TOPIC = "user_topic:";
private static final String UPDATE_TAG = "update_tag";
private static final String INSERT_TAG = "insert_tag";
private RocketMQTemplate rocketMQTemplate;
public UserServiceImpl(RocketMQTemplate rocketMQTemplate) {
this.rocketMQTemplate = rocketMQTemplate;
}
@Override
public Boolean update(UserEntity user) {
user.setUpdateTime(new Date());
rocketMQTemplate.asyncSend(USER_TOPIC + UPDATE_TAG, user, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
String msgId = sendResult.getMsgId();
log.info(">>>> async message success, send status={},message id={} <<<<",sendResult.getSendStatus().name(),msgId);
// 其他
}
@Override
public void onException(Throwable throwable) {
log.error(">>>> async message success, exception message={} <<<<",throwable.getMessage());
// 其他
}
});
return Boolean.TRUE;
}
@Override
public Boolean insert(UserEntity user) {
Date now = new Date();
user.setCreateTime(now);
user.setUpdateTime(now);
SendResult sendResult = rocketMQTemplate.syncSend(USER_TOPIC + INSERT_TAG,user);
log.info(">>>> send message success, send status={} <<<<",sendResult.getSendStatus().name());
if (sendResult.getSendStatus().name().equals(SendStatus.SEND_OK.name())) {
return Boolean.TRUE;
}
return Boolean.FALSE;
}
}
之所以选择使用RocketMq的starter，就是因为它在官方的基础之上又进行了一层的封装，使用起来非常的方便，消息的发送通过RocketMQTemplate就可以了。RocketMq有几个核心的概念分别是：Producer Group、Consumer Group、Topic、Tag等需要提前了解一下，这里简单说一下Producer Group即生产者组，就是相同类型消息生产者的集合，如果这个组的某一个生产者挂掉，那么其这个组的其他生产者依然可以提供服务。Consumer Group即消费者组，同Producer Group类似。Topic，即主题，也就是生产者发生消息以及消费者拉取消息的地方，它和生产者、消费者的关联比较松散。Tag也叫子主题，相对与Topic而言它的粒度更小一些，它可以让用户对消息处理更精细一点。在发送消息时的目的地址就是Topic:Tag，当然如果也可以使用*代替Tag，这样消息会发送到所有订阅该主题的消费者。
这里使用了RocketMQTemplate发送消息，消息发出去了，该如何接收消息呢，starter提供了RocketMQListener接口，该接口主要是普通消息的接收，如果是事务消息，则需实现RocketMQLocalTransactionListener接口。我编写两个消费者，消费不同的Tag，代码如下：
## 订阅user_topic insert_tag的消息监听器，注意consumerGroup名称
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "user_group",topic = "user_topic",selectorType = SelectorType.TAG,selectorExpression = "insert_tag")
public class InsertUserMessageListener implements RocketMQListener<UserEntity> {
private UserRepository userRepository;
private static Gson GSON = new Gson();
public InsertUserMessageListener(UserRepository userRepository) {
this.userRepository = userRepository;
}
@Override
public void onMessage(UserEntity userEntity) {
log.info(">>>> user message listener start,Tag=insert_tag, message={} <<<<",GSON.toJson(userEntity));
// 写入数据库
userRepository.save(userEntity);
}
}
## 订阅user_topic update_tag的消息监听器，注意consumerGroup名称
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "user_group",topic = "user_topic",selectorType = SelectorType.TAG,selectorExpression = "update_tag")
public class UpdateUserMessageListener implements RocketMQListener<UserEntity> {
private UserRepository userRepository;
private static Gson GSON = new Gson();
public UpdateUserMessageListener(UserRepository userRepository) {
this.userRepository = userRepository;
}
@Override
public void onMessage(UserEntity userEntity) {
log.info(">>>> user message listener start,Tag=update_tag, message={} <<<<",GSON.toJson(userEntity));
// 写入数据库
userRepository.save(userEntity);
}
}
好了，接下来启动本地的RocketMq进行测试。
二、测试消息的发送、接收
RocketMq其实对内存占用是非常高的，JVM堆内存大小在4G到8G，个人觉得除非你的内存真的很大，不然还是改小一些比较好。主要是修改bin目录下的runserver.sh和runbroker.sh(windows系统修改runserver.cmd和runbroker.cmd)，然后分别启动nameserver和broker
## 启动namesrv
nohup sh bin/mqnamesrv &
## 启动broker
nohup sh bin/mqbroker -n localhost:9876 -c conf/broker.conf &
启动好RocketMq之后，需要在项目的配置文件配置好RocketMq的相关配置，application.properties如下：
spring.application.name=rocketMQ-service
spring.datasource.url=jdbc:postgresql://localhost:5432/pgsql?useSSL=false&characterEncoding=utf8
spring.datasource.username=postgres
spring.datasource.password=123456
spring.datasource.driver-class-name=org.postgresql.Driver
spring.jpa.show-sql=true
spring.jpa.database=postgresql
spring.jpa.hibernate.ddl-auto=update
spring.jpa.hibernate.naming.physical-strategy=org.hibernate.boot.model.naming.PhysicalNamingStrategyStandardImpl
spring.jpa.properties.hibernate.dialect=org.hibernate.dialect.PostgreSQLDialect
spring.jpa.properties.hibernate.temp.use_jdbc_metadata_defaults=false
## rocketmq config
rocketmq.name-server=localhost:9876
rocketmq.producer.group=rocket_group
rocketmq.producer.send-message-timeout=10000
主要配置了rocketMq的namesrv地址和生产者组名以及发送消息超时时间，启动项目，并调用分别执行新增和更新的接口。
但是在项目启动过程中遇到了问题，导致项目启动失败，异常信息如下：
java.lang.RuntimeException: java.lang.IllegalStateException: Failed to start RocketMQ push consumer
at org.apache.rocketmq.spring.autoconfigure.ListenerContainerConfiguration.registerContainer(ListenerContainerConfiguration.java:123) ~[rocketmq-spring-boot-2.0.4.jar:2.0.4]
at java.util.LinkedHashMap.forEach(LinkedHashMap.java:684) ~[na:1.8.0_171]
at org.apache.rocketmq.spring.autoconfigure.ListenerContainerConfiguration.afterSingletonsInstantiated(ListenerContainerConfiguration.java:82) ~[rocketmq-spring-boot-2.0.4.jar:2.0.4]
at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:910) ~[spring-beans-5.2.7.RELEASE.jar:5.2.7.RELEASE]
at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:879) ~[spring-context-5.2.7.RELEASE.jar:5.2.7.RELEASE]
at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:551) ~[spring-context-5.2.7.RELEASE.jar:5.2.7.RELEASE]
at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:143) ~[spring-boot-2.3.1.RELEASE.jar:2.3.1.RELEASE]
at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:758) [spring-boot-2.3.1.RELEASE.jar:2.3.1.RELEASE]
at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:750) [spring-boot-2.3.1.RELEASE.jar:2.3.1.RELEASE]
at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:397) [spring-boot-2.3.1.RELEASE.jar:2.3.1.RELEASE]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:315) [spring-boot-2.3.1.RELEASE.jar:2.3.1.RELEASE]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1237) [spring-boot-2.3.1.RELEASE.jar:2.3.1.RELEASE]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1226) [spring-boot-2.3.1.RELEASE.jar:2.3.1.RELEASE]
at com.ypc.mq.rocketmq.RocketmqApplication.main(RocketmqApplication.java:10) [classes/:na]
Caused by: java.lang.IllegalStateException: Failed to start RocketMQ push consumer
at org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer.start(DefaultRocketMQListenerContainer.java:277) ~[rocketmq-spring-boot-2.0.4.jar:2.0.4]
at org.apache.rocketmq.spring.autoconfigure.ListenerContainerConfiguration.registerContainer(ListenerContainerConfiguration.java:120) ~[rocketmq-spring-boot-2.0.4.jar:2.0.4]
... 13 common frames omitted
Caused by: org.apache.rocketmq.client.exception.MQClientException: The consumer group[user_group] has been created before, specify another name please.
See http://rocketmq.apache.org/docs/faq/ for further details.
at org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl.start(DefaultMQPushConsumerImpl.java:628) ~[rocketmq-client-4.5.2.jar:4.5.2]
at org.apache.rocketmq.client.consumer.DefaultMQPushConsumer.start(DefaultMQPushConsumer.java:693) ~[rocketmq-client-4.5.2.jar:4.5.2]
at org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer.start(DefaultRocketMQListenerContainer.java:275) ~[rocketmq-spring-boot-2.0.4.jar:2.0.4]
... 14 common frames omitted
首先根据报错信息The consumer group[user_group] has been created before, specify another name please.可以得出一个结论，那就是消费者组的名称需要唯一，至少在一个项目中要唯一。所以先修改UpdateUserMessageListener消费者的消费者组名称为user_update_group。这样的话如果订阅相同Topic不同的Tag就需要指定不同的消费者组名称，我之前的理解是，一个Topic一个消费者组，看来这种理解是错误的，而应该是一个Topic:Tag一个消费者组。因为消息发送的地址是Topic:Tag，那么显即使Topic相同但Tag不同的话，其实消费者也是不同的，自然消费者的组也就不同。也就是说消费者的分组归根到底是按照Topic:Tag来区分的，我之所以认为是按照Topic，是因为之前 消费者中Tag定义的都是*（默认值），所以才有这样的误解。
修改名称之后，重新启动项目，首先调用新增数据借口，即发送一个同步消息，日志如下：
2020-07-18 17:49:52.977 INFO 21075 --- [nio-9090-exec-1] c.y.m.r.service.impl.UserServiceImpl : >>>> send message success, send status=SEND_OK <<<<
2020-07-18 17:49:52.983 INFO 21075 --- [MessageThread_2] c.y.m.r.l.InsertUserMessageListener : >>>> user message listener start,Tag=insert_tag, message={"id":0,"userName":"李四","mobile":"12364455","sex":"男","birthday":"Apr 3, 2015 12:15:14 PM","createTime":"Jul 18, 2020 5:49:50 PM","updateTime":"Jul 18, 2020 5:49:50 PM"} <<<<
Hibernate: select nextval ('hibernate_sequence')
Hibernate: insert into t_user (birthday, createTime, email, mobile, sex, updateTime, userName, id) values (?, ?, ?, ?, ?, ?, ?, ?)
这里开始自己也有点迷惑，为什么在写库之前返回的状态就是SEND_OK呢？看一下官方文档的解释：
When sending a message, you will get SendResult which contains SendStatus. Firstly, we assume that Message’s isWaitStoreMsgOK=true(default is true). If not, we will always get SEND_OK if no exception is thrown.
...
SEND_OK does not mean it is reliable. To make sure no message would be lost, you should also enable SYNC_MASTER or SYNC_FLUSH.
这里解释的很清楚了，如果在没有修改isWaitStoreMsgOK配置的情况下，只要发送的消息成功写到磁盘，也就是消息持久化到broker的服务器，就会返回SEND_OK，至于消费者有没有消费，生产者并不关心，生产者在消息落地之后，其任务就已经完成。不过如果是事务消息结果应该就不同了。
然后再调用一个异步消息的接口，日志如下：
2020-07-18 18:03:51.677 INFO 21075 --- [ublicExecutor_4] c.y.m.r.service.impl.UserServiceImpl : >>>> async message success, send status=SEND_OK,message id=C0A80069525318B4AAC25B6C69F7000B <<<<
2020-07-18 18:03:51.692 INFO 21075 --- [MessageThread_1] c.y.m.r.l.UpdateUserMessageListener : >>>> user message listener start,Tag=update_tag, message={"id":19,"userName":"李sisi","mobile":"88844111","sex":"男","birthday":"Apr 3, 2015 12:15:14 PM","updateTime":"Jul 18, 2020 6:03:51 PM"} <<<<
Hibernate: select userentity0_.id as id1_1_0_, userentity0_.birthday as birthday2_1_0_, userentity0_.createTime as createti3_1_0_, userentity0_.email as email4_1_0_, userentity0_.mobile as mobile5_1_0_, userentity0_.sex as sex6_1_0_, userentity0_.updateTime as updateti7_1_0_, userentity0_.userName as username8_1_0_ from t_user userentity0_ where userentity0_.id=?
Hibernate: update t_user set birthday=?, createTime=?, email=?, mobile=?, sex=?, updateTime=?, userName=? where id=?
其实不管是新增还是修改操作，其实使用的都是相同的save方法，所以可以将两个消息监听器合并成一个。另外RocketMQMessageListener默认使用的消息模式为Clustering，即集群模式，当然还有Broadcasting，即广播模式，广播模式下所有的消费着都会收到相同的消息。使用广播模式的好处就是可以自动的实现负载均衡，一般场景还是建议使用集群模式，但是如果又真的想使用广播模式的话需要最好通过集群模式来实现，不建议直接使用广播模式。
将上面的两个消息监听器合并，并创建一个新的消息监听器模拟广播模式，代码如下：
## 监听多个tag，tag使用'||'隔开
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "user_group",topic = "user_topic",selectorType = SelectorType.TAG,selectorExpression = "insert_tag || update_tag")
public class UserMessageListener implements RocketMQListener<UserEntity> {
private UserRepository userRepository;
private static Gson GSON = new Gson();
public UserMessageListener(UserRepository userRepository) {
this.userRepository = userRepository;
}
@Override
public void onMessage(UserEntity userEntity) {
log.info(">>>> user message listener start,Tag=insert_tag, message={} <<<<",GSON.toJson(userEntity));
// 写入数据库
userRepository.save(userEntity);
}
}
## 另一个监听器，模拟广播模式，默认是所有的 tag
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "broad_group",topic = "user_topic")
public class BroadMessageListener implements RocketMQListener<UserEntity> {
private UserRepository userRepository;
private static Gson GSON = new Gson();
public BroadMessageListener(UserRepository userRepository) {
this.userRepository = userRepository;
}
@Override
public void onMessage(UserEntity userEntity) {
log.info(">>>> broadcasting model listener start, message={} <<<<",GSON.toJson(userEntity));
// 处理
userRepository.save(userEntity);
}
}
使用集群模式模拟广播模式，只要监听的Topic:Tag一致即可。重新启动项目，并发送一个新增接口，日志如下：
2020-07-18 18:35:21.326 INFO 23280 --- [essageThread_11] c.y.m.r.listener.BroadMessageListener : >>>> broadcasting model listener start, message={"id":0,"userName":"王五","mobile":"25464455","sex":"男","birthday":"Apr 3, 2010 12:15:14 PM","createTime":"Jul 18, 2020 6:35:19 PM","updateTime":"Jul 18, 2020 6:35:19 PM"} <<<<
Hibernate: select nextval ('hibernate_sequence')
Hibernate: insert into t_user (birthday, createTime, email, mobile, sex, updateTime, userName, id) values (?, ?, ?, ?, ?, ?, ?, ?)
2020-07-18 18:35:21.333 INFO 23280 --- [nio-9090-exec-1] c.y.m.r.service.impl.UserServiceImpl : >>>> send message success, send status=SEND_OK <<<<
2020-07-18 18:35:21.342 INFO 23280 --- [MessageThread_2] c.y.m.r.listener.UserMessageListener : >>>> user message listener start,Tag=insert_tag, message={"id":0,"userName":"王五","mobile":"25464455","sex":"男","birthday":"Apr 3, 2010 12:15:14 PM","createTime":"Jul 18, 2020 6:35:19 PM","updateTime":"Jul 18, 2020 6:35:19 PM"} <<<<
Hibernate: select nextval ('hibernate_sequence')
Hibernate: insert into t_user (birthday, createTime, email, mobile, sex, updateTime, userName, id) values (?, ?, ?, ?, ?, ?, ?, ?)
可以看到有两个消费者都有日志输出，且有两条插入sql，查看数据库也有两条完全一样的数据，说明模拟实现广播模式是成功的。
三、总结
本次的内容主要是学习在Spring Boot下使用rocketmq-spring-boot-starter如何进行普通消息的发送、接收以及如何使用集群模式来实现广播模式。这次学习自己对RocketMq的一些核心概念也有了进一步的了解，收获还是蛮多的，当然RocketMq内容还是很多的，通过一两个demo来搞清楚是不现实的，以后需要更多时间和精力来学习，因为时间的问题自己也没有能尝试使用它的事务消息，等之后有时间再通过demo来进一步学习吧。
网友评论