文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

如何使用RocketMQTemplate发送带tags的消息

2023-06-20 13:34

关注

这篇文章将为大家详细讲解有关如何使用RocketMQTemplate发送带tags的消息,小编觉得挺实用的,因此分享给大家做个参考,希望大家阅读完这篇文章后可以有所收获。

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发送消息时,调用的方法是:

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

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

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

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

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

使用RocketMQ 处理消息

消息发送(生产者)

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

<dependency>    <groupId>org.apache.rocketmq</groupId>    <artifactId>rocketmq-spring-boot-starter</artifactId>    <version>2.0.1</version></dependency>

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

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

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

如何使用RocketMQTemplate发送带tags的消息

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

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

如何使用RocketMQTemplate发送带tags的消息

出错的时候是这样的

如何使用RocketMQTemplate发送带tags的消息

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

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

处理消息(消费)

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

import lombok.extern.slf4j.Slf4j;import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;import org.apache.rocketmq.spring.core.RocketMQListener;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.data.redis.core.RedisTemplate;import org.springframework.data.redis.core.StringRedisTemplate;import org.springframework.stereotype.Component;@Slf4j@Component@RocketMQMessageListener(consumerGroup = "xxx-consumer", topic = "topic-name",selectorExpression = "tag-name")public class XXXEventMQListener implements RocketMQListener<XXXEvent> {    private  String repeatCheckRedisKeyTemplate = "topic-name:tag:repeat-check:%s";    @Autowired private StringRedisTemplate redisTemplate;    @Override    public void onMessage(XXXEvent message) {        log.info("consumer message {}",message);        //处理消息        try{            xxxService.doSomething(message);        }catch(Exception ex){            log.warn(String.format("message [%s] 消费失败",message),ex);            //抛出异常后,MQClient会返回ConsumeConcurrentlyStatus.RECONSUME_LATER,这条消息会再次尝试消费            throw new RuntimException(ex);        }    }}

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

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

public class DefaultMessageListenerConcurrently implements MessageListenerConcurrently {    @SuppressWarnings("unchecked")    @Override    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {        for (MessageExt messageExt : msgs) {            log.debug("received msg: {}", messageExt);            try {                long now = System.currentTimeMillis();                rocketMQListener.onMessage(doConvertMessage(messageExt));                long costTime = System.currentTimeMillis() - now;                log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime);            } catch (Exception e) {                log.warn("consume message failed. messageExt:{}", messageExt, e);                context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume);                return ConsumeConcurrentlyStatus.RECONSUME_LATER;            }        }        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;    }}

关于“如何使用RocketMQTemplate发送带tags的消息”这篇文章就分享到这里了,希望以上内容可以对大家有一定的帮助,使各位可以学到更多知识,如果觉得文章不错,请把它分享出去让更多的人看到。

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     807人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     351人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     314人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     433人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     221人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯