使用RocketMQ怎么对消息进行处理

这期内容当中小编将会给大家带来有关使用RocketMQ怎么对消息进行处理，文章内容丰富且以专业的角度为大家分析和叙述，阅读完这篇文章希望大家可以有所收获。

创新互联建站成立10多年来，这条路我们正越走越好，积累了技术与客户资源，形成了良好的口碑。为客户提供成都网站制作、成都网站设计、外贸营销网站建设、网站策划、网页设计、域名注册、网络营销、VI设计、网站改版、漏洞修补等服务。网站是否美观、功能强大、用户体验好、性价比高、打开快等等，这些对于网站建设都非常重要，创新互联建站通过对建站技术性的掌握、对创意设计的研究为客户提供一站式互联网解决方案，携手广大客户，共同发展进步。

消息发送（生产者）

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

    org.apache.rocketmq
    rocketmq-spring-boot-starter
    2.0.1

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

    XXXEvent xxxDto = new XXXEvent();
    Message 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 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 {
    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 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;
    }
}

上述就是小编为大家分享的使用RocketMQ怎么对消息进行处理了，如果刚好有类似的疑惑，不妨参照上述分析进行理解。如果想知道更多相关知识，欢迎关注创新互联行业资讯频道。