前言
批量消费是 RocketMQ 提供的一种消费模式，能够有效提升消费者的处理能力，减少网络 IO 操作，从而提高性能。
在使用 @RocketMQMessageListener 注解时，Spring Boot 默认的消费模式并不支持批量消费。为了在高吞吐量场景下优化消费性能，我们可以通过手动配置 RocketMQ 消费者，来实现批量消息消费。
本文将详细介绍如何从零开始，在 Spring Boot 项目中集成 RocketMQ 的批量消费功能，并提供一些性能优化建议。
接入流程
- 引入依赖
- 配置文件 bootstrap.yaml
- 配置类 MqConfigProperties
- 消费者代码实现
- 生产者示例代码
- 其他优化建议
1. 引入依赖
<!--RocketMQ的Spring Boot依赖，适用于Spring Boot 3版本-->
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.3.1</version>
<exclusions>
<exclusion>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
</exclusion>
</exclusions>
</dependency>
<!-- 兼容MQ集群5.3.0版本的依赖 -->
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>5.3.0</version>
</dependency>
2. 配置文件 bootstrap.yaml
rocketmq:
name-server: 192.168.1.1:9876;192.168.1.2:9876;192.168.1.3:9876 # 替换为实际的NameServer地址
consumer:
group: consume-group-test
access-key: access # 如果使用了ACL，则需要配置
secret-key: secret
consume-message-batch-max-size: 50 # 每次批量消费的最大消息数
pull-batch-size: 100 # 每次从Broker拉取的消息数
topics:
project: "group-topic-1"
groups:
project: "consume-group-1" # 不同业务推荐使用不同的消费组
3. 配置类 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 配置类
*/
@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;
/**
* 主题配置类
*/
@Data
public static class TopicProperties implements Serializable {
private static final long serialVersionUID = 1L;
private String project;
}
/**
* 消费组配置类
*/
@Data
public static class GroupProperties implements Serializable {
private static final long serialVersionUID = 1L;
private String project;
}
}
4. 消费者代码实现
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;
/**
* 批量消费实现
*/
@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);
consumer.subscribe(topic, "*");
consumer.setMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> 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());
// 处理用户相关的业务逻辑
}
}
5. 生产者示例代码
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<User> users, String topic) {
List<Message> 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，避免陷入无限重试循环，可以根据具体的业务场景设置最大重试次数。
租户隔离：不同业务模块使用不同的 group，避免误消费数据，尤其在生产环境下这点非常重要。
原文链接: https://blog.csdn.net/weixin_43864927/article/details/142215529