SpringBoot整合RocketMQ

这篇具有很好参考价值的文章主要介绍了SpringBoot整合RocketMQ。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

目录

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)

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

c.本地创建四个springboot的项目(创建项目传送门:https://blog.csdn.net/qq_26154077/article/details/105458395)

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

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

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

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

SpringBoot整合RocketMQ

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/> <!-- lookup parent from repository -->
    </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>
        <!--springboot-->
        <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>
        <!--lombok-->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.4</version>
            <scope>provided</scope>
        </dependency>
        <!--rocketmq-->
        <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.触发接口

SpringBoot整合RocketMQ

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

SpringBoot整合RocketMQ

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获取消息

SpringBoot整合RocketMQ

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

2.1消息生产者(mqproductservice服务)

   SpringBoot整合RocketMQ

(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服务)

SpringBoot整合RocketMQ

(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://www.cnblogs.com/zhixie/p/13813044.html文章来源地址https://www.toymoban.com/news/detail-466850.html

到了这里,关于SpringBoot整合RocketMQ的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • SpringBoot3.0整合RocketMQ时出现未能加载bean文件

    问题 APPLICATION FAILED TO START Description: Field rocketMQTemplate in com.spt.message.service.MqProducerService required a bean of type ‘org.apache.rocketmq.spring.core.RocketMQTemplate’ that could not be found. The injection point has the following annotations: - @org.springframework.beans.factory.annotation.Autowired(required=true) Action: Consider

    2024年02月12日
    浏览(36)
  • SpringBoot整合消息中间件(ActiveMQ,RabbitMQ,RocketMQ,Kafka)

    消息的发送方:生产者 消息的接收方:消费者 同步消息:发送方发送消息到接收方,接收方有所回应后才能够进行下一次的消息发送 异步消息:不需要接收方回应就可以进行下一步的发送 什么是消息队列? 当此时有很多个用户同时访问服务器,需要服务器进行操作,但此

    2024年04月27日
    浏览(48)
  • springboot整合rocketmq:一个消费者组怎么订阅多个topic

            一个消费者组中的所有消费者订阅关系,可以多个topic,多个tag,但是必须一致,否则就倒沫子了,如下图:  下面贴了如下结构的代码  一个消费组(消费者)订阅多个topic的代码(只写了一个消费组的,其他类似): 结果:

    2024年02月15日
    浏览(50)
  • ElasticSearch系列 - SpringBoot整合ES:查询条件 query 和过滤条件 filter 的区别

    01. Elasticsearch 查询条件和过滤条件的区别? Elasticsearch中的查询条件和过滤条件都是用于搜索和过滤文档的条件,但它们之间有一些区别。 查询条件是用于计算文档相关度得分的条件,它会将所有符合条件的文档按照相关度得分从高到低排序,并返回前N个文档。查询条件可以

    2024年02月14日
    浏览(57)
  • 2.搭建Fabric区块链网络环境——前提条件和fabric的安装

    这些前提条件的满足确保了你可以顺利地搭建和运行 Fabric 区块链网络,并进行链码的开发、部署和执行。 安装 Docker:确保系统上已经安装了 Docker,并且 Docker 服务正在运行。 Docker:Fabric 使用 Docker 容器化技术来部署和管理区块链网络的各个组件。Docker 提供了一个轻量级、

    2024年02月11日
    浏览(53)
  • SpringBoot整合Elasticsearch实现分页条件查询及注意事项

    项目环境: springboot 2.3.7.RELEASE es 6.8.3 这里需要注意es中日期格式,ES默认是不支持yyyy-MM-dd HH:mm:ss格式的,需要通过 @Field(type = FieldType.Date, format = DateFormat.custom,pattern = \\\"yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_second\\\") 来指定日期格式。 直接看业务层实现分页条件查询: 范围查询: es en

    2023年04月16日
    浏览(49)
  • ElasticSearch系列 - SpringBoot整合ES:组合多个查询条件 bool 查询

    01. ElasticSearch 布尔查询是什么? 在实际应用中,我们很有可能会查询多个值或字段。 一个 bool 查询由三部分组成: must:所有的语句都必须(must) 匹配,与 AND 等价。 must_not:所有的语句都不能(must not)匹配,与 NOT 等价。 should:至少有一个语句要匹配,与 OR 等价。 02.

    2023年04月08日
    浏览(72)
  • 【100个 Unity实用技能】☀️ | 关于触发器互相检测的必要前提条件配置

    老规矩,先介绍一下 Unity 的科普小知识: Unity 是 实时3D互动内容创作和运营平台 。 包括 游戏开发 、 美术 、 建筑 、 汽车设计 、 影视 在内的所有创作者,借助 Unity 将创意变成现实。 Unity 平台提供一整套完善的软件解决方案,可用于创作、运营和变现任何实时互动的2D和

    2023年04月09日
    浏览(71)
  • java springboot整合MyBatis实现分页查询以及带条件的分页查询

    之前的文章 java springboot整合MyBatis做数据库查询操作操作了springboot整合MyBatis,然后简单做了个按id查询的操作 那么 我们按上文搭建起的环境继续 我们直接在staffDao接口中声明一个分页函数 这里 我们直接在 sql语句中写limit 分页逻辑 参数是方法接收的 这个函数接收两个参数

    2024年02月10日
    浏览(45)
  • 电脑C盘空间不够,直接物理扩容。有前提条件,扩容的空间需要在同一个硬盘上

    扩容前提条件: ①扩容的空间需要在同一个硬盘上,比如你电脑有固态硬盘和机械硬盘,你不能把机械硬盘的空间放在和固态同一个C盘中 ②能扩容的是连续的磁盘空间 因为磁盘管理是这样的,C盘的扩盘只能从连续的空间扩大,也就是说D盘要删了才能扩大C盘。扩大的简单逻

    2024年02月09日
    浏览(65)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包