189 8069 5689
这篇“SpringBoot如何整合RocketMQ事务、广播以及顺序消息”文章的知识点大部分人都不太理解，所以小编给大家总结了以下内容，内容详细，步骤清晰，具有一定的借鉴价值，希望大家阅读完这篇文章能有所收获，下面我们一起来看看这篇“SpringBoot如何整合RocketMQ事务、广播以及顺序消息”文章吧。
环境：springboot2.3.9RELEASE + RocketMQ4.8.0
org.springframework.boot spring-boot-starter-web org.apache.rocketmq rocketmq-spring-boot-starter 2.2.0
server: port: 8080 --- rocketmq: nameServer: localhost:9876 producer: group: demo-mq
发送
@Resource private RocketMQTemplate rocketMQTemplate ; public void send(String message) { rocketMQTemplate.convertAndSend("test-topic:tag2", MessageBuilder.withPayload(message).build()); }接受
@RocketMQMessageListener(topic = "test-topic", consumerGroup = "consumer01-group", selectorExpression = "tag1 || tag2") @Component public class ConsumerListener implements RocketMQListener{ @Override public void onMessage(String message) { System.out.println("接收到消息：" + message) ; } }
发送
@Resource private RocketMQTemplate rocketMQTemplate ; public void sendOrder(String topic, String message, String tags, int id) { rocketMQTemplate.asyncSendOrderly(topic + ":" + tags, MessageBuilder.withPayload(message).build(), "order-" + id, new SendCallback() { @Override public void onSuccess(SendResult sendResult) { System.err.println("msg-id: " + sendResult.getMsgId() + ": " + message +"\tqueueId: " + sendResult.getMessageQueue().getQueueId()) ; } @Override public void onException(Throwable e) { e.printStackTrace() ; } }); }这里是根据hashkey将消息发送到不同的队列中
@RocketMQMessageListener(topic = "order-topic", consumerGroup = "consumer02-group", selectorExpression = "tag3 || tag4", consumeMode = ConsumeMode.ORDERLY) @Component public class ConsumerOrderListener implements RocketMQListener{ @Override public void onMessage(String message) { System.out.println(Thread.currentThread().getName() + " 接收到Order消息：" + message) ; } }
consumeMode = ConsumeMode.ORDERLY，指明了消息模式为顺序模式，一个队列，一个线程。
结果
当consumeMode = ConsumeMode.CONCURRENTLY执行结果如下：
发送端
@Resource private RocketMQTemplate rocketMQTemplate ; public void send(String topic, String message, String tags) { rocketMQTemplate.send(topic + ":" + tags, MessageBuilder.withPayload(message).build()) ; }消费端
@RocketMQMessageListener(topic = "broad-topic", consumerGroup = "consumer03-group", selectorExpression = "tag6 || tag7", messageModel = MessageModel.CLUSTERING) @Component public class ConsumerBroadListener implements RocketMQListener{ @Override public void onMessage(String message) { System.out.println("ConsumerBroadListener1接收到消息：" + message) ; } }
messageModel = MessageModel.CLUSTERING
测试
启动两个服务分别端口是8080，8081
8080服务
8081服务
集群消息模式下，每个服务分别接收一部分消息，实现了负载均衡
消费端
@RocketMQMessageListener(topic = "broad-topic", consumerGroup = "consumer03-group", selectorExpression = "tag6 || tag7", messageModel = MessageModel.BROADCASTING) @Component public class ConsumerBroadListener implements RocketMQListener{ @Override public void onMessage(String message) { System.out.println("ConsumerBroadListener1接收到消息：" + message) ; } }
messageModel = MessageModel.BROADCASTING
测试
启动两个服务分别端口是8080，8081
8080服务
8081服务
集群消息模式下，每个服务分别都接受了同样的消息。
RocketMQ事务的3个状态
TransactionStatus.CommitTransaction：提交事务消息，消费者可以消费此消息
TransactionStatus.RollbackTransaction：回滚事务，它代表该消息将被删除，不允许被消费。
TransactionStatus.Unknown ：中间状态，它代表需要检查消息队列来确定状态。
RocketMQ实现事务消息主要分为两个阶段：正常事务的发送及提交、事务信息的补偿流程 整体流程为：
正常事务发送与提交阶段
1、生产者发送一个半消息给MQServer(半消息是指消费者暂时不能消费的消息)
2、服务端响应消息写入结果，半消息发送成功
3、开始执行本地事务
4、根据本地事务的执行状态执行Commit或者Rollback操作
事务信息的补偿流程
1、如果MQServer长时间没收到本地事务的执行状态会向生产者发起一个确认回查的操作请求
2、生产者收到确认回查请求后，检查本地事务的执行状态
3、根据检查后的结果执行Commit或者Rollback操作
补偿阶段主要是用于解决生产者在发送Commit或者Rollback操作时发生超时或失败的情况。
发送端
@Resource private RocketMQTemplate rocketMQTemplate ; public void sendTx(String topic, Long id, String tags) { rocketMQTemplate.sendMessageInTransaction(topic + ":" + tags, MessageBuilder.withPayload( new Users(id, UUID.randomUUID().toString().replaceAll("-", ""))). setHeader("BID", UUID.randomUUID().toString().replaceAll("-", "")).build(), UUID.randomUUID().toString().replaceAll("-", "")) ; }生产者对应的监听器
@RocketMQTransactionListener public class ProducerTxListener implements RocketMQLocalTransactionListener { @Resource private BusinessService bs ; @Override public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) { // 这里执行本地的事务操作，比如保存数据。 try { // 创建一个日志记录表，将这的ID存入数据库中，在下面的check方法中可以根据这个id查询是否有数据 String id = (String) msg.getHeaders().get("BID") ; Users users = new JsonMapper().readValue((byte[])msg.getPayload(), Users.class) ; System.out.println("消息内容：" + users + "\t参与数据：" + arg + "\t本次事务的编号：" + id) ; bs.save(users, new UsersLog(users.getId(), id)) ; } catch (Exception e) { e.printStackTrace() ; return RocketMQLocalTransactionState.ROLLBACK ; } return RocketMQLocalTransactionState.COMMIT ; } @Override public RocketMQLocalTransactionState checkLocalTransaction(Message msg) { // 这里检查本地事务是否执行成功 String id = (String) msg.getHeaders().get("BID") ; System.out.println("执行查询ID为：" + id + " 的数据是否存在") ; UsersLog usersLog = bs.queryUsersLog(id) ; if (usersLog == null) { return RocketMQLocalTransactionState.ROLLBACK ; } return RocketMQLocalTransactionState.COMMIT ; } }消费端
@RocketMQMessageListener(topic = "tx-topic", consumerGroup = "consumer05-group", selectorExpression = "tag10") @Component public class ConsumerTxListener implements RocketMQListener{ @Override public void onMessage(Users users) { System.out.println("TX接收到消息：" + users) ; } }
Service
@Transactional public boolean save(Users users, UsersLog usersLog) { usersRepository.save(users) ; usersLogRepository.save(usersLog) ; if (users.getId() == 1) { throw new RuntimeException("数据错误") ; } return true ; } public UsersLog queryUsersLog(String bid) { return usersLogRepository.findByBid(bid) ; }Controller
@GetMapping("/tx/{id}") public Object sendTx(@PathVariable("id")Long id) { ps.sendTx("tx-topic", id, "tag10") ; return "send transaction success" ; }测试
调用接口后，控制台输出：
从打印日志看出来都保存完毕了后 消费端才接受到消息。
删除数据，再测试ID为1会报错的。
数据库中没有数据。。。
是不是也不是很复杂，2个阶段来处理。
以上就是关于“SpringBoot如何整合RocketMQ事务、广播以及顺序消息”这篇文章的内容，相信大家都有了一定的了解，希望小编分享的内容对大家有帮助，若想了解更多相关的知识内容，请关注创新互联行业资讯频道。