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);
这种投递方式能保证投递成功的消息不会丢失,但是不能保证投递一定成功。假设一次调用的流程是这样的
如果在步骤3的时候发生错误,因为出错mqClient会认为消息投递失败而把事务回滚。如果消息已经被消费,那就会导致业务错误。我们可以用事务消息解决这个问题。
以带事务方式投递的消息,正常情况下的处理流程是这样的
出错的时候是这样的
由于普通消息没有消息回查,普通消息用的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;
- }
- }
以上为个人经验,希望能给大家一个参考,也希望大家多多支持我们。
原文链接:https://blog.csdn.net/youxijishu/article/details/105042136