SpringBoot - 內建RocketMQ實作延遲消息隊列

目錄

• 前言
• • 環境
• 具體實作

前言

RocketMQ是阿裡巴巴在2012年開源的分布式消息中間件，記錄下SpringBoot整合RocketMQ的方式，RocketMQ的安裝可以檢視：Windows下安裝RocketMQ

環境

SpringBoot2.5.3 + RocketMQ4.7.0

具體實作

• pom.xml

<!-- rocketmq -->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.7.0</version>
</dependency>
           

• application.yml

rocketmq:
  producer:
    producer-group: CoisiniProducerGroup
  consumer:
    consumer-group: CoisiniConsumerGroup
  namesrv-addr: 127.0.0.1:9876
           

• MQ生産者

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.util.Objects;

/**
 * @Description MQ生産者
 * @author coisini
 * @date Aug 25, 2021
 * @Version 1.0
 */
@Component
public class ProducerSchedule {

    private DefaultMQProducer producer;

    @Value("${rocketmq.producer.producer-group}")
    private String producerGroup;

    @Value("${rocketmq.namesrv-addr}")
    private String nameSrvAddr;

    public ProducerSchedule() {

    }

    /**
     * 生産者構造
     * @PostConstruct該注解被用來修飾一個非靜态的void（）方法
     * Bean初始化的執行順序：
     * Constructor(構造方法) -> @Autowired(依賴注入) -> @PostConstruct(注釋的方法)
     */
    @PostConstruct
    public void defaultMQProducer() {
        if (Objects.isNull(this.producer)) {
            this.producer = new DefaultMQProducer(this.producerGroup);
            this.producer.setNamesrvAddr(this.nameSrvAddr);
        }

        try {
            this.producer.start();
            System.out.println("Producer start");
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    /**
     * 消息釋出
     * @param topic
     * @param tag
     * @param messageText
     * @return
     */
    public String send(String topic, String messageText){
        Message message = new Message(topic, messageText.getBytes());

        /**
         * 延遲消息級别設定
         * messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
         */
        message.setDelayTimeLevel(4);

        SendResult result = null;
        try {
            result = this.producer.send(message);
            System.out.println("MessageQueue: " + result.getMessageQueue());
            System.out.println("MsgId: " + result.getMsgId());
            System.out.println("SendStatus: " + result.getSendStatus());
        } catch (Exception e) {
            e.printStackTrace();
        }

        return result.getMsgId();
    }
}
           

• MQ消費者

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

/**
 * @Description MQ消費者
 *      CommandLineRunner 初始化預加載資料
 * @author coisini
 * @date Aug 25, 2021
 * @Version 1.0
 */
@Component
public class ConsumerSchedule implements CommandLineRunner {

    @Value("${rocketmq.consumer.consumer-group}")
    private String consumerGroup;

    @Value("${rocketmq.namesrv-addr}")
    private String nameSrvAddr;

    public void messageListener() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(this.consumerGroup);
        consumer.setNamesrvAddr(this.nameSrvAddr);

        /**
         * 訂閱主題
         */
        consumer.subscribe("Topic", "*");

        /**
         * 設定消費消息數
         */
        consumer.setConsumeMessageBatchMaxSize(1);

        /**
         * 注冊消息監聽
         */
        consumer.registerMessageListener((MessageListenerConcurrently) (messages, context) -> {
            for (Message message : messages) {
                System.out.println("監聽到消息：" + new String(message.getBody()));
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        consumer.start();
    }

    @Override
    public void run(String... args) throws Exception {
        this.messageListener();
    }
}
           

• 測試接口

@RestController
@RequestMapping("/test")
public class TestController {

    @Autowired
    private ProducerSchedule producerSchedule;
    
    @GetMapping("/push")
    public void pushMessageToMQ() throws Exception {
        producerSchedule.send("Topic", "Coisini");
    }
}
           

• 接口調用：

• 30s後延遲消息觸發：

- End - ﹀ ﹀ ﹀ 夢想是鹹魚 關注一下吧