RocketMQ与Spring集成（含生产者消费者）

╰+攻爆jí腚メ 2022-06-06 23:37 172阅读 0赞

RocketMQ与Spring集成，采用push方式接收消息。后面有生产者与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>

### 监听处理类 ###

public class MessageListenerImpl implements MessageListenerConcurrently { 
        private static final Logger LOG = LoggerFactory.getLogger(MessageListenerImpl.class);
    
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            LOG.info(Thread.currentThread().getName()
                    + " Receive New Messages: " + msgs.size()+";msg:" + msgs);
    
            for (MessageExt msg : msgs) {
                try {
                    System.out.println(">>>>"+new String(msg.getBody(),"UTF-8"));
                } catch (UnsupportedEncodingException e) {
                    LOG.error(e.toString());
                }
            }
            // 有异常抛出来，不要全捕获了，这样保证不能消费的消息下次重推，每次重新消费间隔：10s,30s,1m,2m,3m
            // 如果没有异常会认为都成功消费
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    }

### Spring配置 ###

<bean id="messageListeners" class="com.xx.service.mq.rocketmq.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.xx.utils.RunTimeUtil</value>
                    </property>
                    <property name="targetMethod">
                        
                        <value>getRocketMqUniqeInstanceName</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>

consumerGroup name不需要同生产者一致，可以自行定义，一般定义为同topic，应该具有唯一性；  
instanceName不需要同生产者一致，可以自行定义，应该具有唯一性。便于将来可以在控制台根据名称搜索到对应服务。

### RunTimeUtil ###

这是个生成唯一id的工具类

import java.lang.management.ManagementFactory;
    import java.lang.management.RuntimeMXBean;
    import java.util.Random;
    import java.util.concurrent.atomic.AtomicInteger;
    
    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 getRocketMqUniqeInstanceName() {
            return "pid" + getPid() + "_index" + index.incrementAndGet();
        }
    }

### 生产者的集成方式 ###

spring配置：

<bean id="rocketMQProducer" class="com.XXX.service.util.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.XXX.utils.RunTimeUtil</value>
                    </property>
                    <property name="targetMethod">
                        
                        <value>getRocketMqUniqeInstanceName</value>
                    </property>
                </bean>
            </property>
            <property name="namesrvAddr" value="${rocketmq.namesrvAddr}"/>
            
            <property name="retryTimes" value="${rocketmq.producer.retryTimes}" />
        </bean>

生产者：

public class RocketMQProducer { 
        private static final Logger logger = LoggerFactory.getLogger(RocketMQProducer.class);
    
        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();
            logger.info("rocketMQ初始化生产者完成[producerGroup：" + producerGroup + "，instanceName："+ instanceName +"]");
        }
    
        public void destroy() {
            defaultMQProducer.shutdown();
            logger.info("rocketMQ生产者[producerGroup: " + producerGroup + ",instanceName: "+ instanceName +"]已停止");
        }
    
        public DefaultMQProducer getDefaultMQProducer() {
            return defaultMQProducer;
        }
    
        public void setProducerGroup(String producerGroup) {
            this.producerGroup = producerGroup;
        }
    
        public void setNamesrvAddr(String namesrvAddr) {
            this.namesrvAddr = namesrvAddr;
        }
    
        public void setInstanceName(String instanceName) {
            this.instanceName = instanceName;
        }
    
        public void setRetryTimes(int retryTimes) {
            this.retryTimes = retryTimes;
        }
    }

调用生产者，发送消息：

// 注入进来
        @Autowired
        @Qualifier("rocketMQProducer")
        private RocketMQProducer rocketMQProducer;
    
        // 发送消息
        for (Message msg : msgs) {
            Message message = new Message(topic, msg.getBytes("UTF-8"));
            rocketMQProducer.getDefaultMQProducer().send(message);
        }