快捷導(dǎo)航

在SpringBoot中利用RocketMQ實現(xiàn)批量消息消費功能

更新時間：2024年11月07日 11:53:31 作者：魔道不誤砍柴功

RocketMQ 是一款分布式消息隊列,支持高吞吐、低延遲的消息傳遞,對于需要一次處理多條消息的場景,RocketMQ 提供了批量消費的機制,這篇文章將展示如何在 Spring Boot 中實現(xiàn)這一功能,感興趣的小伙伴跟著小編一起來看看吧

準(zhǔn)備工作

在開始之前，請確保你已經(jīng)安裝和配置好 RocketMQ。如果還沒安裝，請參考 RocketMQ 官網(wǎng) 獲取安裝指南。

項目依賴

首先，我們需要在 Spring Boot 項目中添加 RocketMQ 的依賴。打開 pom.xml 文件，添加以下內(nèi)容：

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

這個依賴包包含了與 RocketMQ 集成所需的所有內(nèi)容。

配置 RocketMQ

在 application.yml 文件中添加 RocketMQ 的相關(guān)配置：

rocketmq:
  name-server: 127.0.0.1:9876
  consumer:
    group: batchConsumerGroup
  producer:
    group: batchProducerGroup

name-server：RocketMQ 服務(wù)的地址
consumer.group：消息消費的分組
producer.group：消息生產(chǎn)的分組

確保 name-server 地址是正確的，指向你的 RocketMQ 服務(wù)。

生產(chǎn)批量消息

創(chuàng)建一個消息生產(chǎn)者，用于發(fā)送批量消息。以下是 BatchProducer.java 的示例代碼：

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;

@Service
public class BatchProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void sendBatchMessages() {
        List<Message<String>> messages = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            Message<String> message = MessageBuilder.withPayload("Hello RocketMQ " + i).build();
            messages.add(message);
        }
        
        rocketMQTemplate.syncSend("BatchTopic", messages, 10000);
        System.out.println("批量消息發(fā)送成功！");
    }
}

這里，我們創(chuàng)建了 10 條消息并將它們添加到列表 messages 中。
調(diào)用 rocketMQTemplate.syncSend 方法將消息批量發(fā)送到主題 BatchTopic。

消費批量消息

接下來，我們創(chuàng)建一個消息消費者，用于批量消費消息。以下是 BatchConsumer.java 的示例代碼：

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;

import java.util.List;

@Service
@RocketMQMessageListener(topic = "BatchTopic", consumerGroup = "batchConsumerGroup", selectorExpression = "*", consumeMessageBatchMaxSize = 10)
public class BatchConsumer implements RocketMQListener<List<String>> {

    @Override
    public void onMessage(List<String> messages) {
        System.out.println("批量接收到消息：");
        messages.forEach(message -> System.out.println("消息內(nèi)容：" + message));
    }
}

在這段代碼中：

@RocketMQMessageListener 注解用于標(biāo)識這是一個 RocketMQ 的消息監(jiān)聽器，指定了監(jiān)聽的主題 BatchTopic 和消費分組 batchConsumerGroup。
consumeMessageBatchMaxSize = 10 表示每次批量消費最多 10 條消息。
onMessage 方法會處理接收到的消息列表，并逐條打印出消息內(nèi)容。

測試批量消息發(fā)送和消費

創(chuàng)建一個簡單的 Spring Boot 控制器，用于觸發(fā)批量消息發(fā)送。以下是 MessageController.java 的代碼：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class MessageController {

    @Autowired
    private BatchProducer batchProducer;

    @GetMapping("/sendBatchMessages")
    public String sendBatchMessages() {
        batchProducer.sendBatchMessages();
        return "批量消息已發(fā)送";
    }
}

通過訪問 http://localhost:8080/sendBatchMessages 觸發(fā)消息發(fā)送。