Environment: springboot2.3.9RELEASE RocketMQ4.8.0

Dependency

<dependency>???<groupid>org.springframework.boot</groupid>?????<artifactid>spring-boot-starter-web</artifactid>?</dependency>?<dependency>?????<groupid>org.apache.rocketmq</groupid>?????<artifactid>rocketmq-spring-boot-starter</artifactid>?????<version>2.2.0</version>?</dependency>

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<string>?{??????@Override?????public?void?onMessage(String?message)?{?????????System.out.println("接收到消息："?+?message)?;?????}??}</string>

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<string>?{??????@Override?????public?void?onMessage(String?message)?{?????????System.out.println(Thread.currentThread().getName()?+?"?接收到Order消息："?+?message)?;?????}??}</string>

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<string>?{??????@Override?????public?void?onMessage(String?message)?{?????????System.out.println("ConsumerBroadListener1接收到消息："?+?message)?;?????}??}</string>

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<string>?{??????@Override?????public?void?onMessage(String?message)?{?????????System.out.println("ConsumerBroadListener1接收到消息："?+?message)?;?????}??}</string>

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)?{?????????//?這里執(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?;?????}??}

Consumer

@RocketMQMessageListener(topic?=?"tx-topic",?consumerGroup?=?"consumer05-group",?selectorExpression?=?"tag10")?@Component?public?class?ConsumerTxListener?implements?RocketMQListener<users>?{??????@Override?????public?void?onMessage(Users?users)?{?????????System.out.println("TX接收到消息："?+?users)?;?????}??}</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"?;?}

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!