本文将主要介绍在SpringBoot项目中如何集成RocketMQ以实现普通消息和事务消息的。
首先是分别创建生产者的springboot项目 springboot-rocketmq-producer，创建消费者的springboot项目 springboot-rocketmq-consumer。
1. 引入依赖
本例中使用的RocketMQ的版本是 5.1.3。所以引入的 rocketmq-spring-boot 版本要与之匹配。
可以通过mvnrepository进行查看。https://mvnrepository.com/artifact/org.apache.rocketmq/rocketmq-spring-boot/2.2.2
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot</artifactId>
<version>2.2.2</version>
</dependency>
2. 配置文件修改
在springboot-rocketmq-producer项目的application.yml文件中添加如下配置：
rocketmq:
name-server: 172.31.184.89:9876
producer:
group: feige-producer-group
consumer:
topic: my-spring-boot-topic
在springboot-rocketmq-consumer项目的application.yml文件中添加如下配置：
server:
port: 8080
rocketmq:
name-server: 172.31.184.89:9876
consumer:
group: feige-consumer-group
topic: my-spring-boot-topic
3. 实现生产者
定义一个生产者类MyProducer，在该类中引入RocketMQTemplate 操作类，然后定义发送消息的方法sendMessage，在此方法中调用 rocketMQTemplate.convertAndSend 方法进行消息发送。
@Component
public class MyProducer {
@Autowired
private RocketMQTemplate rocketMQTemplate;
/**
* 发送普通消息
*
* @param topic 主题
* @param message 消息
*/
public void sendMessage(String topic, String message) {
rocketMQTemplate.convertAndSend(topic, message);
}
3.1. 编写生产者单元测试
@Autowired
private MyProducer myProducer;
@Value("${rocketmq.consumer.topic:}")
private String consumerTopic;
@Test
void sendMessage() {
myProducer.sendMessage(consumerTopic,"飞哥SpringBoot集成RocketMQ消息测试");
}
4.实现消费者
定义消费者类MyConsumer。此类实现了RocketMQListener接口并重写了onMessage方法用于接收broker推送过来的消息。
@Component
@RocketMQMessageListener(topic = "${rocketmq.consumer.topic:}", consumerGroup = "generalConsumerGroup")
public class MyConsumer implements RocketMQListener<String> {
@Override
public void onMessage(String s) {
System.out.println("收到的消息是=" + s);
}
}
5. 实现事务消息
在SpringBoot中实现RocketMQ的事务消息，整体思路与 【RocketMQ系列六】RocketMQ事务消息 文中提到的思路相同。
5.1. 实现事务消息的生产者
在前面创建的MyProducer类中添加实现事务消息的方法 sendTransactionMessage。
/**
* 发送事务消息
*
* @param topic 话题
* @param msg 消息
*/
public void sendTransactionMessage(String topic, String msg) throws InterruptedException {
String[] tags = {"tagA", "tagB", "tagC", "tagD", "tagE"};
for (int i = 0; i < 10; i++) {
// 2. 将topic和tag整合在一起，以:分割，
String destination = topic + ":" + tags[i % tags.length];
// 1.注意该message是org.springframework.messaging.Message
Message<String> message = MessageBuilder.withPayload(msg + "_" + tags[i % tags.length] + "_" + i)
.setHeader("destination", destination).build();
// 第一个参数是发布的目的地，第二个参数是消息，第三个参数是额外的参数
rocketMQTemplate.sendMessageInTransaction(destination, message, destination);
Thread.sleep(10);
}
}
这里需要注意的是传入的Message类是org.springframework.messaging.Message ，不是RocketMQ的Message。
5.2. 实现本地事务消息
接着在定义生产者本地事务实现类 MyTransactionListener，该类实现了RocketMQLocalTransactionListener接口，并重写了executeLocalTransaction方法和checkLocalTransaction方法。这里多了一步就是将 org.springframework.messaging.Message 转成 org.apache.rocketmq.common.message.Message 。
@RocketMQTransactionListener(rocketMQTemplateBeanName = "rocketMQTemplate")
public class MyTransactionListener implements RocketMQLocalTransactionListener {
@Override
public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
// 将消息转成rocketmq下的message
org.apache.rocketmq.common.message.Message message = RocketMQUtil.convertToRocketMessage(new StringMessageConverter(), "utf-8", (String) arg, msg);
String tags = message.getTags();
if (tags.equals("tagA")) {
return RocketMQLocalTransactionState.COMMIT;
} else if (tags.equals("tagB")) {
return RocketMQLocalTransactionState.ROLLBACK;
}
return RocketMQLocalTransactionState.UNKNOWN;
}
@Override
public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
// 将消息转成rocketmq下的message
String destination = (String) msg.getHeaders().get("destination");
org.apache.rocketmq.common.message.Message message = RocketMQUtil.convertToRocketMessage(new StringMessageConverter(),
"utf-8",destination, msg);
String tags = message.getTags();
if (tags.equals("tagC")) {
return RocketMQLocalTransactionState.COMMIT;
} else if (tags.equals("tagD")) {
return RocketMQLocalTransactionState.ROLLBACK;
}
return RocketMQLocalTransactionState.UNKNOWN;
}
}