springboot干货——（二十）整合rocketmq

喜欢ヅ旅行 2021-09-26 13:26 330阅读 0赞

本篇主要讲springboot如何整合rocketmq，具体的rocketmq相关知识可以查看我的rocketmq相关专栏：[RocketMQ菜鸟进化系列][RocketMQ]

1.项目结构

![SouthEast][]

2.pom.xml

这边主要是加入了mq的jar包

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

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    	<modelVersion>4.0.0</modelVersion>
    
    	<groupId>com.gwd</groupId>
    	<artifactId>springboot-rocketmq</artifactId>
    	<version>0.0.1-SNAPSHOT</version>
    	<packaging>jar</packaging>
    
    	<name>springboot-rocketmq</name>
    	<description>Demo project for springboot-rocketmq</description>
    
    	<parent>
    		<groupId>org.springframework.boot</groupId>
    		<artifactId>spring-boot-starter-parent</artifactId>
    		<version>2.0.0.RELEASE</version>
    		<relativePath/> 
    	</parent>
    
    	<properties>
    		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
    		<java.version>1.8</java.version>
    	</properties>
    
    	<dependencies>
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter</artifactId>
    		</dependency>
    
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-test</artifactId>
    			<scope>test</scope>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-web</artifactId>
    		</dependency>
    	 	<dependency>
                <groupId>com.alibaba.rocketmq</groupId>
                <artifactId>rocketmq-client</artifactId>
                <version>3.2.6</version>
            </dependency>
    	</dependencies>
    
    	<build>
    		<plugins>
    			<plugin>
    				<groupId>org.springframework.boot</groupId>
    				<artifactId>spring-boot-maven-plugin</artifactId>
    			</plugin>
    		</plugins>
    	</build>
    </project>

3.application.properties

# 消费者的组名
    apache.rocketmq.consumer.PushConsumer=PushConsumer
    # 生产者的组名
    apache.rocketmq.producer.producerGroup=Producer
    # NameServer地址
    apache.rocketmq.namesrvAddr=XXX.14.250.114:9876

4.消息生产者RocketMQProvider

以顺序发送消息为例

package com.gwd.rocketmq;
    
    import java.util.List;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Service;
    import org.springframework.util.StopWatch;
    import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
    import com.alibaba.rocketmq.client.producer.MessageQueueSelector;
    import com.alibaba.rocketmq.client.producer.SendResult;
    import com.alibaba.rocketmq.common.message.Message;
    import com.alibaba.rocketmq.common.message.MessageQueue;
    /** 
    * @FileName RocketMQServer.java
    * @Description:TODO
    * @author JackHisen(gu.weidong)
    * @version V1.0
    * @createtime 2018年3月22日 下午5:47:49 
    * 修改历史：
    * 时间           作者          版本        描述
    *====================================================  
    *
    */
    //@Component
    @Service
    public class RocketMQProvider {
    	/**
         * 生产者的组名
         */
        @Value("${apache.rocketmq.producer.producerGroup}")
        private String producerGroup;
    
        /**
         * NameServer 地址
         */
        @Value("${apache.rocketmq.namesrvAddr}")
        private String namesrvAddr;
        // @PostConstruct //@PostContruct是spring框架的注解，在方法上加该注解会在项目启动的时候执行该方法，也可以理解为在spring容器初始化的时候执行该方法。
        public void defaultMQProducer() {
            //生产者的组名
            DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
            
            //指定NameServer地址，多个地址以 ; 隔开
            producer.setNamesrvAddr(namesrvAddr);
            try {
                /**
                 * Producer对象在使用之前必须要调用start初始化，初始化一次即可
                 * 注意：切记不可以在每次发送消息时，都调用start方法
                 */
                producer.start();
    
                 //创建一个消息实例，包含 topic、tag 和 消息体
                 //如下：topic 为 "TopicTest"，tag 为 "push"
                Message message = new Message("TopicTest", "push", "发送消息----zhisheng-----".getBytes());
    
                StopWatch stop = new StopWatch();
                stop.start();
    
                for (int i = 0; i < 10; i++) {
                    SendResult result = producer.send(message,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);
    					}
    				},1);
                    System.out.println("发送响应：MsgId:" + result.getMsgId() + "，发送状态:" + result.getSendStatus());
                }
                stop.stop();
                System.out.println("----------------发送十条消息耗时：" + stop.getTotalTimeMillis());
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                producer.shutdown();
            }
        }
    }

一般消息是通过轮询所有队列来发送的（负载均衡策略），顺序消息可以根据业务，比如说订单号相同的消息发送到同一个队列，这边在代码中指定了1，1处这个值相同的获取到的队列是同一个队列。

producer.send(message,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);
    					}
    				},1);

5.消息消费者RocketMQConsumer

package com.gwd.rocketmq;
    
    import javax.annotation.PostConstruct;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Service;
    import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
    import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
    import com.alibaba.rocketmq.common.message.MessageExt;
    /** 
    * @FileName RocketMQClient.java
    * @Description:TODO
    * @author JackHisen(gu.weidong)
    * @version V1.0
    * @createtime 2018年3月22日 下午6:58:24 
    * 修改历史：
    * 时间           作者          版本        描述
    *====================================================  
    *
    */
    @Service
    public class RocketMQConsumer {
    	/**
         * 消费者的组名
         */
        @Value("${apache.rocketmq.consumer.PushConsumer}")
        private String consumerGroup;
    
        /**
         * NameServer 地址
         */
        @Value("${apache.rocketmq.namesrvAddr}")
        private String namesrvAddr;
    
       // @PostConstruct //@PostContruct是spring框架的注解，在方法上加该注解会在项目启动的时候执行该方法，也可以理解为在spring容器初始化的时候执行该方法。
        public void defaultMQPushConsumer() {
            //消费者的组名
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
    
            //指定NameServer地址，多个地址以 ; 隔开
            consumer.setNamesrvAddr(namesrvAddr);
            try {
                //订阅PushTopic下Tag为push的消息
                consumer.subscribe("TopicTest", "push");
    
                //设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
                //如果非第一次启动，那么按照上次消费的位置继续消费
                consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
                consumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> {//->为Java8的lambda表达式,就是匿名函数,具体可以参考该文章https://segmentfault.com/q/1010000007518474。
                    try {
                        for (MessageExt messageExt : list) {
    
                            System.out.println("messageExt: " + messageExt);//输出消息内容
    
                            String messageBody = new String(messageExt.getBody());
    
                            System.out.println("消费响应：msgId : " + messageExt.getMsgId() + ",  msgBody : " + messageBody);//输出消息内容
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER; //稍后再试
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消费成功
                });
                consumer.start();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

6.TestController

package com.gwd.controller;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import com.gwd.rocketmq.RocketMQProvider;
    
    /** 
    * @FileName TestController.java
    * @Description:TODO
    * @author JackHisen(gu.weidong)
    * @version V1.0
    * @createtime 2018年3月22日 下午7:07:24 
    * 修改历史：
    * 时间           作者          版本        描述
    *====================================================  
    *
    */
    @RestController
    public class TestController {
    	@Autowired
    	RocketMQProvider rocketMQProvider;
    	@RequestMapping("/testMQ")
    	public String testMq() {
    		rocketMQProvider.defaultMQProducer();
    		return null;
    	}
    }

测试结果

发送响应：MsgId:6A0EFA7200002AB20000000002F1D269，发送状态:SEND_OK
    messageExt: MessageExt [queueId=1, storeSize=149, queueOffset=334, sysFlag=0, bornTimestamp=1521719131994, bornHost=/116.226.223.81:20033, storeTimestamp=1521719133733, storeHost=/106.14.250.114:10930, msgId=6A0EFA7200002AB20000000002F1D269, commitLogOffset=49402473, bodyCRC=476929171, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=TopicTest, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=335, WAIT=true, TAGS=push}, body=29]]
    消费响应：msgId : 6A0EFA7200002AB20000000002F1D269,  msgBody : 发送消息----zhisheng-----
    发送响应：MsgId:6A0EFA7200002AB20000000002F1D2FE，发送状态:SEND_OK
    messageExt: MessageExt [queueId=1, storeSize=149, queueOffset=335, sysFlag=0, bornTimestamp=1521719132002, bornHost=/116.226.223.81:20033, storeTimestamp=1521719133739, storeHost=/106.14.250.114:10930, msgId=6A0EFA7200002AB20000000002F1D2FE, commitLogOffset=49402622, bodyCRC=476929171, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=TopicTest, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=336, WAIT=true, TAGS=push}, body=29]]
    消费响应：msgId : 6A0EFA7200002AB20000000002F1D2FE,  msgBody : 发送消息----zhisheng-----
    发送响应：MsgId:6A0EFA7200002AB20000000002F1D393，发送状态:SEND_OK
    messageExt: MessageExt [queueId=1, storeSize=149, queueOffset=336, sysFlag=0, bornTimestamp=1521719132008, bornHost=/116.226.223.81:20033, storeTimestamp=1521719133745, storeHost=/106.14.250.114:10930, msgId=6A0EFA7200002AB20000000002F1D393, commitLogOffset=49402771, bodyCRC=476929171, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=TopicTest, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=337, WAIT=true, TAGS=push}, body=29]]
    消费响应：msgId : 6A0EFA7200002AB20000000002F1D393,  msgBody : 发送消息----zhisheng-----
    发送响应：MsgId:6A0EFA7200002AB20000000002F1D428，发送状态:SEND_OK
    messageExt: MessageExt [queueId=1, storeSize=149, queueOffset=337, sysFlag=0, bornTimestamp=1521719132014, bornHost=/116.226.223.81:20033, storeTimestamp=1521719133750, storeHost=/106.14.250.114:10930, msgId=6A0EFA7200002AB20000000002F1D428, commitLogOffset=49402920, bodyCRC=476929171, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=TopicTest, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=338, WAIT=true, TAGS=push}, body=29]]
    消费响应：msgId : 6A0EFA7200002AB20000000002F1D428,  msgBody : 发送消息----zhisheng-----
    发送响应：MsgId:6A0EFA7200002AB20000000002F1D4BD，发送状态:SEND_OK
    messageExt: MessageExt [queueId=1, storeSize=149, queueOffset=338, sysFlag=0, bornTimestamp=1521719132019, bornHost=/116.226.223.81:20033, storeTimestamp=1521719133755, storeHost=/106.14.250.114:10930, msgId=6A0EFA7200002AB20000000002F1D4BD, commitLogOffset=49403069, bodyCRC=476929171, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=TopicTest, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=339, WAIT=true, TAGS=push}, body=29]]
    消费响应：msgId : 6A0EFA7200002AB20000000002F1D4BD,  msgBody : 发送消息----zhisheng-----
    发送响应：MsgId:6A0EFA7200002AB20000000002F1D552，发送状态:SEND_OK
    messageExt: MessageExt [queueId=1, storeSize=149, queueOffset=339, sysFlag=0, bornTimestamp=1521719132024, bornHost=/116.226.223.81:20033, storeTimestamp=1521719133761, storeHost=/106.14.250.114:10930, msgId=6A0EFA7200002AB20000000002F1D552, commitLogOffset=49403218, bodyCRC=476929171, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=TopicTest, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=340, WAIT=true, TAGS=push}, body=29]]
    消费响应：msgId : 6A0EFA7200002AB20000000002F1D552,  msgBody : 发送消息----zhisheng-----
    发送响应：MsgId:6A0EFA7200002AB20000000002F1D5E7，发送状态:SEND_OK
    messageExt: MessageExt [queueId=1, storeSize=149, queueOffset=340, sysFlag=0, bornTimestamp=1521719132036, bornHost=/116.226.223.81:20033, storeTimestamp=1521719133773, storeHost=/106.14.250.114:10930, msgId=6A0EFA7200002AB20000000002F1D5E7, commitLogOffset=49403367, bodyCRC=476929171, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=TopicTest, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=341, WAIT=true, TAGS=push}, body=29]]
    消费响应：msgId : 6A0EFA7200002AB20000000002F1D5E7,  msgBody : 发送消息----zhisheng-----
    发送响应：MsgId:6A0EFA7200002AB20000000002F1D67C，发送状态:SEND_OK
    messageExt: MessageExt [queueId=1, storeSize=149, queueOffset=341, sysFlag=0, bornTimestamp=1521719132044, bornHost=/116.226.223.81:20033, storeTimestamp=1521719133781, storeHost=/106.14.250.114:10930, msgId=6A0EFA7200002AB20000000002F1D67C, commitLogOffset=49403516, bodyCRC=476929171, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=TopicTest, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=342, WAIT=true, TAGS=push}, body=29]]
    消费响应：msgId : 6A0EFA7200002AB20000000002F1D67C,  msgBody : 发送消息----zhisheng-----
    发送响应：MsgId:6A0EFA7200002AB20000000002F1D711，发送状态:SEND_OK
    messageExt: MessageExt [queueId=1, storeSize=149, queueOffset=342, sysFlag=0, bornTimestamp=1521719132050, bornHost=/116.226.223.81:20033, storeTimestamp=1521719133788, storeHost=/106.14.250.114:10930, msgId=6A0EFA7200002AB20000000002F1D711, commitLogOffset=49403665, bodyCRC=476929171, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=TopicTest, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=343, WAIT=true, TAGS=push}, body=29]]
    消费响应：msgId : 6A0EFA7200002AB20000000002F1D711,  msgBody : 发送消息----zhisheng-----
    发送响应：MsgId:6A0EFA7200002AB20000000002F1D7A6，发送状态:SEND_OK
    ----------------发送十条消息耗时：79
    messageExt: MessageExt [queueId=1, storeSize=149, queueOffset=343, sysFlag=0, bornTimestamp=1521719132057, bornHost=/116.226.223.81:20033, storeTimestamp=1521719133794, storeHost=/106.14.250.114:10930, msgId=6A0EFA7200002AB20000000002F1D7A6, commitLogOffset=49403814, bodyCRC=476929171, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=TopicTest, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=344, WAIT=true, TAGS=push}, body=29]]
    消费响应：msgId : 6A0EFA7200002AB20000000002F1D7A6,  msgBody : 发送消息----zhisheng-----

[RocketMQ]: https://blog.csdn.net/column/details/23707.html
[SouthEast]: /images/20210923/66f9c0ef213b43a9af1dd44019bf148e.png