How SpringBoot integrates RocketMQ transactions, broadcasts and sequential messages

Environment: springboot2.3.9RELEASE RocketMQ4.8.0

Dependency

   org.springframework.boot     spring-boot-starter-web       org.apache.rocketmq     rocketmq-spring-boot-starter     2.2.0 

Configuration file

server:   port: 8080 --- rocketmq:   nameServer: localhost:9876   producer:     group: demo-mq

Normal message

Send

@Resource private RocketMQTemplate rocketMQTemplate ;      public void send(String message) {   rocketMQTemplate.convertAndSend("test-topic:tag2", MessageBuilder.withPayload(message).build()); }

Accept

@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) ;     }  }

Sequential messages

Send

@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() ;                 }             }); }

Here are messages sent to different queues based on 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, Indicates that the message mode is sequential mode, one queue and one thread.

Result

When consumeMode = ConsumeMode.CONCURRENTLY, the execution result is as follows:

Cluster/Broadcast Message mode

Sender

@Resource private RocketMQTemplate rocketMQTemplate ;      public void send(String topic, String message, String tags) {     rocketMQTemplate.send(topic + ":" + tags, MessageBuilder.withPayload(message).build()) ; }

Cluster message mode

Consumer

@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

Test

Start two services with ports 8080 and 8081

8080 service

8081 service

In the cluster message mode, each service receives a part of the message separately to achieve load balancing

Broadcast message mode

Consumer end

@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

Test

Start two services with ports 8080 and 8081

8080 service

8081 service

In cluster message mode, each service receives the same message.

Transaction message

The 3 statuses of RocketMQ transactions

TransactionStatus.CommitTransaction: Submit transaction message, consumers can consume this message

TransactionStatus.RollbackTransaction: Rolling back the transaction means that the message will be deleted and not allowed to be consumed.

TransactionStatus.Unknown: Intermediate status, which represents the need to check the message queue to determine the status.

RocketMQ's implementation of transaction messages is mainly divided into two stages: normal transaction sending and submission, and transaction information compensation process. The overall process is:

Normal transaction sending and submission stage

1. The producer sends a half message to MQServer (a half message refers to a message that the consumer cannot consume temporarily)

2. The server responds to the message writing result and the half message is sent successfully

3. Start executing the local transaction

4. Perform the Commit or Rollback operation according to the execution status of the local transaction

Compensation process of transaction information

1. If MQServer does not receive it for a long time The execution status of the local transaction will initiate a confirmation review operation request to the producer

2. After the producer receives the confirmation review request, it checks the execution status of the local transaction

3. According to After checking the results, execute the Commit or Rollback operation.

The compensation phase is mainly used to solve the problem of timeout or failure when the producer sends the Commit or Rollback operation.

Sender

@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("-", "")) ; }

The listener corresponding to the producer

@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 ;     }  }

Consumer

@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" ; }

Test

After calling the interface, the console output:

It can be seen from the print log that the consumer receives the message only after all the messages have been saved.

Deleting the data and then testing the ID as 1 will result in an error.

There is no data in the database. . .

Isn’t it very complicated? It can be handled in 2 stages.

The above is the detailed content of How SpringBoot integrates RocketMQ transactions, broadcasts and sequential messages. For more information, please follow other related articles on the PHP Chinese website!