springboot：整合rocketmq-6ren

springboot：整合rocketmq

转载作者：知者更新时间：2024-03-13 06:33:19

springboot：整合rocketmq

一、简易消息操作

生产者整合mq

导入依赖

<dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.1.1</version>
        </dependency>
    </dependencies>

application.properties

server.port=8080
rocketmq.name-server=127.0.0.1:9876
rocketmq.producer.group=my-producer

生产者发送消息

@RestController
@RequestMapping("/mqMessageController")
public class MqMessageController {
    private static final Logger log = LoggerFactory.getLogger(MqMessageController.class);
 
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
 
    @RequestMapping("/pushMessage.action")
    public String get(@RequestParam("id") int id) {
        rocketMQTemplate.convertAndSend("first-topic","你好,Java旅途" + id);
        return "OK";
    }
 
}

测试生产者服务和RocketMQ通讯

创建消费者服务

pom和生产者一样引用RocketMQ
application.properties配置文件中配置消费者信息(my-producer)
创建消费者监听类

package com.hl.mqconsumer.consumer;

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")
public class Consumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println(message);
    }
}

测试

二、对各种不同类型的消息发送

2.1 消息生产者(mq-product)

配置文件

server.port=8080
#nameservice服务器地址（多个以英文逗号隔开）
rocketmq.name-server=127.0.0.1:9876
#组名
rocketmq.producer.group=hl-producer-group
#主题
rocketmq.producer.topic=hl-topic
#sync tag(同步消息tag)
rocketmq.producer.sync-tag=hl-sync-tags
#async tag(异步消息tag)
rocketmq.producer.async-tag=hl-async-tags
#oneway tag(单向消息tag)
rocketmq.producer.oneway-tag=hl-oneway-tags

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

OrderStep.java

package com.hl.mqproduct.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.hl.mqproduct.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.hl.mqproduct.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;
    }
}

SendCallbackListener.java

package com.hl.mqproduct.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.hl.mqproduct.listener;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
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;
 
    }
}

MqMessageController.java

package com.hl.mqproduct.controller;

import com.alibaba.fastjson.JSONObject;
import com.hl.mqproduct.listener.SendCallbackListener;
import com.hl.mqproduct.model.OrderStep;
import com.hl.mqproduct.model.ResponseMsg;
import com.hl.mqproduct.util.ListSplitter;
import lombok.extern.slf4j.Slf4j;
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.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;

@Slf4j
@RestController
@RequestMapping("/mqMessageController")
public class MqMessageController {

    @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;
    }
}

2.2 消息消费者（mq-consumer）

配置文件

server.port=8081
#nameservice服务器地址（多个以英文逗号隔开）
rocketmq.name-server=127.0.0.1:9876
#组名
rocketmq.consumer.group=hl-consumer-group
#监听主题
rocketmq.consumer.topic=hl-topic
#tags（监听多个tag时使用 || 进行分割，如果监听所有使用*或者不填）
rocketmq.consumer.tags=hl-sync-tags||hl-async-tags||hl-oneway-tags

监听接口配置

package com.hl.mqconsumer.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 消息消费者（mq-consumer-02）

配置文件

server.port=8082
rocketmq.name-server=127.0.0.1:9876
rocketmq.consumer.group=hl-consumer-group
rocketmq.consumer.topic=hl-topic
#tags（监听多个tag时使用 || 进行分割，如果监听所有使用*或者不填）,支持sql过滤
#tags: anran-sync-tags||anran-async-tags||anran-oneway-tags
rocketmq.consumer.tags=money > 5

监听接口配置

package com.hl.mqconsumer02.controller;
 
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)：不用和生产者group相同(在rocketmq中消费者和发送者组没有关系)
 * 监听topic(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 消息消费者（mq-consumer-03）

配置文件

server.port=8083
rocketmq.name-server=127.0.0.1:9876
rocketmq.consumer.group=hl-consumer-group
rocketmq.consumer.topic=hl-topic
rocketmq.consumer.tags=hl-sync-tags||hl-async-tags||hl-oneway-tags

监听接口配置

package com.hl.mqconsumer03.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();
        }
    }
}

注意：
1.如果两个消费者group和topic都一样，则二者轮循接收消息
2.如果两个消费者topic一样，而group不一样，则消息变成广播机制

文章推荐： Netty 框架学习 —— 预置的 ChannelHandler 和编解码器

文章推荐： java安装环境配置

文章推荐：入门ClickHouse和Elasticsearch

文章推荐： JVM第六卷---类加载机制

RocketMQ 详解系列
什么是RocketMQ RocketMQ作为一款纯java、分布式、队列模型的开源消息中间件，支持事务消息、顺序消息、批量消息、定时消息、消息回溯等。主要功能是异步解耦和流量削峰：。常见的MQ主要有
RocketMQ 面试题
1、说说你们公司线上生产环境用的是什么消息中间件? 见【2、多个mq如何选型？】 2、多个mq如何选型？ MQ 描述 Rabb
【RocketMQ】消息的消费总结
消费者从Broker拉取到消息之后，会将消息提交到线程池中进行消费，RocketMQ消息消费是批量进行的，如果一批消息的个数小于预先设置的批量消费大小，直接构建消费请求 ConsumeReques
【RocketMQ】顺序消息实现总结
全局有序在RocketMQ中，如果使消息全局有序，可以为Topic设置一个消息队列，使用一个生产者单线程发送数据，消费者端也使用单线程进行消费，从而保证消息的全局有序，但是这种方式效率低，一
【RocketMQ】主从同步实现原理
RocketMQ支持集群部署来保证高可用。它基于主从模式，将节点分为Master、Slave两个角色，集群中可以有多个Master节点，一个Master节点可以有多个Slave节点。Master节点
【RocketMQ】DLedger模式下的选主流程分析
RocketMQ 4.5版本之前，可以采用主从架构进行集群部署，但是如果master节点挂掉，不能自动在集群中选举出新的Master节点，需要人工介入，在4.5版本之后提供了DLedger模式，使用
【RocketMQ】MQ消息发送总结
RocketMQ是通过 DefaultMQProducer 进行消息发送的，它实现了 MQProducer 接口， MQProducer 接口中定义了消息发送的方法，方法主要分为三大类：
【RocketMQ】消息的存储总结
当Broker收到生产者的消息发送请求时，会对请求进行处理，从请求中解析发送的消息数据，接下来以单个消息的接收为例，看一下消息的接收过程。数据校验封装消息首先Broker会创
【RocketMQ】消息的拉取总结
在上一讲中，介绍了消息的存储，生产者向Broker发送消息之后，数据会写入到CommitLog中，这一讲，就来看一下消费者是如何从Broker拉取消息的。 RocketMQ消息的消费以组为单位
【RocketMQ】NameServer总结
NameServer是一个注册中心，提供服务注册和服务发现的功能。NameServer可以集群部署，集群中每个节点都是对等的关系（没有像ZooKeeper那样在集群中选举出一个Master节点），节
【RocketMQ】DLedger选主源码分析
RocketMQ 4.5版本之前，可以采用主从架构进行集群部署，但是如果master节点挂掉，不能自动在集群中选举出新的Master节点，需要人工介入，在4.5版本之后提供了DLedger模式，使用
【RocketMQ】Dledger日志复制源码分析
消息存储在【RocketMQ】消息的存储一文中提到，Broker收到消息后会调用 CommitLog 的asyncPutMessage方法写入消息，在DLedger模式下使用的是
【RocketMQ】负载均衡源码分析
RocketMQ在集群模式下，同一个消费组内，一个消息队列同一时间只能分配给组内的某一个消费者，也就是一条消息只能被组内的一个消费者进行消费，为了合理的对消息队列进行分配，于是就有了负载均衡.
【RocketMQ】消息拉模式分析
RocketMQ有两种获取消息的方式，分别为推模式和拉模式。推模式推模式在【RocketMQ】消息的拉取一文中已经讲过，虽然从名字上看起来是消息到达Broker后推送给消费者
【RocketMQ】主从同步实现原理
主从同步的实现逻辑主要在 HAService 中，在 DefaultMessageStore 的构造函数中，对 HAService 进行了实例化，并在start方法中，启动了 HAService :
【RocketMQ】主从模式下的消费进度管理
在【RocketMQ】消息的拉取一文中可知，消费者在启动的时候，会创建消息拉取API对象 PullAPIWrapper ，调用pullKernelImpl方法向Broker发送拉取消息的
【RocketMQ】顺序消息实现原理
全局有序在RocketMQ中，如果使消息全局有序，可以为Topic设置一个消息队列，使用一个生产者单线程发送数据，消费者端也使用单线程进行消费，从而保证消息的全局有序，但是这种方式效率低，一
RocketMQ-延迟消息的处理流程介绍
概述 RocketMQ 支持发送延迟消息，但不支持任意时间的延迟消息的设置，仅支持内置预设值的延迟时间间隔的延迟消息；预设值的延迟时间间隔为： 1s、 5s、 10s、 30s、 1m
RocketMQ 延时级别配置方式
RocketMQ 支持定时消息，但是不支持任意时间精度，仅支持特定的 level，例如定时 5s， 10s， 1m 等。其中，level=0 级表示不延时，level=1 表示 1 级延时，le
apache rocketMQ 代理没有启动
我尝试给 rockerMQ broker 加注星标，但我收到了错误消息: There is insufficient memory for the Java Runtime Environment t

知者

个人简介

我是一名优秀的程序员,十分优秀！

作者热门文章

滴滴打车优惠券免费领取

全站热门文章

首页

博学

6Ren·AI

商城

springboot：整合rocketmq

springboot：整合rocketmq

一、简易消息操作

生产者整合mq

创建消费者服务

二、对各种不同类型的消息发送

2.1 消息生产者(mq-product)

配置文件

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

2.2 消息消费者（mq-consumer）

配置文件

监听接口配置

2.3 消息消费者（mq-consumer-02）

配置文件

监听接口配置

2.4 消息消费者（mq-consumer-03）

配置文件

监听接口配置