SpringBoot整合RocketMQ

ゞ浴缸里的玫瑰 2024-03-16 11:45 79阅读 0赞

**目录**

0.前提条件

1.简易消息操作

1.1生产消息服务整合MQ

1.2创建消费者服务

2.各种不同类型的消息发送集合

2.1消息生产者（mqproductservice服务）

（1）配置文件中配置rocket配置（application-dev.yml）

（2）提供不同类型的接口下发不同类型的消息

2.2消息消费者（mqconsumerservice服务）

（1）配置文件中配置rocket配置（application-dev.yml）

（2）监听接口配置

2.3消息消费者（mqconsumer1service服务）

（1）配置文件中配置rocket配置（application-dev.yml）

（2）监听接口配置

2.4消息消费者（mqconsumer2service服务）

（1）配置文件中配置rocket配置（application-dev.yml）

（2）监听接口配置

--------------------

## 0.前提条件 ##

a.本地已经安装RocketMQ(安装传送门：[https://blog.csdn.net/qq\_26154077/article/details/110847184][https_blog.csdn.net_qq_26154077_article_details_110847184])

b.安装RocketMQ的版本需要和springboot的版本一致（否则会报错，本文使用的是 rocketmq 4.5.2 版本，使用默认端口:9876）

c.本地创建四个springboot的项目（创建项目传送门：[https://blog.csdn.net/qq\_26154077/article/details/105458395][https_blog.csdn.net_qq_26154077_article_details_105458395]）

mqproductservice -- 消息生产服务（提供接口用于创建消息到MQ中）

mqconsumerservice -- 处理类型1的消息服务

mqconsumer1service -- 处理类型1的消息服务

mqconsumer2service -- 处理类型2的消息服务

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70][]

## 1.简易消息操作 ##

### 1.1生产消息服务整合MQ ###

（1）pom.xml中引入rocketmq-spring-boot-starter相关jar包（springboot 2.4.2 rocketmq 2.1.1）

<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
        <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>2.4.2</version>
            <relativePath/> 
        </parent>
        <groupId>com.anran.mq</groupId>
        <artifactId>mqproductservice</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <name>mqproductservice</name>
        <description>mq product service</description>
     
        <properties>
            <java.version>1.8</java.version>
        </properties>
     
        <dependencies>
            
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
     
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>
            
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <version>1.18.4</version>
                <scope>provided</scope>
            </dependency>
            
            <dependency>
                <groupId>org.apache.rocketmq</groupId>
                <artifactId>rocketmq-spring-boot-starter</artifactId>
                <version>2.1.1</version>
            </dependency>
        </dependencies>
     
        <build>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                </plugin>
            </plugins>
        </build>
     
    </project>

（2）application-dev.yml配置文件中配置生成者信息(rocket my-producer-group)

server:
      port: 8082
     
    rocketmq:
      name-server: 127.0.0.1:9876
      producer:
        group: my-producer-group
     
    logging:
      file:
        path: /usr/log/mqproductservice/mqproductservice.log
      level:
        root: INFO
        com.anran.projectmanage.mapper: DEBUG

（3）添加接口触发生产者产生消息

package com.anran.mq.mqproductservice.controller;
     
    import com.anran.mq.mqproductservice.model.ResponseMsg;
    import org.apache.rocketmq.spring.core.RocketMQTemplate;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RequestParam;
    import org.springframework.web.bind.annotation.RestController;
     
    @RestController
    @RequestMapping("/mqMessageController")
    public class MqMessageController {
        private static final Logger log = LoggerFactory.getLogger(MqMessageController.class);
     
        @Autowired
        private RocketMQTemplate rocketMQTemplate;
     
        @RequestMapping("/pushMessage.action")
        public ResponseMsg get(@RequestParam("id") int id) {
            rocketMQTemplate.convertAndSend("first-topic","你好,Java旅途" + id);
            ResponseMsg msg = new ResponseMsg();
            msg.setSuccessData(null);
            return msg;
        }
     
    }

（4）.测试生产者服务和RocketMQ通讯

a.触发接口

![20210119223048392.png][]

b.在RocketMQ的web页面查看消息信息

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 1][]

### 1.2创建消费者服务 ###

（1）pom和生产者一样引用RocketMQ(rocketmq-spring-boot-starter 2.1.1)

（2）application-dev.yml配置文件中配置消费者信息(rocket my-producer-group)

（3）创建消费者监听类

package com.anran.mq.mqconsumerservice.consumer;
     
    import lombok.extern.slf4j.Slf4j;
    import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
    import org.apache.rocketmq.spring.core.RocketMQListener;
    import org.springframework.stereotype.Component;
     
    @Component
    @RocketMQMessageListener(topic = "first-topic",consumerGroup = "my-consumer-group")
    @Slf4j
    public class Consumer implements RocketMQListener<String> {
        @Override
        public void onMessage(String message) {
            System.out.println(message);
        }
    }

（3）测试消费者监听MQ获取消息

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 2][]

## 2.各种不同类型的消息发送集合 ##

### 2.1消息生产者（mqproductservice服务） ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 3][]

#### （1）配置文件中配置rocket配置（application-dev.yml） ####

server:
      port: 8082
     
    #rocketmq配置信息
    rocketmq:
      #nameservice服务器地址（多个以英文逗号隔开）
      name-server: 127.0.0.1:9876
      #生产者配置
      producer:
        #组名
        group: anran-producer-group
        #目的地（topic:tag）
        #topic
        topic: anran-topic
        #sync tag(同步消息tag)
        sync-tag: anran-sync-tags
        #async tag(异步消息tag)
        async-tag: anran-async-tags
        #oneway tag(单向消息tag)
        oneway-tag: anran-oneway-tags
     
    logging:
      file:
        path: /usr/log/mqproductservice/mqproductservice.log
      level:
        root: INFO
        com.anran.projectmanage.mapper: DEBUG

#### （2）提供不同类型的接口下发不同类型的消息 ####

MqMessageController.java

package com.anran.mq.mqproductservice.controller;
     
    import com.alibaba.fastjson.JSONObject;
    import com.anran.mq.mqproductservice.listener.SendCallbackListener;
    import com.anran.mq.mqproductservice.model.OrderStep;
    import com.anran.mq.mqproductservice.model.ResponseMsg;
    import com.anran.mq.mqproductservice.util.ListSplitter;
    import org.apache.rocketmq.client.producer.MessageQueueSelector;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.client.producer.SendStatus;
    import org.apache.rocketmq.client.producer.TransactionSendResult;
    import org.apache.rocketmq.common.message.MessageQueue;
    import org.apache.rocketmq.spring.core.RocketMQTemplate;
    import org.apache.rocketmq.spring.support.RocketMQHeaders;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.messaging.Message;
    import org.springframework.messaging.support.MessageBuilder;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RequestParam;
    import org.springframework.web.bind.annotation.RestController;
     
    import java.util.ArrayList;
    import java.util.List;
     
    @RestController
    @RequestMapping("/mqMessageController")
    public class MqMessageController {
        private static final Logger log = LoggerFactory.getLogger(MqMessageController.class);
     
        @Autowired
        private RocketMQTemplate rocketMQTemplate;
     
        @Value(value = "${rocketmq.producer.topic}:${rocketmq.producer.sync-tag}")
        private String syncTag;
     
        @Value(value = "${rocketmq.producer.topic}:${rocketmq.producer.async-tag}")
        private String asyncag;
     
        @Value(value = "${rocketmq.producer.topic}:${rocketmq.producer.oneway-tag}")
        private String onewayTag;
     
        /**
         * rocketmq 同步消息
         *
         * @param id 消息
         * @return 结果
         */
        @RequestMapping("/pushMessage.action")
        public ResponseMsg pushMessage(@RequestParam("id") int id) {
            log.info("pushMessage start : " + id);
            // 构建消息
            String messageStr = "order id : " + id;
            Message<String> message = MessageBuilder.withPayload(messageStr)
                    .setHeader(RocketMQHeaders.KEYS, id)
                    .build();
            // 设置发送地和消息信息并发送同步消息
            SendResult sendResult = rocketMQTemplate.syncSend(syncTag, message);
            log.info("pushMessage finish : " + id + ", sendResult : " + JSONObject.toJSONString(sendResult));
            ResponseMsg msg = new ResponseMsg();
            // 解析发送结果
            if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
                msg.setSuccessData(sendResult.getMsgId() + " : " + sendResult.getSendStatus());
            }
            return msg;
        }
     
        /**
         * 发送异步消息
         *
         * @param id 消息
         * @return 结果
         */
        @RequestMapping("/pushAsyncMessage.action")
        public ResponseMsg pushAsyncMessage(@RequestParam("id") int id) {
            log.info("pushAsyncMessage start : " + id);
            // 构建消息
            String messageStr = "order id : " + id;
            Message<String> message = MessageBuilder.withPayload(messageStr)
                    .setHeader(RocketMQHeaders.KEYS, id)
                    .build();
            // 设置发送地和消息信息并发送异步消息
            rocketMQTemplate.asyncSend(asyncag, message, new SendCallbackListener(id));
            log.info("pushAsyncMessage finish : " + id);
            ResponseMsg msg = new ResponseMsg();
            msg.setSuccessData(null);
            return msg;
        }
     
        /**
         * 发送单向消息（不关注发送结果：记录日志）
         *
         * @param id 消息
         * @return 结果
         */
        @RequestMapping("/pushOneWayMessage.action")
        public ResponseMsg pushOneWayMessage(@RequestParam("id") int id) {
            log.info("pushOneWayMessage start : " + id);
            // 构建消息
            String messageStr = "order id : " + id;
            Message<String> message = MessageBuilder.withPayload(messageStr)
                    .setHeader(RocketMQHeaders.KEYS, id)
                    .build();
            // 设置发送地和消息信息并发送单向消息
            rocketMQTemplate.sendOneWay(onewayTag, message);
            log.info("pushOneWayMessage finish : " + id);
            ResponseMsg msg = new ResponseMsg();
            msg.setSuccessData(null);
            return msg;
        }
     
        /**
         * 发送包含顺序的单向消息
         *
         * @param id 消息
         * @return 结果
         */
        @RequestMapping("/pushSequeueMessage.action")
        public ResponseMsg pushSequeueMessage(@RequestParam("id") int id) {
            log.info("pushSequeueMessage start : " + id);
            // 创建三个不同订单的不同步骤
            for (int i = 0; i < 3; i++) {
                // 处理当前订单唯一标识
                String myId = id + "" + i;
                // 获取当前订单的操作步骤列表
                List<OrderStep> myOrderSteps = OrderStep.buildOrderSteps(myId);
                // 依次操作步骤下发消息队列
                for (OrderStep orderStep : myOrderSteps) {
                    // 构建消息
                    String messageStr = String.format("order id : %s, desc : %s", orderStep.getId(), orderStep.getDesc());
                    Message<String> message = MessageBuilder.withPayload(messageStr)
                            .setHeader(RocketMQHeaders.KEYS, orderStep.getId())
                            .build();
                    // 设置顺序下发
                    rocketMQTemplate.setMessageQueueSelector(new MessageQueueSelector() {
                        /**
                         * 设置放入同一个队列的规则
                         * @param list 消息列表
                         * @param message 当前消息
                         * @param o 比较的关键信息
                         * @return 消息队列
                         */
                        @Override
                        public MessageQueue select(List<MessageQueue> list, org.apache.rocketmq.common.message.Message message, Object o) {
                            // 根据当前消息的id，使用固定算法获取需要下发的队列
                            // （使用当前id和消息队列个数进行取模获取需要下发的队列，id和队列数量一样时，选择的队列坑肯定一样）
                            int queueNum = Integer.valueOf(String.valueOf(o)) % list.size();
                            log.info(String.format("queueNum : %s, message : %s", queueNum, new String(message.getBody())));
                            return list.get(queueNum);
                        }
                    });
                    // 设置发送地和消息信息并发送消息（Orderly）
                    rocketMQTemplate.syncSendOrderly(syncTag, message, orderStep.getId());
                }
            }
            log.info("pushSequeueMessage finish : " + id);
            ResponseMsg msg = new ResponseMsg();
            msg.setSuccessData(null);
            return msg;
        }
     
        /**
         * rocketmq 延迟消息
         *
         * @param id 消息
         * @return 结果
         */
        @RequestMapping("/pushDelayMessage.action")
        public ResponseMsg pushDelayMessage(@RequestParam("id") int id) {
            log.info("pushDelayMessage start : " + id);
            // 构建消息
            String messageStr = "order id : " + id;
            Message<String> message = MessageBuilder.withPayload(messageStr)
                    .setHeader(RocketMQHeaders.KEYS, id)
                    .build();
            // 设置超时和延时推送
            // 超时时针对请求broker然后结果返回给product的耗时
            // 现在RocketMq并不支持任意时间的延时，需要设置几个固定的延时等级，从1s到2h分别对应着等级1到18
            // private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
            SendResult sendResult = rocketMQTemplate.syncSend(syncTag, message, 1 * 1000l, 4);
            log.info("pushDelayMessage finish : " + id + ", sendResult : " + JSONObject.toJSONString(sendResult));
            ResponseMsg msg = new ResponseMsg();
            // 解析发送结果
            if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
                msg.setSuccessData(sendResult.getMsgId() + " : " + sendResult.getSendStatus());
            }
            return msg;
        }
     
        /**
         * 同时发送10个单向消息（真正的批量）
         *
         * @param id 消息
         * @return 结果
         */
        @RequestMapping("/pushBatchMessage.action")
        public ResponseMsg pushBatchMessage(@RequestParam("id") int id) {
            log.info("pushBatchMessage start : " + id);
            // 创建消息集合
            List<Message> messages = new ArrayList<>();
            for (int i = 0; i < 3; i++) {
                String myId = id + "" + i;
                // 处理当前订单唯一标识
                String messageStr = "order id : " + myId;
                Message<String> message = MessageBuilder.withPayload(messageStr)
                        .setHeader(RocketMQHeaders.KEYS, myId)
                        .build();
                messages.add(message);
            }
            // 批量下发消息到broker,不支持消息顺序操作，并且对消息体有大小限制（不超过4M）
            ListSplitter splitter = new ListSplitter(messages, 1024 * 1024 * 4);
            while (splitter.hasNext()) {
                List<Message> listItem = splitter.next();
                rocketMQTemplate.syncSend(syncTag, listItem);
            }
            log.info("pushBatchMessage finish : " + id);
            ResponseMsg msg = new ResponseMsg();
            msg.setSuccessData(null);
            return msg;
        }
     
        /**
         * sql过滤消息
         *
         * @param id 消息
         * @return 结果
         */
        @RequestMapping("/pushSqlMessage.action")
        public ResponseMsg pushSqlMessage(@RequestParam("id") int id) {
            log.info("pushSqlMessage start : " + id);
            // 创建消息集合
            List<Message> messages = new ArrayList<>();
            for (int i = 0; i < 10; i++) {
                String myId = id + "" + i;
                // 处理当前订单唯一标识
                String messageStr = "order id : " + myId;
                Message<String> message = MessageBuilder.withPayload(messageStr)
                        .setHeader(RocketMQHeaders.KEYS, myId)
                        .setHeader("money", i)
                        .build();
                messages.add(message);
            }
            rocketMQTemplate.syncSend(syncTag, messages);
            log.info("pushSqlMessage finish : " + id);
            ResponseMsg msg = new ResponseMsg();
            msg.setSuccessData(null);
            return msg;
        }
     
        /**
         * 事务消息
         *
         * @param id 消息
         * @return 结果
         */
        @RequestMapping("/pushTransactionMessage.action")
        public ResponseMsg pushTransactionMessage(@RequestParam("id") int id) {
            log.info("pushTransactionMessage start : " + id);
            // 创建消息
            String messageStr = "order id : " + id;
            Message<String> message = MessageBuilder.withPayload(messageStr)
                    .setHeader(RocketMQHeaders.KEYS, id)
                    .setHeader("money", 10)
                    .setHeader(RocketMQHeaders.TRANSACTION_ID, id)
                    .build();
            TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction(syncTag, message, null);
            log.info("pushTransactionMessage result : " + JSONObject.toJSONString(transactionSendResult));
            log.info("pushTransactionMessage finish : " + id);
            ResponseMsg msg = new ResponseMsg();
            msg.setSuccessData(null);
            return msg;
        }
    }

SendCallbackListener.java

package com.anran.mq.mqproductservice.listener;
     
    import com.alibaba.fastjson.JSONObject;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.rocketmq.client.producer.SendCallback;
    import org.apache.rocketmq.client.producer.SendResult;
     
    /**
     * rocketmq异步回调监听
     */
    @Slf4j
    public class SendCallbackListener implements SendCallback {
     
        private int id;
     
        public SendCallbackListener(int id) {
            this.id = id;
        }
     
        @Override
        public void onSuccess(SendResult sendResult) {
            log.info("CallBackListener on success : " + JSONObject.toJSONString(sendResult));
        }
     
        @Override
        public void onException(Throwable throwable) {
            log.error("CallBackListener on exception : ", throwable);
        }
    }

TransactionListener.java

package com.anran.mq.mqproductservice.listener;
     
    import com.alibaba.fastjson.JSONObject;
    import lombok.RequiredArgsConstructor;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.rocketmq.client.producer.SendCallback;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
    import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
    import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
    import org.apache.rocketmq.spring.support.RocketMQHeaders;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.messaging.Message;
    import org.springframework.messaging.MessageHeaders;
     
    import java.util.HashMap;
    import java.util.Map;
     
    /**
     * rocketmq异步回调监听
     */
    @Slf4j
    @RocketMQTransactionListener
    @RequiredArgsConstructor(onConstructor = @__(@Autowired))
    public class TransactionListener implements RocketMQLocalTransactionListener {
        private static final Map<String, RocketMQLocalTransactionState> TRANSACTION_STATE_MAP = new HashMap<>();
     
        /**
         * 处理本地事务
         * @param message
         * @param o
         * @return
         */
        @Override
        public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
            log.info("执行本地事务");
            MessageHeaders headers = message.getHeaders();
            //获取事务ID
            String transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);
            TRANSACTION_STATE_MAP.put(transactionId, RocketMQLocalTransactionState.UNKNOWN);
            log.info("transactionId is {}",transactionId);
            try {
                Thread.sleep(10 * 1000l);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            RocketMQLocalTransactionState state = RocketMQLocalTransactionState.ROLLBACK;
            if (Integer.parseInt(transactionId) % 2 == 0) {
                //执行成功，可以提交事务
                state = RocketMQLocalTransactionState.COMMIT;
            }
            log.info("transactionId is {}, state {}",transactionId, state.toString());
            TRANSACTION_STATE_MAP.remove(transactionId);
            return state;
        }
     
        /**
         * 校验事务状态
         * @param message
         * @return
         */
        @Override
        public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
            MessageHeaders headers = message.getHeaders();
            //获取事务ID
            String transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);
            log.info("检查本地事务,事务ID:{}",transactionId);
            RocketMQLocalTransactionState state = TRANSACTION_STATE_MAP.get(transactionId);
            if(null != state){
                return state;
            }
            return RocketMQLocalTransactionState.ROLLBACK;
     
        }
    }

OrderStep.java

package com.anran.mq.mqproductservice.model;
     
    import lombok.Data;
     
    import java.util.ArrayList;
    import java.util.List;
     
    /**
     * 订单消息模型
     */
    @Data
    public class OrderStep {
        /**
         * 订单id
         */
        private String id;
     
        /**
         * 操作步骤
         */
        private String desc;
     
        /**
         * 步骤构建方法
         * @param id 订单唯一标识
         * @param desc 步骤描述
         */
        public OrderStep(String id, String desc) {
            this.id = id;
            this.desc = desc;
        }
     
        /**
         * 构建订单的所有操作步骤：创建、支付、回调
         * @param id 订单唯一标识
         * @return 订单操作步骤列表
         */
        public static List<OrderStep> buildOrderSteps(String id) {
            List<OrderStep> orderSteps = new ArrayList<>();
            OrderStep createStep = new OrderStep(id, "create order");
            orderSteps.add(createStep);
            OrderStep payStep = new OrderStep(id, "pay order");
            orderSteps.add(payStep);
            OrderStep callbackStep = new OrderStep(id, "call back order");
            orderSteps.add(callbackStep);
            return orderSteps;
        }
    }

ResponseMsg.java

package com.anran.mq.mqproductservice.model;
     
    import lombok.Data;
     
    @Data
    public class ResponseMsg {
        public static final int CODE_FAIL = 500;
     
        public static final int CODE_SUCCESS = 200;
     
        public static final String MSG_SUCCESS = "success";
     
        public static final String MSG_FAIL = "fail";
     
        private int code;
     
        private String msg;
     
        private Object data;
     
        public ResponseMsg() {
            this.code = CODE_FAIL;
            this.msg = MSG_FAIL;
        }
     
        public void setSuccessData(Object data) {
            this.code = CODE_SUCCESS;
            this.msg = MSG_SUCCESS;
            this.data = data;
        }
    }

ListSplitter.java

package com.anran.mq.mqproductservice.util;
     
    import java.util.Iterator;
    import java.util.List;
     
    /**
     * LIST对象进行分割
     */
    public class ListSplitter<T> implements Iterator<List<T>> {
        /**
         * 所有内容
         */
        private final List<T> ts;
     
        /**
         * 多少数据进行分割
         */
        private int size;
     
        /**
         * 已经分割的索引
         */
        private int currIndex;
     
        /**
         * 构造方法
         * @param ts 所有内容
         * @param size 分割数量
         */
        public ListSplitter(List<T> ts, int size) {
            this.ts = ts;
            this.size = size;
        }
     
        /**
         * 实现是否还厚后续数据
         * @return true:有 false:无
         */
        @Override
        public boolean hasNext() {
            return currIndex < ts.size();
        }
     
        @Override
        public List<T> next() {
            int nextIndex = currIndex;
            int totalSize = 0;
            for (; nextIndex < ts.size(); nextIndex++) {
                T t = ts.get(nextIndex);
                totalSize = totalSize + t.toString().length();
                if (totalSize > size) {
                    break;
                }
            }
            List<T> subList = ts.subList(currIndex, nextIndex);
            currIndex = nextIndex;
            return subList;
        }
    }

### 2.2消息消费者（mqconsumerservice服务） ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 4][]

#### （1）配置文件中配置rocket配置（application-dev.yml） ####

server:
      port: 8083
     
    #rocketmq配置信息
    rocketmq:
      #nameservice服务器地址（多个以英文逗号隔开）
      name-server: 127.0.0.1:9876
      #消费者配置
      consumer:
        #组名
        group: anran-consumer-group
        #监听主题
        topic: anran-topic
        #tags（监听多个tag时使用 || 进行分割，如果监听所有使用*或者不填）
        tags: anran-sync-tags||anran-async-tags||anran-oneway-tags
     
    logging:
      file:
        path: /usr/log/mqconsumerservice/mqconsumerservice.log
      level:
        root: INFO
        com.anran.projectmanage.mapper: DEBUG

#### （2）监听接口配置 ####

Consumer.java

package com.anran.mq.mqconsumerservice.consumer;
     
    import lombok.extern.slf4j.Slf4j;
    import org.apache.rocketmq.spring.annotation.MessageModel;
    import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
    import org.apache.rocketmq.spring.core.RocketMQListener;
    import org.springframework.stereotype.Component;
     
    /**
     * 设置消息监听
     * 监听组：监听topic：监听tag(默认监听topic下所有)
     * 监听消费模式:默认负载均衡：CLUSTERING（每一个消息只发给一个消费者）、广播模式：BROADCASTING（发送给所有消费者）
     *
     */
    @Component
    @RocketMQMessageListener(consumerGroup = "${rocketmq.consumer.group}", topic = "${rocketmq.consumer.topic}",
            selectorExpression = "${rocketmq.consumer.tags}", messageModel = MessageModel.BROADCASTING)
    @Slf4j
    public class Consumer implements RocketMQListener<String> {
        @Override
        public void onMessage(String message) {
            System.out.println(message);
        }
    }

### 2.3消息消费者（mqconsumer1service服务） ###

#### （1）配置文件中配置rocket配置（application-dev.yml） ####

server:
      port: 8085
     
    #rocketmq配置信息
    rocketmq:
      #nameservice服务器地址（多个以英文逗号隔开）
      name-server: 127.0.0.1:9876
      #消费者配置
      consumer:
        #组名
        group: anran-consumer-group
        #监听主题
        topic: anran-topic
        #tags（监听多个tag时使用 || 进行分割，如果监听所有使用*或者不填）,支持sql过滤
    #    tags: anran-sync-tags||anran-async-tags||anran-oneway-tags
        tags: money > 5
     
    logging:
      file:
        path: /usr/log/mqproductservice/mqproductservice.log
      level:
        root: INFO
        com.anran.projectmanage.mapper: DEBUG

#### （2）监听接口配置 ####

package com.anran.mq.mqconsumer1service.consumer;
     
    import lombok.extern.slf4j.Slf4j;
    import org.apache.rocketmq.spring.annotation.ConsumeMode;
    import org.apache.rocketmq.spring.annotation.MessageModel;
    import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
    import org.apache.rocketmq.spring.annotation.SelectorType;
    import org.apache.rocketmq.spring.core.RocketMQListener;
    import org.springframework.stereotype.Component;
     
    /**
     * 设置消息监听
     * 1.监听组(consumerGroup)：监听topic(topic)：监听tag(selectorExpression)(默认监听topic下所有)
     * 2.监听消费模式(messageModel):默认负载均衡：CLUSTERING（每一个消息只发给一个消费者）、广播模式：BROADCASTING（发送给所有消费者）
     * 3.设置顺序消息处理模式(consumeMode)（默认是所有线程可以处理同一个消息队列（ConsumeMode.CONCURRENTLY）,当前消息没有线程在执行时其他线程才能够执行（ConsumeMode.ORDERLY）。
     *   ps:一个线程顺序执行一个队列表时消息监听必须使用负载均衡messageModel = MessageModel.BROADCASTING）
     * 4.设置过滤tags类型：默认时tag（SelectorType.TAG）,可以修改为sql语法(SelectorType.SQL92)：
     */
    @Component
    @RocketMQMessageListener(consumerGroup = "${rocketmq.consumer.group}", topic = "${rocketmq.consumer.topic}",
            selectorExpression = "${rocketmq.consumer.tags}", messageModel = MessageModel.CLUSTERING,
            consumeMode = ConsumeMode.ORDERLY, selectorType = SelectorType.SQL92)
    @Slf4j
    public class Consumer implements RocketMQListener<String> {
        @Override
        public void onMessage(String message) {
            log.info(message);
        }
    }

### 2.4消息消费者（mqconsumer2service服务） ###

#### （1）配置文件中配置rocket配置（application-dev.yml） ####

server:
      port: 8084
     
    #rocketmq配置信息
    rocketmq:
      #nameservice服务器地址（多个以英文逗号隔开）
      name-server: 127.0.0.1:9876
      #消费者配置
      consumer:
        #组名
        group: anran-consumer-group
        #监听主题
        topic: anran-topic
        #tags（监听多个tag时使用 || 进行分割，如果监听所有使用*或者不填）
        tags: anran-sync-tags||anran-async-tags||anran-oneway-tags
     
    logging:
      file:
        path: /usr/log/mqproductservice/mqproductservice.log
      level:
        root: INFO
        com.anran.projectmanage.mapper: DEBUG

#### （2）监听接口配置 ####

package com.anran.mq.mqconsumer2service.consumer;
     
    import lombok.extern.slf4j.Slf4j;
    import org.apache.rocketmq.spring.annotation.ConsumeMode;
    import org.apache.rocketmq.spring.annotation.MessageModel;
    import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
    import org.apache.rocketmq.spring.core.RocketMQListener;
    import org.springframework.stereotype.Component;
     
    /**
     * 设置消息监听
     * 1.监听组(consumerGroup)：监听topic(topic)：监听tag(selectorExpression)(默认监听topic下所有)
     * 2.监听消费模式(messageModel):默认负载均衡：CLUSTERING（每一个消息只发给一个消费者）、广播模式：BROADCASTING（发送给所有消费者）
     * 3.设置顺序消息处理模式(consumeMode)（默认是所有线程可以处理同一个消息队列（ConsumeMode.CONCURRENTLY）,当前消息没有线程在执行时其他线程才能够执行（ConsumeMode.ORDERLY）。
     *   ps:一个线程顺序执行一个队列表时消息监听必须使用负载均衡messageModel = MessageModel.BROADCASTING）
     */
    @Component
    @RocketMQMessageListener(consumerGroup = "${rocketmq.consumer.group}", topic = "${rocketmq.consumer.topic}",
            selectorExpression = "${rocketmq.consumer.tags}", messageModel = MessageModel.CLUSTERING,
            consumeMode = ConsumeMode.ORDERLY)
    @Slf4j
    public class Consumer implements RocketMQListener<String> {
        @Override
        public void onMessage(String message) {
            log.info(message);
     
            try {
                Thread.sleep(10 * 1000l);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

参考链接：

[https://blog.csdn.net/csj50/article/details/103122131][https_blog.csdn.net_csj50_article_details_103122131]

[https://www.cnblogs.com/zhixie/p/13813044.html][https_www.cnblogs.com_zhixie_p_13813044.html]

[https_blog.csdn.net_qq_26154077_article_details_110847184]: https://blog.csdn.net/qq_26154077/article/details/110847184
[https_blog.csdn.net_qq_26154077_article_details_105458395]: https://blog.csdn.net/qq_26154077/article/details/105458395
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/1043d28cb677455e9d162d759397a91a.png
[20210119223048392.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/3927810ab4674f09b9964203eb7e4394.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/029805370828405f99df46876f65083f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/c03c77fc9d2e46b581d4f6f5978fc82d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 3]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/dbb491cf6f4c48c983fc43bf05016a92.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 4]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/ccf5cba2e0fa4f9ba1a8447b01556e42.png
[https_blog.csdn.net_csj50_article_details_103122131]: https://blog.csdn.net/csj50/article/details/103122131
[https_www.cnblogs.com_zhixie_p_13813044.html]: https://www.cnblogs.com/zhixie/p/13813044.html