建站
咨詢(xún)
售后
建站咨詢(xú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è)線(xiàn)程。
結(jié)果
當(dāng)consumeMode = ConsumeMode.CONCURRENTLY執(zhí)行結(jié)果如下：
集群/廣播消息模式
發(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ù)
8081服務(wù)
集群消息模式下，每個(gè)服務(wù)分別接收一部分消息，實(shí)現(xiàn)了負(fù)載均衡
廣播消息模式
消費(fèi)端
- @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
測(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)輸出：
從打印日志看出來(lái)都保存完畢了后 消費(fèi)端才接受到消息。
刪除數(shù)據(jù)，再測(cè)試ID為1會(huì)報(bào)錯(cuò)的。
數(shù)據(jù)庫(kù)中沒(méi)有數(shù)據(jù)。
是不是也不是很復(fù)雜，2個(gè)階段來(lái)處理。
完畢!!!
新聞名稱(chēng)：SpringBoot整合RocketMQ事務(wù)/廣播/順序消息
網(wǎng)頁(yè)路徑：http://www.dlmjj.cn/article/cddccog.html
