RocketMQ结合Spring配置实现

忘是亡心i 2022-04-23 03:22 309阅读 0赞

> ### RocketMQ结合Spring配置实现 ###

**【maven依赖】 ：**

<dependency>
          <groupId>com.alibaba.rocketmq</groupId>
          <artifactId>rocketmq-client</artifactId>
          <version>3.2.6</version>
        </dependency>
        <dependency>
          <groupId>com.alibaba.rocketmq</groupId>
          <artifactId>rocketmq-all</artifactId>
          <version>3.2.6</version>
          <type>pom</type>
        </dependency>

**【配置文件 spring-rocketmq.xml 】 ：**

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
           xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
        
        <bean id="messageListeners" class="com.caox.rocketmq.listener.MessageListenerImpl"></bean>
        
    
        
        <bean id="rocketmqConsumer" class="com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer" init-method="start" destroy-method="shutdown">
            <property name="consumerGroup" value="${rocketmq.consumerGroup}"/>
            <property name="instanceName">
                
                <bean class="org.springframework.beans.factory.config.MethodInvokingFactoryBean">
                    <property name="targetClass">
                        <value>com.caox.rocketmq.util.RunTimeUtil</value>
                    </property>
                    <property name="targetMethod">
                        
                        <value>getRocketMqUniqueInstanceName</value>
                    </property>
                </bean>
            </property>
            <property name="namesrvAddr" value="${rocketmq.namesrvAddr}"/>
            <property name="messageListener" ref="messageListeners"/>
            <property name="subscription">
                <map>
                    <entry key="${rocketmq.topic}" value="${rocketmq.tags}" />
                </map>
            </property>
        </bean>
    
        
        <bean id="rocketMQProducer" class="com.caox.rocketmq.producer.RocketMQProducer" init-method="init" destroy-method="destroy">
            <property name="producerGroup" value="${rocketmq.producer.group}"/>
            <property name="instanceName">
                
                <bean class="org.springframework.beans.factory.config.MethodInvokingFactoryBean">
                    <property name="targetClass">
                        <value>com.caox.rocketmq.util.RunTimeUtil</value>
                    </property>
                    <property name="targetMethod">
                        
                        <value>getRocketMqUniqueInstanceName</value>
                    </property>
                </bean>
            </property>
            <property name="namesrvAddr" value="${rocketmq.namesrvAddr}"/>
            
            <property name="retryTimes" value="${rocketmq.producer.retryTimes}" />
        </bean>
    </beans>

**【配置文件 rocketmq.properties 】 ：**

rocketmq.topic=test_rocketmq_topic2
    rocketmq.namesrvAddr=127.0.0.1:9876
    rocketmq.consumerGroup=rmq-consumer-group
    rocketmq.tags=rocketmq_tag2
    rocketmq.producer.group=rmq-producer-group
    rocketmq.producer.retryTimes=3

**【消费者-消息监听实现类】：**

package com.caox.rocketmq.listener;
    
    import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import com.alibaba.rocketmq.common.message.MessageExt;
    import lombok.extern.slf4j.Slf4j;
    
    import java.io.UnsupportedEncodingException;
    import java.util.List;
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/1/14 17:18
     */
    @Slf4j
    public class MessageListenerImpl implements MessageListenerConcurrently {
    
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            log.info(Thread.currentThread().getName()
                    + " Receive New Messages: " + msgs.size()+";msg:" + msgs);
    
    
            try{
                for(MessageExt msg : msgs){
                    log.info(">>>>>>"+new String(msg.getBody(),"UTF-8"));
    //                int a = 1 / 0;
                }
    
            }catch (Exception e){
                log.info("call consumeMessage | EXCEPTION :{}",e);
                // 加入重试机制
                if(msgs.get(0).getReconsumeTimes() == 3){
                    // 成功
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }else{
                    // 重试
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
    
    //        for (MessageExt msg : msgs) {
    //            try {
    //                log.info(">>>>>>"+new String(msg.getBody(),"UTF-8"));
    //            } catch (Exception e) {
    //                log.info("call consumeMessage | EXCEPTION :{}",e);
    //
    //            }
    //        }
            // 有异常抛出来，不要全捕获了，这样保证不能消费的消息下次重推，每次重新消费间隔：10s,30s,1m,2m,3m
            // 如果没有异常会认为都成功消费
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    }

【备注】： 后面注释掉的for循环 重试机制有问题，采用未注释段，(1) try...catch放在for循环外边一旦异常跳出整个循环,重试机制限制次数此时起作用；(2)try...catch放在for循环里边一旦异常继续循环，此时属于不全部捕获异常，实现保证不能消费的消息下次重推，持续，每次重新消费间隔：10s,30s,1m,2m,3m

**【生产者】：**

package com.caox.rocketmq.producer;
    
    import com.alibaba.rocketmq.client.exception.MQClientException;
    import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
    import lombok.Getter;
    import lombok.Setter;
    import lombok.ToString;
    import lombok.extern.slf4j.Slf4j;
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/1/14 17:07
     */
    @Setter
    @Getter
    @ToString
    @Slf4j
    public class RocketMQProducer {
        private DefaultMQProducer defaultMQProducer;
        private String producerGroup;
        private String namesrvAddr;
        private String instanceName;
        private int retryTimes;
    
        public void init() throws MQClientException {
            this.defaultMQProducer = new DefaultMQProducer(this.producerGroup);
            defaultMQProducer.setNamesrvAddr(this.namesrvAddr);
            defaultMQProducer.setInstanceName(this.instanceName);
            defaultMQProducer.setRetryTimesWhenSendFailed(this.retryTimes);
            defaultMQProducer.start();
            log.info("rocketMQ初始化生产者完成[producerGroup：" + producerGroup + "，instanceName："+ instanceName +"]");
        }
    
        public void destroy() {
            defaultMQProducer.shutdown();
            log.info("rocketMQ生产者[producerGroup: " + producerGroup + ",instanceName: "+ instanceName +"]已停止");
        }
    
        public DefaultMQProducer getDefaultMQProducer() {
            return defaultMQProducer;
        }
    }

**【服务层RocketService】：**

package com.caox.rocketmq.service;
    
    import com.alibaba.rocketmq.common.message.Message;
    import com.caox.rocketmq.producer.RocketMQProducer;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.stereotype.Component;
    
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/1/14 17:27
     */
    @Component("rocketService")
    @Slf4j
    public class RocketService {
        /**
         * 注入进来
         */
        @Autowired
        @Qualifier("rocketMQProducer")
        private RocketMQProducer rocketMQProducer;
    
        public void send(String topic, String tag, String msg) throws Exception{
            Message message = new Message(topic, tag, msg.getBytes("UTF-8"));
            rocketMQProducer.getDefaultMQProducer().send(message);
        }
    
    }

**【工具类util】： **

package com.caox.rocketmq.util;
    
    import java.lang.management.ManagementFactory;
    import java.lang.management.RuntimeMXBean;
    import java.util.Random;
    import java.util.concurrent.atomic.AtomicInteger;
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/1/14 17:00
     */
    public class RunTimeUtil {
        private static AtomicInteger index = new AtomicInteger();
    
        public RunTimeUtil() {
        }
    
        public static int getPid() {
            String info = getRunTimeInfo();
            int pid = (new Random()).nextInt();
            int index = info.indexOf("@");
            if(index > 0) {
                pid = Integer.parseInt(info.substring(0, index));
            }
    
            return pid;
        }
    
        public static String getRunTimeInfo() {
            RuntimeMXBean runtime = ManagementFactory.getRuntimeMXBean();
            String info = runtime.getName();
            return info;
        }
    
        public static String getRocketMqUniqueInstanceName() {
            return "pid" + getPid() + "_index" + index.incrementAndGet();
        }
    }

**【事务监听（未决事务 服务器回查客户端）生产者】：**

package com.caox.rocketmq.demo.transaction;
    
    import com.alibaba.rocketmq.client.producer.LocalTransactionState;
    import com.alibaba.rocketmq.client.producer.TransactionCheckListener;
    import com.alibaba.rocketmq.common.message.MessageExt;
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/1/15 15:38
     * 未决事务，服务器回查客户端
     */
    public class TransactionCheckListenerImpl implements TransactionCheckListener {
        // private AtomicInteger transactionIndex = new AtomicInteger(0);
        // 在这里，我们可以根据由MQ回传的key去数据库查询，这条数据到底是成功了还是失败了。
        @Override
        public LocalTransactionState checkLocalTransactionState(MessageExt msg) {
            System.out.println("未决事务，服务器回查客户端msg =" + new String(msg.getBody().toString()));
            // return LocalTransactionState.ROLLBACK_MESSAGE;
            return LocalTransactionState.COMMIT_MESSAGE;
            // return LocalTransactionState.UNKNOW;
        }
    }

**【事务监听（执行本地事务）生产者】：**

package com.caox.rocketmq.demo.transaction;
    
    import com.alibaba.rocketmq.client.producer.LocalTransactionExecuter;
    import com.alibaba.rocketmq.client.producer.LocalTransactionState;
    import com.alibaba.rocketmq.common.message.Message;
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/1/15 15:34
     * 执行本地事务
     */
    public class TransactionExecuterImpl implements LocalTransactionExecuter {
        @Override
        public LocalTransactionState executeLocalTransactionBranch(final Message msg, final Object arg) {
            System.out.println("执行本地事务msg = " + new String(msg.getBody()));
            // 接收 arg  在这里，我们可以根据由MQ回传的key去数据库查询，这条数据到底是成功了还是失败了。
            System.out.println("执行本地事务arg = " + arg);
            String tags = msg.getTags();
            if ("transaction2".equals(tags)) {
                System.out.println("======我的操作============，失败了  -进行ROLLBACK");
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
            return LocalTransactionState.COMMIT_MESSAGE;
            // return LocalTransactionState.UNKNOW;
        }
    }

**【顺序消费原理】：**

**producer在发送消息的时候，把消息发到同一个队列（queue）中,消费者注册消息监听器为MessageListenerOrderly，这样就可以保证消费端只有一个线程去消费消息 注意：是把把消息发到同一个队列（queue），不是同一个topic，默认情况下一个topic包括4个queue【说明】：**实现了MessageListenerOrderly表示一个队列只会被一个线程取到，第二个线程无法访问这个队列。

RocketMQ通过**轮询**所有队列的方式来确定消息被发送到哪一个队列（负载均衡策略）。订单号相同的消息会被先后发送到同一个队列中：

// RocketMQ通过MessageQueueSelector中实现的算法来确定消息发送到哪一个队列上
    // RocketMQ默认提供了两种MessageQueueSelector实现：随机/Hash
    // 当然你可以根据业务实现自己的MessageQueueSelector来决定消息按照何种策略发送到消息队列中
    SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
        @Override
        public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
            Integer id = (Integer) arg;
            int index = id % mqs.size();
            return mqs.get(index);
        }
    }, orderId);

在获取到路由信息以后，会根据**`MessageQueueSelector`**实现的算法来选择一个队列，同一个OrderId获取到的肯定是同一个队列。

private SendResult send()  {
        // 获取topic路由信息
        TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
        if (topicPublishInfo != null && topicPublishInfo.ok()) {
            MessageQueue mq = null;
            // 根据我们的算法，选择一个发送队列
            // 这里的arg = orderId
            mq = selector.select(topicPublishInfo.getMessageQueueList(), msg, arg);
            if (mq != null) {
                return this.sendKernelImpl(msg, mq, communicationMode, sendCallback, timeout);
            }
        }
    }

**【事务消费原理】： **TransactionExecuterImpl【见上述描述】

**【RocketMQ事务过程】：**

**（1）准备发送`Prepared消息，`会拿到消息的地址；****`（2）`执行本地事物；（3）发送确认消息（通过第一阶段拿到的地址去访问消息，并修改消息的状态****）**

**       【发送`Prepared消息伪代码过程`】：**

// =============================发送事务消息的一系列准备工作=======================
    // 未决事务，MQ服务器回查客户端
    // 也就是上文所说的，当RocketMQ发现`Prepared消息`时，会根据这个Listener实现的策略来决断事务
    TransactionCheckListener transactionCheckListener = new TransactionCheckListenerImpl();
    // 构造事务消息的生产者
    TransactionMQProducer producer = new TransactionMQProducer("groupName");
    // 设置事务决断处理类
    producer.setTransactionCheckListener(transactionCheckListener);
    // 本地事务的处理逻辑，相当于示例中检查Bob账户并扣钱的逻辑
    TransactionExecuterImpl tranExecuter = new TransactionExecuterImpl();
    producer.start()
    // 构造MSG，省略构造参数
    Message msg = new Message(......);
    // 发送消息
    SendResult sendResult = producer.sendMessageInTransaction(msg, tranExecuter, null);
    producer.shutdown();

**其中事务消息发送过程：（发送确认消息）**

//  ================================事务消息的发送过程======================================
    public TransactionSendResult sendMessageInTransaction(.....)  {
        // 逻辑代码，非实际代码
        // 1.发送消息
        sendResult = this.send(msg);
        // sendResult.getSendStatus() == SEND_OK
        // 2.如果消息发送成功，处理与消息关联的本地事务单元
        LocalTransactionState localTransactionState = tranExecuter.executeLocalTransactionBranch(msg, arg);
        // 3.结束事务
        this.endTransaction(sendResult, localTransactionState, localException);
    }

![webp][]

[【参考文献】：分布式开放消息系统(RocketMQ)的原理与实践][RocketMQ]

[webp]: /images/20220323/61c4424d72a745b3bd58253b10ed281e.png
[RocketMQ]: https://www.jianshu.com/p/453c6e7ff81c