使用RocketMQTemplate发送带tags的消息

时间：2021-11-07 08:01:57

RocketMQTemplate发送带tags的消息

RocketMQTemplate是RocketMQ集成到Spring cloud之后提供的个方便发送消息的模板类，它是基本Spring 的消息机制实现的，对外只提供了Spring抽象出来的消息发送接口。

在单独使用RocketMQ的时候，发送消息使用的Message是‘org.apache.rocketmq.common.message'包下面的Message，而使用RocketMQTemplate发送消息时，使用的Message是org.springframework.messaging的Message，猛一看，没办法发送带tags的消息了，其实在RocketMQ集成的时候已经解决了这个问题。

在RocketMQTemplate发送消息时，调用的方法是：

  1. public SendResult syncSendOrderly(String destination, Message<?> message, String hashKey, long timeout) {
  2. if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
  3. log.error("syncSendOrderly failed. destination:{}, message is null ", destination);
  4. throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
  5. }
  6. try {
  7. long now = System.currentTimeMillis();
  8. //在这里对消息进行了转化，将Spring的message转化为rocketmq自己的message
  9. org.apache.rocketmq.common.message.Message rocketMsg = RocketMQUtil.convertToRocketMessage(objectMapper,
  10. charset, destination, message);
  11. SendResult sendResult = producer.send(rocketMsg, messageQueueSelector, hashKey, timeout);
  12. long costTime = System.currentTimeMillis() - now;
  13. log.debug("send message cost: {} ms, msgId:{}", costTime, sendResult.getMsgId());
  14. return sendResult;
  15. } catch (Exception e) {
  16. log.error("syncSendOrderly failed. destination:{}, message:{} ", destination, message);
  17. throw new MessagingException(e.getMessage(), e);
  18. }
  19. }

在上面的代码中，对消息进行了转化，将Spring的message转化为rocketmq自己的message，在RocketMQUtil.convertToRocketMessage方法中有个地方就是获取tags的：

  1. String[] tempArr = destination.split(":", 2);
  2. String topic = tempArr[0];
  3. String tags = "";
  4. if (tempArr.length > 1) {
  5. tags = tempArr[1];
  6. }

所以，在发送消息的时候，我们只要把tags使用":"添加到topic后面就可以了。

例如：xxxx:tag1 || tag2 || tag3

使用RocketMQ 处理消息

消息发送（生产者）

以maven + SpringBoot 工程为例，先在pom.xml增加依赖

  1. <dependency>
  2. <groupId>org.apache.rocketmq</groupId>
  3. <artifactId>rocketmq-spring-boot-starter</artifactId>
  4. <version>2.0.1</version>
  5. </dependency>

由于，这个依赖是一个starter，直接引入依赖就可以开始写投递消息的代码了。这个starter注册了一个叫org.apache.rocketmq.spring.core.RocketMQTemplate的bean，用它就可以直接把消息投递出去。 具体的API是这样的

  1. XXXEvent xxxDto = new XXXEvent();
  2. Message<XXXEvent> message = MessageBuilder.withPayload(xxxDto).build();
  3. String dest = String.format("%s:%s",topic-name","tag-name");
  4. //默认投递：同步发送 不会丢失消息。如果在投递成功后发生网络异常，客户端会认为投递失败而回滚本地事务
  5. this.rocketMQTemplate.send(dest, xxxDto);

这种投递方式能保证投递成功的消息不会丢失，但是不能保证投递一定成功。假设一次调用的流程是这样的

使用RocketMQTemplate发送带tags的消息

如果在步骤3的时候发生错误，因为出错mqClient会认为消息投递失败而把事务回滚。如果消息已经被消费，那就会导致业务错误。我们可以用事务消息解决这个问题。

以带事务方式投递的消息，正常情况下的处理流程是这样的

使用RocketMQTemplate发送带tags的消息

出错的时候是这样的

使用RocketMQTemplate发送带tags的消息

由于普通消息没有消息回查，普通消息用的producer不支持回查操作，不同业务的回查处理也不一样，事务消息需要使用单独的producer。消息发送代码大概是这样的

  1. //调用这段代码之前别做会影响数据的操作
  2. XXXEvent xxxDto = new XXXEvent();
  3. Message<XXXEvent> message = MessageBuilder.withPayload(xxxDto).build();
  4. String dest = String.format("%s:%s",topic-name","tag-name");
  5. TransactionSendResult transactionSendResult = this.rocketMQTemplate.sendMessageInTransaction("poducer-name","topic-name:tag-name",message,"xxxid");
  6. if (LocalTransactionState.ROLLBACK_MESSAGE.equals(transactionSendResult.getLocalTransactionState())){
  7. throw new RuntimeException("事务消息投递失败");
  8. }
  9. //按照RocketMQ的写法，这个地方不应该有别的代码
  1. @RocketMQTransactionListener(txProducerGroup = "producer")
  2. class TransactionListenerImpl implements RocketMQLocalTransactionListener {
  3.  
  4. //消息投递成功后执行的逻辑(半消息)
  5. //原文:When send transactional prepare(half) message succeed, this method will be invoked to execute local transaction.
  6. @Override
  7. public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
  8. try{
  9. //
  10. xxxService.doSomething();
  11. return RocketMQLocalTransactionState.COMMIT;
  12. catch(IOException e){
  13. //不确定最终是否成功
  14. return RocketMQLocalTransactionState.UNKNOWN;
  15. }catch(Exception e){
  16. return RocketMQLocalTransactionState.ROLLBACK;
  17. }
  18. }
  19. //回查事务执行状态
  20. @Override
  21. public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
  22. Boolean result = xxxService.isSuccess(msg,arg);
  23. if(result != null){
  24. if(result){
  25. return RocketMQLocalTransactionState.COMMIT;
  26. }else{
  27. return RocketMQLocalTransactionState.ROLLBACK;
  28. }
  29. }
  30. return RocketMQLocalTransactionState.UNKNOWN;
  31. }
  32. }

处理消息（消费）

普通消息和事务消息的区别只在投递的时候才明显，对应的消费端代码比较简单

  1. import lombok.extern.slf4j.Slf4j;
  2. import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
  3. import org.apache.rocketmq.spring.core.RocketMQListener;
  4. import org.springframework.beans.factory.annotation.Autowired;
  5. import org.springframework.data.redis.core.RedisTemplate;
  6. import org.springframework.data.redis.core.StringRedisTemplate;
  7. import org.springframework.stereotype.Component;
  8. @Slf4j
  9. @Component
  10. @RocketMQMessageListener(consumerGroup = "xxx-consumer", topic = "topic-name",selectorExpression = "tag-name")
  11. public class XXXEventMQListener implements RocketMQListener<XXXEvent> {
  12. private String repeatCheckRedisKeyTemplate = "topic-name:tag:repeat-check:%s";
  13. @Autowired private StringRedisTemplate redisTemplate;
  14. @Override
  15. public void onMessage(XXXEvent message) {
  16. log.info("consumer message {}",message);
  17. //处理消息
  18. try{
  19. xxxService.doSomething(message);
  20. }catch(Exception ex){
  21. log.warn(String.format("message [%s] 消费失败",message),ex);
  22. //抛出异常后，MQClient会返回ConsumeConcurrentlyStatus.RECONSUME_LATER,这条消息会再次尝试消费
  23. throw new RuntimException(ex);
  24. }
  25. }
  26. }

RocketMQ用ACK机制保证NameServer知道消息是否被消费在

org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer里是这么处理的

  1. public class DefaultMessageListenerConcurrently implements MessageListenerConcurrently {
  2. @SuppressWarnings("unchecked")
  3. @Override
  4. public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
  5. for (MessageExt messageExt : msgs) {
  6. log.debug("received msg: {}", messageExt);
  7. try {
  8. long now = System.currentTimeMillis();
  9. rocketMQListener.onMessage(doConvertMessage(messageExt));
  10. long costTime = System.currentTimeMillis() - now;
  11. log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
  12. } catch (Exception e) {
  13. log.warn("consume message failed. messageExt:{}", messageExt, e);
  14. context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume);
  15. return ConsumeConcurrentlyStatus.RECONSUME_LATER;
  16. }
  17. }
  18. return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  19. }
  20. }

以上为个人经验，希望能给大家一个参考，也希望大家多多支持我们。

原文链接：https://blog.csdn.net/youxijishu/article/details/105042136

标签：

相关文章