”工欲善其事,必先利其器。“—孔子《论语.录灵公》
首页 > 编程 > Spring Boot中如何使用RocketMQ实现批量消息消费

Spring Boot中如何使用RocketMQ实现批量消息消费

发布于2024-11-07
浏览:265

How to Implement Batch Message Consumption with RocketMQ in Spring Boot

1.添加依赖

首先,将必要的依赖项添加到您的 pom.xml 文件中:



    org.apache.rocketmq
    rocketmq-spring-boot-starter
    2.3.1
    
        
            org.apache.rocketmq
            rocketmq-client
        
    




    org.apache.rocketmq
    rocketmq-client
    5.3.0

2.配置文件bootstrap.yaml

在 bootstrap.yaml 文件中配置您的 RocketMQ 设置:

rocketmq:
  name-server: 192.168.1.1:9876;192.168.1.2:9876;192.168.1.3:9876 # Replace with actual NameServer addresses
  consumer:
    group: consume-group-test
    access-key: access # Configure if ACL is used
    secret-key: secret
    consume-message-batch-max-size: 50  # Max messages per batch
    pull-batch-size: 100  # Max messages pulled from Broker
  topics:
    project: "group-topic-1"
  groups:
    project: "consume-group-1"  # Use different groups for different business processes

3.配置类MqConfigProperties

创建配置类MqConfigProperties:

import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
import lombok.Data;

import java.io.Serializable;

/**
 * RocketMQ Configuration Class
 */
@Data
@Component
@ConfigurationProperties(prefix = "rocketmq")
public class MqConfigProperties implements Serializable {

    private static final long serialVersionUID = 1L;

    @Autowired
    private RocketMQProperties rocketMQProperties;

    private TopicProperties topics;
    private GroupProperties groups;

    /**
     * Topic Configuration Class
     */
    @Data
    public static class TopicProperties implements Serializable {
        private static final long serialVersionUID = 1L;
        private String project;
    }

    /**
     * Consumer Group Configuration Class
     */
    @Data
    public static class GroupProperties implements Serializable {
        private static final long serialVersionUID = 1L;
        private String project;
    }
}

4. 实施消费者守则

创建消费者类UserConsumer:

import com.alibaba.fastjson2.JSONObject;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.springframework.context.ApplicationContext;
import org.springframework.context.SmartLifecycle;
import org.springframework.stereotype.Component;
import lombok.extern.slf4j.Slf4j;

import javax.annotation.Resource;
import java.util.List;

/**
 * Batch Consumer Implementation
 */
@Component
@Slf4j
public class UserConsumer implements SmartLifecycle {

    @Resource
    private MqConfigProperties mqConfigProperties;

    @Resource
    private ApplicationContext applicationContext;

    private volatile boolean running;
    private DefaultMQPushConsumer consumer;

    @Override
    public void start() {
        if (isRunning()) {
            throw new IllegalStateException("Consumer is already running");
        }
        initConsumer();
        setRunning(true);
        log.info("UserConsumer started successfully.");
    }

    @Override
    public void stop() {
        if (isRunning() && consumer != null) {
            consumer.shutdown();
            setRunning(false);
            log.info("UserConsumer stopped.");
        }
    }

    @Override
    public boolean isRunning() {
        return running;
    }

    private void setRunning(boolean running) {
        this.running = running;
    }

    private void initConsumer() {
        String topic = mqConfigProperties.getTopics().getProject();
        String group = mqConfigProperties.getGroups().getProject();
        String nameServer = mqConfigProperties.getRocketMQProperties().getNameServer();
        String accessKey = mqConfigProperties.getRocketMQProperties().getConsumer().getAccessKey();
        String secretKey = mqConfigProperties.getRocketMQProperties().getConsumer().getSecretKey();

        RPCHook rpcHook = RocketMQUtil.getRPCHookByAkSk(applicationContext.getEnvironment(), accessKey, secretKey);
        consumer = rpcHook != null
                ? new DefaultMQPushConsumer(group, rpcHook, new AllocateMessageQueueAveragely())
                : new DefaultMQPushConsumer(group);

        consumer.setNamesrvAddr(nameServer);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.setConsumeMessageBatchMaxSize(100);  // Set the batch size for consumption
        consumer.subscribe(topic, "*");
        consumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext context) {
                log.info("Received {} messages", msgs.size());
                for (MessageExt message : msgs) {
                    String body = new String(message.getBody());
                    log.info("Processing message: {}", body);
                    User user = JSONObject.parseObject(body, User.class);
                    processUser(user);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        log.info("UserConsumer initialized with topic [{}] and group [{}].", topic, group);
    }

    private void processUser(User user) {
        log.info("Processing user with ID: {}", user.getId());
        // Handle user-related business logic
    }
}

5. 生产者示例代码

要生成批量消息,您可以使用以下 UserProducer 类:

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import java.util.ArrayList;
import java.util.List;

public class UserProducer {

    private DefaultMQProducer producer;

    public void sendBatchMessages(List users, String topic) {
        List messages = new ArrayList();
        for (User user : users) {
            messages.add(new Message(topic, JSONObject.toJSONString(user).getBytes()));
        }
        try {
            producer.send(messages);
        } catch (Exception e) {
            log.error("Error sending batch messages", e);
        }
    }
}

6. 额外的优化建议

  • 性能优化:可以调整消费者线程池的大小。默认情况下,它设置为consumeThreadMin = 20和consumeThreadMax = 20。高并发场景下,增大线程池大小可以提升性能。

  • 错误处理:消费失败时,谨慎使用RECONSUME_LATER,避免无限重试循环。根据您的业务需求设置最大重试次数。

  • 租户隔离:不同的业务模块使用不同的组,避免消费错误组的数据。这在生产环境中尤其重要。

版本声明 本文转载于:https://dev.to/wilson_evan_1efa5910f8855/how-to-implement-batch-message-consumption-with-rocketmq-in-spring-boot-6gi?1如有侵犯,请联系[email protected]删除
最新教程 更多>
  • 如何简化PHP中的JSON解析以获取多维阵列?
    如何简化PHP中的JSON解析以获取多维阵列?
    php 试图在PHP中解析JSON数据的JSON可能具有挑战性,尤其是在处理多维数组时。 To simplify the process, it's recommended to parse the JSON as an array rather than an object.To do...
    编程 发布于2025-04-27
  • 如何在Java字符串中有效替换多个子字符串?
    如何在Java字符串中有效替换多个子字符串?
    在java 中有效地替换多个substring,需要在需要替换一个字符串中的多个substring的情况下,很容易求助于重复应用字符串的刺激力量。 However, this can be inefficient for large strings or when working with nu...
    编程 发布于2025-04-27
  • 为什么PYTZ最初显示出意外的时区偏移?
    为什么PYTZ最初显示出意外的时区偏移?
    与pytz 最初从pytz获得特定的偏移。例如,亚洲/hong_kong最初显示一个七个小时37分钟的偏移: 差异源利用本地化将时区分配给日期,使用了适当的时区名称和偏移量。但是,直接使用DateTime构造器分配时区不允许进行正确的调整。 example pytz.timezone(...
    编程 发布于2025-04-27
  • 如何在JavaScript对象中动态设置键?
    如何在JavaScript对象中动态设置键?
    在尝试为JavaScript对象创建动态键时,如何使用此Syntax jsObj['key' i] = 'example' 1;不工作。正确的方法采用方括号: jsobj ['key''i] ='example'1; 在JavaScript中,数组是一...
    编程 发布于2025-04-27
  • 大批
    大批
    [2 数组是对象,因此它们在JS中也具有方法。 切片(开始):在新数组中提取部分数组,而无需突变原始数组。 令ARR = ['a','b','c','d','e']; // USECASE:提取直到索引作...
    编程 发布于2025-04-27
  • 如何在Java中正确显示“ DD/MM/YYYY HH:MM:SS.SS”格式的当前日期和时间?
    如何在Java中正确显示“ DD/MM/YYYY HH:MM:SS.SS”格式的当前日期和时间?
    如何在“ dd/mm/yyyy hh:mm:mm:ss.ss”格式“ gormat 解决方案: args)抛出异常{ 日历cal = calendar.getInstance(); SimpleDateFormat SDF =新的SimpleDateFormat(“...
    编程 发布于2025-04-27
  • 在C#中如何高效重复字符串字符用于缩进?
    在C#中如何高效重复字符串字符用于缩进?
    在基于项目的深度下固定字符串时,重复一个字符串以进行凹痕,很方便有效地有一种有效的方法来返回字符串重复指定的次数的字符串。使用指定的次数。 constructor 这将返回字符串“ -----”。 字符串凹痕= new String(' - ',depth); console.Wr...
    编程 发布于2025-04-27
  • 如何在GO编译器中自定义编译优化?
    如何在GO编译器中自定义编译优化?
    在GO编译器中自定义编译优化 GO中的默认编译过程遵循特定的优化策略。 However, users may need to adjust these optimizations for specific requirements.Optimization Control in Go Compi...
    编程 发布于2025-04-27
  • Async Void vs. Async Task在ASP.NET中:为什么Async Void方法有时会抛出异常?
    Async Void vs. Async Task在ASP.NET中:为什么Async Void方法有时会抛出异常?
    在ASP.NET async void void async void void void void void void void的设计无需返回asynchroncon而无需返回任务对象。他们在执行过程中增加未偿还操作的计数,并在完成后减少。在某些情况下,这种行为可能是有益的,例如未期望或明确...
    编程 发布于2025-04-27
  • Go web应用何时关闭数据库连接?
    Go web应用何时关闭数据库连接?
    在GO Web Applications中管理数据库连接很少,考虑以下简化的web应用程序代码:出现的问题:何时应在DB连接上调用Close()方法?,该特定方案将自动关闭程序时,该程序将在EXITS EXITS EXITS出现时自动关闭。但是,其他考虑因素可能保证手动处理。选项1:隐式关闭终止数...
    编程 发布于2025-04-27
  • 切换到MySQLi后CodeIgniter连接MySQL数据库失败原因
    切换到MySQLi后CodeIgniter连接MySQL数据库失败原因
    Unable to Connect to MySQL Database: Troubleshooting Error MessageWhen attempting to switch from the MySQL driver to the MySQLi driver in CodeIgniter,...
    编程 发布于2025-04-27
  • 将图片浮动到底部右侧并环绕文字的技巧
    将图片浮动到底部右侧并环绕文字的技巧
    在Web设计中围绕在Web设计中,有时可以将图像浮动到页面右下角,从而使文本围绕它缠绕。这可以在有效地展示图像的同时创建一个吸引人的视觉效果。 css位置在右下角,使用css float and clear properties: img { 浮点:对; ...
    编程 发布于2025-04-27
  • 如何避免Go语言切片时的内存泄漏?
    如何避免Go语言切片时的内存泄漏?
    ,a [j:] ...虽然通常有效,但如果使用指针,可能会导致内存泄漏。这是因为原始的备份阵列保持完整,这意味着新切片外部指针引用的任何对象仍然可能占据内存。 copy(a [i:] 对于k,n:= len(a)-j i,len(a); k
    编程 发布于2025-04-27
  • 在PHP中如何高效检测空数组?
    在PHP中如何高效检测空数组?
    在PHP 中检查一个空数组可以通过各种方法在PHP中确定一个空数组。如果需要验证任何数组元素的存在,则PHP的松散键入允许对数组本身进行直接评估:一种更严格的方法涉及使用count()函数: if(count(count($ playerList)=== 0){ //列表为空。 } 对...
    编程 发布于2025-04-27
  • Laravel 数据库:什么是及用途详解
    Laravel 数据库:什么是及用途详解
    laravel 是一个框架,在数据库连接中提供了许多设施。它具有几个高级功能,可以帮助处理最多样化的数据库系统,结合了访问我们应用程序所需信息的便利性。该框架通过其查询构建器(咨询构建器)和ORM(对象相关映射) eloquent 来实现此操作。 查询构建器 表单,以在数据库中创...
    编程 发布于2025-04-27

免责声明: 提供的所有资源部分来自互联网,如果有侵犯您的版权或其他权益,请说明详细缘由并提供版权或权益证明然后发到邮箱:[email protected] 我们会第一时间内为您处理。

Copyright© 2022 湘ICP备2022001581号-3