SpringBoot怎么整合RocketMQ事務(wù)、廣播以及順序消息

本篇內(nèi)容介紹了“SpringBoot怎么整合RocketMQ事務(wù)、廣播以及順序消息”的有關(guān)知識(shí)，在實(shí)際案例的操作過(guò)程中，不少人都會(huì)遇到這樣的困境，接下來(lái)就讓小編帶領(lǐng)大家學(xué)習(xí)一下如何處理這些情況吧！希望大家仔細(xì)閱讀，能夠?qū)W有所成！

為原州等地區(qū)用戶(hù)提供了全套網(wǎng)頁(yè)設(shè)計(jì)制作服務(wù)，及原州網(wǎng)站建設(shè)行業(yè)解決方案。主營(yíng)業(yè)務(wù)為網(wǎng)站建設(shè)、網(wǎng)站制作、原州網(wǎng)站設(shè)計(jì)，以傳統(tǒng)方式定制建設(shè)網(wǎng)站，并提供域名空間備案等一條龍服務(wù)，秉承以專(zhuān)業(yè)、用心的態(tài)度為用戶(hù)提供真誠(chéng)的服務(wù)。我們深信只要達(dá)到每一位用戶(hù)的要求，就會(huì)得到認(rèn)可，從而選擇與我們長(zhǎng)期合作。這樣，我們也可以走得更遠(yuǎn)！

環(huán)境：springboot2.3.9RELEASE + RocketMQ4.8.0

依賴(lài)

   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

普通消息

發(fā)送

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

順序消息

發(fā)送

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

這里是根據(jù)hashkey將消息發(fā)送到不同的隊(duì)列中

@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，指明了消息模式為順序模式，一個(gè)隊(duì)列，一個(gè)線程。

結(jié)果

SpringBoot怎么整合RocketMQ事務(wù)、廣播以及順序消息

當(dāng)consumeMode = ConsumeMode.CONCURRENTLY執(zhí)行結(jié)果如下：

SpringBoot怎么整合RocketMQ事務(wù)、廣播以及順序消息

集群/廣播消息模式

發(fā)送端

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

集群消息模式

消費(fèi)端

@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

測(cè)試

啟動(dòng)兩個(gè)服務(wù)分別端口是8080，8081

8080服務(wù)

SpringBoot怎么整合RocketMQ事務(wù)、廣播以及順序消息

8081服務(wù)

SpringBoot怎么整合RocketMQ事務(wù)、廣播以及順序消息

集群消息模式下，每個(gè)服務(wù)分別接收一部分消息，實(shí)現(xiàn)了

messageModel = MessageModel.BROADCASTING

測(cè)試

啟動(dòng)兩個(gè)服務(wù)分別端口是8080，8081

8080服務(wù)

8081服務(wù)

集群消息模式下，每個(gè)服務(wù)分別都接受了同樣的消息。

事務(wù)消息

RocketMQ事務(wù)的3個(gè)狀態(tài)

TransactionStatus.CommitTransaction：提交事務(wù)消息，消費(fèi)者可以消費(fèi)此消息

TransactionStatus.RollbackTransaction：回滾事務(wù)，它代表該消息將被刪除，不允許被消費(fèi)。

TransactionStatus.Unknown ：中間狀態(tài)，它代表需要檢查消息隊(duì)列來(lái)確定狀態(tài)。

RocketMQ實(shí)現(xiàn)事務(wù)消息主要分為兩個(gè)階段：正常事務(wù)的發(fā)送及提交、事務(wù)信息的補(bǔ)償流程整體流程為：

正常事務(wù)發(fā)送與提交階段

1、生產(chǎn)者發(fā)送一個(gè)半消息給MQServer(半消息是指消費(fèi)者暫時(shí)不能消費(fèi)的消息)

2、服務(wù)端響應(yīng)消息寫(xiě)入結(jié)果，半消息發(fā)送成功

3、開(kāi)始執(zhí)行本地事務(wù)

4、根據(jù)本地事務(wù)的執(zhí)行狀態(tài)執(zhí)行Commit或者Rollback操作

事務(wù)信息的補(bǔ)償流程

1、如果MQServer長(zhǎng)時(shí)間沒(méi)收到本地事務(wù)的執(zhí)行狀態(tài)會(huì)向生產(chǎn)者發(fā)起一個(gè)確認(rèn)回查的操作請(qǐng)求

2、生產(chǎn)者收到確認(rèn)回查請(qǐng)求后，檢查本地事務(wù)的執(zhí)行狀態(tài)

3、根據(jù)檢查后的結(jié)果執(zhí)行Commit或者Rollback操作

補(bǔ)償階段主要是用于解決生產(chǎn)者在發(fā)送Commit或者Rollback操作時(shí)發(fā)生超時(shí)或失敗的情況。

發(fā)送端

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

生產(chǎn)者對(duì)應(yīng)的監(jiān)聽(tīng)器

@RocketMQTransactionListener public class ProducerTxListener implements RocketMQLocalTransactionListener {          @Resource     private BusinessService bs ;      @Override     public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {         // 這里執(zhí)行本地的事務(wù)操作，比如保存數(shù)據(jù)。         try {             // 創(chuàng)建一個(gè)日志記錄表，將這唯一的ID存入數(shù)據(jù)庫(kù)中，在下面的check方法中可以根據(jù)這個(gè)id查詢(xún)是否有數(shù)據(jù)             String id = (String) msg.getHeaders().get("BID") ;             Users users = new JsonMapper().readValue((byte[])msg.getPayload(), Users.class) ;             System.out.println("消息內(nèi)容：" + users + "\t參與數(shù)據(jù)：" + arg + "\t本次事務(wù)的唯一編號(hào)：" + 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) {         // 這里檢查本地事務(wù)是否執(zhí)行成功         String id = (String) msg.getHeaders().get("BID") ;         System.out.println("執(zhí)行查詢(xún)ID為：" + id + " 的數(shù)據(jù)是否存在") ;         UsersLog usersLog = bs.queryUsersLog(id) ;         if (usersLog == null) {             return RocketMQLocalTransactionState.ROLLBACK ;         }         return RocketMQLocalTransactionState.COMMIT ;     }  }

消費(fèi)端

@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("數(shù)據(jù)錯(cuò)誤") ;     }     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" ; }

測(cè)試

調(diào)用接口后，控制臺(tái)輸出：