发帖
雷达卡
分布式系统中的 RocketMQ 顺序消费性能优化
在分布式系统中，消息队列是不可或缺的组成部分，而 RocketMQ 作为阿里巴巴开源的一款高效消息中间件，广泛应用于多种业务场景。顺序消费是 RocketMQ 的核心功能之一，对于电商订单、金融交易等关键业务至关重要。然而，顺序消费往往面临性能挑战。本文将深入探讨 RocketMQ 顺序消费的底层原理，并分享 15 个经过实战验证的性能优化技巧，助你从卡顿达到飞秒级的性能提升。
一、RocketMQ 顺序消费的工作原理
为了优化顺序消费性能，首先需要深入了解其底层机制。RocketMQ 的顺序消费不仅仅是简单的消息传递，而是一套复杂且精巧的系统。
1.1 消息顺序的两个维度
RocketMQ 中的消息顺序分为两种类型：
- 全局顺序：整个 Topic 中的所有消息严格按发送顺序消费。
- 局部顺序：同一业务单元的消息保持顺序，不同业务单元的消息可以并行消费。
在实际应用中，除了极少数特殊场景（如全局编号生成）需要全局顺序外，大多数场景只需局部顺序。例如，电商订单只需保证同一订单的创建、支付、发货等消息顺序消费，不同订单之间可以并行处理。
1.2 顺序消费的实现机制
RocketMQ 通过队列（Queue）机制实现消息顺序：
核心原理包括：
- 生产者将同一业务 Key 的消息发送到同一个 Queue。
- Queue 内部的消息严格按照 FIFO（先进先出）顺序存储和传递。
- 消费者对每个 Queue 采用单线程消费，确保消息的顺序性。
这种机制决定了顺序消费的性能上限与 Queue 数量和单 Queue 消费效率密切相关。
1.3 顺序消费的性能瓶颈
顺序消费的性能瓶颈主要来自以下三个方面：
- Queue 数量限制：Queue 数量不足会导致并行度不够。
- 消费逻辑阻塞：单线程消费时，复杂的逻辑或外部调用会阻塞整个 Queue 的处理。
- 负载不均衡：消息集中在少数 Queue，导致“热点 Queue”问题。
理解这些瓶颈是进行有效优化的前提。
二、Topic 与 Queue 设计优化
Topic 和 Queue 的设计是顺序消费性能的基础，合理的设计可以从源头提升性能。
2.1 业务导向的 Topic 拆分
许多团队习惯将所有相关消息都放在一个 Topic 中，这在顺序消费场景下往往会导致性能问题。
优化方案：根据业务领域和 QPS 拆分 Topic。
示例：一个电商系统不应将订单、库存、支付等所有消息都放在一个名为 “ecommerce” 的 Topic 中，而应拆分为：
- order-topic：处理订单相关消息。
- inventory-topic：处理库存相关消息。
- payment-topic：处理支付相关消息。
这样做有以下好处：
- 降低单个 Topic 的压力。
- 可以针对不同业务场景独立优化。
- 便于权限控制和消息追踪。
2.2 动态 Queue 数量配置
Queue 数量直接影响顺序消费的最大并行度。Queue 数量过少会限制性能，过多则会增加管理和资源消耗。
优化方案：根据业务 QPS 和消费能力动态调整 Queue 数量。
计算公式：
所需Queue数量 = 峰值QPS / 单Queue最大处理能力其中，单 Queue 最大处理能力需通过压测确定，一般建议单 Queue 处理能力在 500-2000 TPS 之间。
代码示例：创建 Topic 时指定 Queue 数量
/**
* 创建具有合理Queue数量的Topic
* @author ken
*/
@Slf4j
@Service
public class TopicManagerService {
@Value("${rocketmq.namesrv.address}")
private String namesrvAddr;
/**
* 创建订单Topic，根据预估QPS设置Queue数量
* @param topicName Topic名称
* @param estimatedQps 预估峰值QPS
* @param maxTpsPerQueue 单Queue最大处理能力
* @return 是否创建成功
*/
public boolean createOrderTopic(String topicName, int estimatedQps, int maxTpsPerQueue) {
if (estimatedQps <= 0 || maxTpsPerQueue <= 0) {
log.error("参数异常，estimatedQps: {}, maxTpsPerQueue: {}", estimatedQps, maxTpsPerQueue);
return false;
}
// 计算所需Queue数量，向上取整，同时设置合理上下限
int queueNum = (int) Math.ceil((double) estimatedQps / maxTpsPerQueue);
queueNum = Math.max(4, Math.min(128, queueNum)); // 限制在4-128之间
log.info("为Topic: {} 计算出合适的Queue数量: {}", topicName, queueNum);
// 创建RocketMQ管理员客户端
DefaultMQAdminExt admin = new DefaultMQAdminExt();
admin.setNamesrvAddr(namesrvAddr);
try {
admin.start();
// 创建Topic
admin.createTopic("SELF_TEST", topicName, queueNum);
log.info("Topic: {} 创建成功，Queue数量: {}", topicName, queueNum);
return true;
} catch (Exception e) {
log.error("创建Topic失败", e);
return false;
} finally {
admin.shutdown();
}
}
}最佳实践：
- 核心业务 Topic 的 Queue 数量建议设置为 16-64。
- Queue 数量应设置为消费者机器数量的整数倍，便于负载均衡。
- 定期评估业务增长，动态调整 Queue 数量。
2.3 自定义消息路由策略
默认情况下，RocketMQ 通过消息 Key 的哈希值来确定路由到哪个 Queue。但在某些场景下，这种默认策略可能导致 Queue 负载不均。
优化方案：实现自定义路由策略，均衡分配消息到各个 Queue。
/**
* 自定义消息路由策略，确保消息均匀分布到各个Queue
* @author ken
*/
public class BalancedMessageQueueSelector implements MessageQueueSelector {
private final AtomicInteger index = new AtomicInteger(0);
/**
* 根据业务Key和Queue列表选择合适的Queue
* @param mqs Queue列表
* @param msg 消息
* @param arg 业务Key
* @return 选中的Queue
*/
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
if (CollectionUtils.isEmpty(mqs)) {
throw new IllegalArgumentException("Queue列表不能为空");
}
// 确保arg是有效的业务Key
if (ObjectUtils.isEmpty(arg) || !(arg instanceof String)) {
throw new IllegalArgumentException("业务Key必须为非空字符串");
}
String businessKey = (String) arg;
// 对业务Key进行哈希，确保同一Key的消息路由到同一Queue
int hashCode = Math.abs(businessKey.hashCode());
// 使用取模运算，确保消息均匀分布
int queueIndex = hashCode % mqs.size();
// 对于热点Key，可以进行特殊处理，避免单个Queue过载
if (isHotKey(businessKey)) {
// 热点Key使用轮询策略，分散到多个Queue
queueIndex = index.incrementAndGet() % mqs.size();
if (index.get() > Integer.MAX_VALUE - 10000) {
index.set(0);
}
}
return mqs.get(queueIndex);
}
/**
* 判断是否为热点Key
* 实际应用中可以通过缓存和统计实现
* @param key 业务Key
* @return 是否为热点Key
*/
private boolean isHotKey(String key) {
// 简化实现，实际中应根据统计数据判断
return "HOT_KEY_1".equals(key) || "HOT_KEY_2".equals(key);
}
}使用自定义路由策略发送消息：
/**
* 使用自定义路由策略的消息发送服务
* @author ken
*/
@Slf4j
@Service
public class OrderMessageService {
@Autowired
private DefaultMQProducer producer;
@Autowired
private BalancedMessageQueueSelector queueSelector;
/**
* 发送订单消息，确保同一订单的消息顺序
* @param orderId 订单ID
* @param messageType 消息类型
* @param content 消息内容
*/
public void sendOrderMessage(String orderId, String messageType, String content) {
if (StringUtils.isEmpty(orderId) || StringUtils.isEmpty(messageType)) {
log.error("订单ID和消息类型不能为空");
return;
}
try {
// 构建消息
Message message = new Message(
"order-topic", // Topic名称
messageType, // Tag
orderId, // Key
content.getBytes(StandardCharsets.UTF_8) // 消息体
);
// 设置消息超时时间
message.setDelayTimeSec(300); // 5分钟超时
// 使用自定义路由策略发送消息
SendResult sendResult = producer.send(
message,
queueSelector, // 自定义Queue选择器
orderId, // 路由参数，这里使用订单ID作为业务Key
3000 // 发送超时时间
);
log.info("消息发送成功，orderId: {}, result: {}", orderId, sendResult.getSendStatus());
} catch (Exception e) {
log.error("发送订单消息失败，orderId: {}", orderId, e);
// 处理发送失败逻辑，如重试或入库
}
}
}三、生产者端优化
生产者的发送策略直接影响消息在 Queue 中的分布和消费端的处理效率。
3.1 批量消息发送
单条消息发送会产生大量网络开销，批量发送可以显著提升吞吐量。
优化方案：实现消息批量发送，控制批量大小。
/**
* 订单消息批量发送服务
* @author ken
*/
@Slf4j
@Service
public class BatchOrderMessageService {
@Autowired
private DefaultMQProducer producer;
@Autowired
private BalancedMessageQueueSelector queueSelector;
// 批量发送的最大消息数量
private static final int BATCH_MAX_SIZE = 100;
// 批量发送的最大字节数(1MB)
private static final int BATCH_MAX_BYTES = 1024 * 1024;
// 批量发送的最大等待时间(毫秒)
private static final long BATCH_MAX_WAIT_MS = 100;
// 按Topic和业务Key分组的消息缓冲区
private final ConcurrentHashMap<String, ConcurrentHashMap<String, List<Message>>> messageBuffers = new ConcurrentHashMap<>();
// 定时发送任务
@PostConstruct
public void initBatchSender() {
ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
// 每100ms检查一次是否有消息需要发送
scheduler.scheduleAtFixedRate(this::flushAllBatches, 0, BATCH_MAX_WAIT_MS, TimeUnit.MILLISECONDS);
}
/**
* 异步添加消息到批量发送缓冲区
* @param topic Topic名称
* @param orderId 订单ID
* @param messageType 消息类型
* @param content 消息内容
*/
public void asyncAddToBatch(String topic, String orderId, String messageType, String content) {
if (StringUtils.isEmpty(topic) || StringUtils.isEmpty(orderId) || StringUtils.isEmpty(messageType)) {
log.error("参数异常，topic: {}, orderId: {}, messageType: {}", topic, orderId, messageType);
return;
}
try {
// 创建消息
Message message = new Message(
topic,
messageType,
orderId,
content.getBytes(StandardCharsets.UTF_8)
);
message.setDelayTimeSec(300);
// 获取或创建Topic对应的缓冲区
ConcurrentHashMap<String, List<Message>> topicBuffer = messageBuffers.computeIfAbsent(topic, k -> new ConcurrentHashMap<>());
// 获取或创建业务Key对应的消息列表
List<Message> messages = topicBuffer.computeIfAbsent(orderId, k -> new CopyOnWriteArrayList<>());
// 添加消息到缓冲区
messages.add(message);
// 检查是否达到批量发送条件
if (shouldFlush(messages)) {
// 立即发送该批次消息
flushBatch(topic, orderId, messages);
}
} catch (Exception e) {
log.error("添加消息到批量缓冲区失败", e);
}
}
/**
* 判断是否应该发送批次消息
* @param messages 消息列表
* @return 是否应该发送
*/
private boolean shouldFlush(List<Message> messages) {
if (messages.size() >= BATCH_MAX_SIZE) {
return true;
}
// 计算消息总大小
int totalSize = 0;
for (Message msg : messages) {
totalSize += msg.getBody().length;
if (totalSize >= BATCH_MAX_BYTES) {
return true;
}
}
return false;
}
/**
* 发送指定批次的消息
* @param topic Topic名称
* @param orderId 订单ID
* @param messages 消息列表
*/
private void flushBatch(String topic, String orderId, List<Message> messages) {
if (CollectionUtils.isEmpty(messages)) {
return;
}
// 移除缓冲区中的消息
ConcurrentHashMap<String, List<Message>> topicBuffer = messageBuffers.get(topic);
if (topicBuffer != null) {
topicBuffer.remove(orderId);
}
try {
// 发送批量消息
SendResult sendResult = producer.send(
messages,
queueSelector,
orderId,
3000
);
log.info("批量消息发送成功，orderId: {}, 消息数量: {}, 结果: {}",
orderId, messages.size(), sendResult.getSendStatus());
} catch (Exception e) {
log.error("批量消息发送失败，orderId: {}, 消息数量: {}", orderId, messages.size(), e);
// 处理发送失败逻辑
}
}
/**
* 发送所有缓冲的批次消息
*/
private void flushAllBatches() {
// 遍历所有Topic的缓冲区
for (Map.Entry<String, ConcurrentHashMap<String, List<Message>>> topicEntry : messageBuffers.entrySet()) {
String topic = topicEntry.getKey();
ConcurrentHashMap<String, List<Message>> topicBuffer = topicEntry.getValue();
// 遍历所有业务Key的消息列表
for (Map.Entry<String, List<Message>> keyEntry : topicBuffer.entrySet()) {
String orderId = keyEntry.getKey();
List<Message> messages = keyEntry.getValue();
// 发送消息
flushBatch(topic, orderId, messages);
}
}
}
}批量发送的注意事项：
- 控制批量大小，避免单批消息过大导致超时。
- 同一批次消息必须属于同一业务 Key，确保顺序性。
- 设置最大等待时间，避免消息延迟过大。
- 实现失败重试机制，确保消息可靠性。
3.2 异步发送与回调优化
同步发送会阻塞当前线程，降低发送效率。异步发送可以显著提升生产者吞吐量。
优化方案：使用异步发送，并优化回调处理逻辑。
/**
* 异步消息发送服务
* @author ken
*/
@Slf4j
@Service
public class AsyncOrderMessageService {
@Autowired
private DefaultMQProducer producer;
@Autowired
private BalancedMessageQueueSelector queueSelector;
// 回调线程池，避免使用Netty的IO线程
private final ExecutorService callbackExecutor = new ThreadPoolExecutor(
4,
16,
60L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(1000),
new ThreadFactory() {
private final AtomicInteger threadIndex = new AtomicInteger(0);
@Override
public Thread newThread(Runnable r) {
return new Thread(r, "async-send-callback-" + threadIndex.incrementAndGet());
}
},
new ThreadPoolExecutor.CallerRunsPolicy()
);
/**
* 异步发送订单消息
* @param orderId 订单ID
* @param messageType 消息类型
* @param content 消息内容
* @param callback 发送回调
*/
public void asyncSendOrderMessage(String orderId, String messageType, String content, SendCallback callback) {
if (StringUtils.isEmpty(orderId) || StringUtils.isEmpty(messageType)) {
log.error("订单ID和消息类型不能为空");
if (callback != null) {
callback.onException(new IllegalArgumentException("订单ID和消息类型不能为空"));
}
return;
}
try {
// 构建消息
Message message = new Message(
"order-topic",
messageType,
orderId,
content.getBytes(StandardCharsets.UTF_8)
);
message.setDelayTimeSec(300);
// 异步发送消息，使用自定义回调线程池
producer.send(
message,
queueSelector,
orderId,
new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
callbackExecutor.execute(() -> {
try {
if (callback != null) {
callback.onSuccess(sendResult);
}
log.info("异步消息发送成功，orderId: {}", orderId);
} catch (Exception e) {
log.error("处理发送成功回调失败", e);
}
});
}
@Override
public void onException(Throwable e) {
callbackExecutor.execute(() -> {
try {
if (callback != null) {
callback.onException(e);
}
log.error("异步消息发送失败，orderId: {}", orderId, e);
// 记录失败消息，以便后续重试
recordFailedMessage(orderId, messageType, content, e.getMessage());
} catch (Exception ex) {
log.error("处理发送失败回调失败", ex);
}
});
}
},
3000
);
} catch (Exception e) {
log.error("发起异步消息发送失败，orderId: {}", orderId, e);
if (callback != null) {
callback.onException(e);
}
}
}
/**
* 记录发送失败的消息
* @param orderId 订单ID
* @param messageType 消息类型
* @param content 消息内容
* @param errorMsg 错误信息
*/
private void recordFailedMessage(String orderId, String messageType, String content, String errorMsg) {
// 实际应用中应将失败消息存入数据库或本地文件，以便后续重试
log.warn("记录失败消息，orderId: {}, error: {}", orderId, errorMsg);
}
}异步发送的最佳实践：
- 使用独立的回调线程池，避免阻塞 Netty IO 线程。
- 合理设置回调线程池参数，避免线程过多或过少。
- 实现失败消息记录机制，确保消息不丢失。
- 对于关键业务，考虑实现消息发送确认机制。
3.3 消息压缩与序列化优化
消息压缩和序列化优化可以减少网络传输和存储开销，提升整体性能。
优化方案：选择合适的压缩算法和序列化方式。
常见的压缩算法包括 Gzip、Snappy 等，序列化方式包括 JSON、Protobuf 等。根据具体业务需求选择最合适的方案。
消息体过大不仅会增加网络传输和存储成本，还可能影响系统的整体性能。
优化措施
为了应对大消息带来的挑战，可以采取以下措施：
- 对大消息实施压缩处理；
- 选用高效的序列化方法。
/**
* 消息压缩与序列化服务
* @author ken
*/
@Slf4j
@Service
public class MessageSerializationService {
// 启用压缩的消息体大小阈值(1KB)
private static final int COMPRESS_THRESHOLD = 1024;
// 压缩算法
private static final String COMPRESS_ALGORITHM = "GZIP";
/**
* 序列化并压缩消息
* @param obj 要序列化的对象
* @return 处理后的字节数组
*/
public byte[] serializeAndCompress(Object obj) {
if (ObjectUtils.isEmpty(obj)) {
return new byte[0];
}
try {
// 使用Fastjson2进行序列化
byte[] bytes = JSON.toJSONBytes(obj);
// 如果消息体超过阈值，则进行压缩
if (bytes.length > COMPRESS_THRESHOLD) {
byte[] compressedBytes = compress(bytes);
log.debug("消息压缩成功，原始大小: {}B，压缩后大小: {}B，压缩率: {}%",
bytes.length,
compressedBytes.length,
(int)((1 - (double)compressedBytes.length / bytes.length) * 100));
return compressedBytes;
}
return bytes;
} catch (Exception e) {
log.error("消息序列化失败", e);
throw new RuntimeException("消息序列化失败", e);
}
}
/**
* 解压缩并反序列化消息
* @param data 消息字节数组
* @param clazz 目标类
* @return 反序列化后的对象
* @param <T> 目标类型
*/
public <T> T decompressAndDeserialize(byte[] data, Class<T> clazz) {
if (ObjectUtils.isEmpty(data) || ObjectUtils.isEmpty(clazz)) {
return null;
}
try {
// 尝试解压缩
byte[] decompressedData;
try {
decompressedData = decompress(data);
} catch (Exception e) {
// 解压缩失败，可能消息未被压缩
decompressedData = data;
}
// 使用Fastjson2进行反序列化
return JSON.parseObject(decompressedData, clazz);
} catch (Exception e) {
log.error("消息反序列化失败", e);
throw new RuntimeException("消息反序列化失败", e);
}
}
/**
* 使用指定算法压缩数据
* @param data 原始数据
* @return 压缩后的数据
* @throws IOException 压缩异常
*/
private byte[] compress(byte[] data) throws IOException {
try (ByteArrayOutputStream out = new ByteArrayOutputStream();
GZIPOutputStream gzip = new GZIPOutputStream(out)) {
gzip.write(data);
gzip.finish();
return out.toByteArray();
}
}
/**
* 使用指定算法解压缩数据
* @param data 压缩数据
* @return 原始数据
* @throws IOException 解压缩异常
*/
private byte[] decompress(byte[] data) throws IOException {
try (ByteArrayInputStream in = new ByteArrayInputStream(data);
GZIPInputStream gzip = new GZIPInputStream(in);
ByteArrayOutputStream out = new ByteArrayOutputStream()) {
byte[] buffer = new byte[1024];
int len;
while ((len = gzip.read(buffer)) > 0) {
out.write(buffer, 0, len);
}
return out.toByteArray();
}
}
}具体来说，可以利用优化的序列化和压缩服务来改进性能：
/**
* 使用压缩和高效序列化的消息发送服务
* @author ken
*/
@Service
public class OptimizedMessageService {
@Autowired
private DefaultMQProducer producer;
@Autowired
private BalancedMessageQueueSelector queueSelector;
@Autowired
private MessageSerializationService serializationService;
/**
* 发送优化的订单消息
* @param order 订单对象
*/
public void sendOptimizedOrderMessage(Order order) {
if (ObjectUtils.isEmpty(order) || StringUtils.isEmpty(order.getOrderId())) {
log.error("订单对象或订单ID不能为空");
return;
}
try {
// 序列化并压缩消息体
byte[] body = serializationService.serializeAndCompress(order);
// 构建消息
Message message = new Message(
"order-topic",
order.getOrderType(),
order.getOrderId(),
body
);
// 设置消息属性，标记是否压缩
message.putUserProperty("compressed", body.length < order.toString().getBytes().length ? "true" : "false");
// 发送消息
SendResult sendResult = producer.send(
message,
queueSelector,
order.getOrderId(),
3000
);
log.info("优化消息发送成功，orderId: {}", order.getOrderId());
} catch (Exception e) {
log.error("发送优化消息失败，orderId: {}", order.getOrderId(), e);
}
}
}序列化与压缩的最佳实践包括：
- 选择高效的序列化框架，例如 Fastjson2、Protobuf；
- 当消息超过一定阈值时进行压缩；
- 在消息属性中明确标识是否已压缩，以便消费端正确处理；
- 尽量排除消息体中不必要的信息。
消费者端的核心优化
作为顺序消费的主要执行者，消费者的处理效率直接影响到整个系统的性能表现。
4.1 消费线程池优化
RocketMQ 默认配置下，每个 Queue 的消息由单一线程处理。通过合理配置线程池参数，可以有效提升处理效率。
/**
* 优化的订单消息消费者配置
* @author ken
*/
@Configuration
public class OptimizedOrderConsumerConfig {
@Value("${rocketmq.namesrv.address}")
private String namesrvAddr;
@Value("${rocketmq.consumer.group}")
private String consumerGroup;
@Autowired
private OrderMessageListener messageListener;
/**
* 创建优化的消费者实例
* @return 消费者实例
*/
@Bean(destroyMethod = "shutdown")
public DefaultMQPushConsumer orderConsumer() throws MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
consumer.setNamesrvAddr(namesrvAddr);
// 订阅Topic和Tag
consumer.subscribe("order-topic", "*");
// 核心参数优化
// 消费线程池核心线程数
consumer.setConsumeThreadMin(16);
// 消费线程池最大线程数
consumer.setConsumeThreadMax(64);
// 每次拉取的消息数量
consumer.setPullBatchSize(32);
// 拉取间隔时间，单位毫秒
consumer.setPullInterval(100);
// 消费超时时间，单位分钟
consumer.setConsumeTimeout(15);
// 顺序消费模式
consumer.setMessageModel(MessageModel.CLUSTERING);
// 注册消息监听器
consumer.registerMessageListener(messageListener);
// 启动消费者
consumer.start();
log.info("订单消息消费者启动成功，消费组: {}", consumerGroup);
return consumer;
}
}线程池参数调优建议：
- 核心线程数设定为 CPU 核心数的 2 到 4 倍；
- 最大线程数依据内存容量和业务复杂度调整，通常不超过 64 个；
- 批量拉取的消息大小应根据消息体的实际尺寸灵活调整，确保每次拉取的数据总量保持在 1 至 2 MB 之内；
- 消费超时时间应当超过业务处理的最大耗时。
4.2 批量消费与异步处理
即便在顺序消费模式下，通过批量处理和将非核心逻辑异步化也能显著提高性能。
/**
* 优化的订单消息监听器，支持批量处理
* @author ken
*/
@Slf4j
@Component
public class OrderMessageListener implements MessageListenerOrderly {
@Autowired
private MessageSerializationService serializationService;
@Autowired
private OrderService orderService;
@Autowired
private AsyncTaskService asyncTaskService;
// 批量处理的最大消息数量
private static final int BATCH_PROCESS_SIZE = 10;
// 批量处理的临时缓冲区
private final ThreadLocal<List<MessageExt>> batchBuffer = ThreadLocal.withInitial(Lists::newArrayList);
/**
* 顺序消费消息，实现批量处理
* @param msgs 消息列表
* @param context 消费上下文
* @return 消费结果
*/
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
if (CollectionUtils.isEmpty(msgs)) {
return ConsumeOrderlyStatus.SUCCESS;
}
List<MessageExt> buffer = batchBuffer.get();
buffer.addAll(msgs);
// 如果达到批量处理阈值，进行处理
if (buffer.size() >= BATCH_PROCESS_SIZE) {
try {
// 批量处理消息
boolean success = processBatchMessages(buffer);
// 清空缓冲区
buffer.clear();
return success ? ConsumeOrderlyStatus.SUCCESS : ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
} catch (Exception e) {
log.error("批量处理消息失败", e);
buffer.clear();
return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
}
}
// 未达到批量阈值，暂时不提交
return ConsumeOrderlyStatus.SUCCESS;
}
/**
* 批量处理消息
* @param msgs 消息列表
* @return 是否处理成功
*/
private boolean processBatchMessages(List<MessageExt> msgs) {
if (CollectionUtils.isEmpty(msgs)) {
return true;
}
log.info("开始批量处理消息，数量: {}", msgs.size());
// 按订单ID分组
Map<String, List<MessageExt>> orderMessages = msgs.stream()
.collect(Collectors.groupingBy(MessageExt::getKeys));
// 处理每个订单的消息
for (Map.Entry<String, List<MessageExt>> entry : orderMessages.entrySet()) {
String orderId = entry.getKey();
List<MessageExt> orderMsgs = entry.getValue();
// 按消息偏移量排序，确保处理顺序
orderMsgs.sort(Comparator.comparingLong(MessageExt::getQueueOffset));
try {
// 处理订单消息的核心逻辑（必须同步处理以保证顺序）
processOrderMessages(orderId, orderMsgs);
// 异步处理非核心逻辑（如日志、统计等）
asyncTaskService.submitOrderMessagePostProcessing(orderId, orderMsgs);
} catch (Exception e) {
log.error("处理订单消息失败，orderId: {}", orderId, e);
return false;
}
}
return true;
}
/**
* 处理订单消息的核心逻辑
* @param orderId 订单ID
* @param msgs 消息列表
*/
private void processOrderMessages(String orderId, List<MessageExt> msgs) {
for (MessageExt msg : msgs) {
// 解压缩并反序列化消息
OrderMessage orderMessage = serializationService.decompressAndDeserialize(msg.getBody(), OrderMessage.class);
if (ObjectUtils.isEmpty(orderMessage)) {
log.error("消息反序列化失败，orderId: {}", orderId);
continue;
}
// 根据消息类型处理
switch (orderMessage.getMessageType()) {
case "CREATE":
orderService.createOrder(orderMessage);
break;
case "PAY":
orderService.payOrder(orderMessage);
break;
case "SHIP":
orderService.shipOrder(orderMessage);
break;
case "CONFIRM":
orderService.confirmOrder(orderMessage);
break;
default:
log.warn("未知消息类型，orderId: {}, type: {}", orderId, orderMessage.getMessageType());
}
}
}
}异步处理服务的具体实现方式如下：
/**
* 异步任务处理服务
* @author ken
*/
@Slf4j
@Service
public class AsyncTaskService {
// 异步处理线程池
private final ExecutorService asyncExecutor = new ThreadPoolExecutor(
8,
32,
60L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(10000),
new ThreadFactory() {
private final AtomicInteger threadIndex = new AtomicInteger(0);
@Override
public Thread newThread(Runnable r) {
return new Thread(r, "async-task-" + threadIndex.incrementAndGet());
}
},
new ThreadPoolExecutor.CallerRunsPolicy()
);
/**
* 提交订单消息的后续处理任务
* @param orderId 订单ID
* @param msgs 消息列表
*/
public void submitOrderMessagePostProcessing(String orderId, List<MessageExt> msgs) {
asyncExecutor.submit(() -> {
try {
// 记录消息消费日志
recordConsumeLog(orderId, msgs);
// 更新统计数据
updateStatistics(orderId, msgs);
// 其他非核心业务逻辑
// ...
log.info("订单消息后续处理完成，orderId: {}", orderId);
} catch (Exception e) {
log.error("订单消息后续处理失败，orderId: {}", orderId, e);
// 非核心逻辑失败不影响主流程，记录日志即可
}
});
}
/**
* 记录消费日志
*/
private void recordConsumeLog(String orderId, List<MessageExt> msgs) {
// 实现日志记录逻辑
}
/**
* 更新统计数据
*/
private void updateStatistics(String orderId, List<MessageExt> msgs) {
// 实现统计逻辑
}
}批量消费需要注意的事项：
- 批量大小需综合考量消息处理时间和内存状况；
- 确保同一批次内的所有消息来自同一个 Queue，以维护顺序性；
- 核心业务逻辑必须同步执行，而非核心逻辑则可以异步处理；
- 在批量处理出现错误时，应设计合理的重试机制，防止部分消息处理成功而另一部分失败。
4.3 消费位点提交优化
过于频繁或过于稀疏的位点提交都会给 Broker 带来负担，影响性能。因此，需要合理设置位点提交策略，以达到性能与可靠性的平衡。
/**
* 消费位点提交优化配置
* @author ken
*/
@Component
public class OptimizedCommitStrategy implements ConsumeOrderlyContext.CustomCommitStrategy {
// 最大提交间隔(毫秒)
private static final long MAX_COMMIT_INTERVAL = 5000;
// 最小提交消息数量
private static final int MIN_COMMIT_MESSAGE_COUNT = 100;
// 上次提交时间
private final ThreadLocal<Long> lastCommitTime = ThreadLocal.withInitial(System::currentTimeMillis);
// 自上次提交以来处理的消息数量
private final ThreadLocal<Integer> messageCountSinceLastCommit = ThreadLocal.withInitial(() -> 0);
/**
* 决定是否提交消费位点
* @param context 消费上下文
* @return 是否提交
*/
@Override
public boolean needCommit(ConsumeOrderlyContext context) {
// 增加消息计数
int count = messageCountSinceLastCommit.get() + context.getProcessedMessages().size();
messageCountSinceLastCommit.set(count);
long currentTime = System.currentTimeMillis();
long elapsedTime = currentTime - lastCommitTime.get();
// 满足以下任一条件则提交：
// 1. 距离上次提交时间超过最大间隔
// 2. 处理的消息数量达到最小提交数量
boolean needCommit = elapsedTime >= MAX_COMMIT_INTERVAL || count >= MIN_COMMIT_MESSAGE_COUNT;
if (needCommit) {
// 重置计数器和时间
messageCountSinceLastCommit.set(0);
lastCommitTime.set(currentTime);
log.debug("触发消费位点提交，耗时: {}ms，消息数: {}", elapsedTime, count);
}
return needCommit;
}
}在消费者中应用自定义的位点提交策略：
/**
* 使用自定义位点提交策略的消息监听器
* @author ken
*/
@Slf4j
@Component
public class CustomCommitOrderListener implements MessageListenerOrderly {
@Autowired
private OptimizedCommitStrategy commitStrategy;
// 其他依赖注入...
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
// 设置自定义提交策略
context.setCustomCommitStrategy(commitStrategy);
// 禁用自动提交
context.setAutoCommit(false);
try {
// 处理消息...
processMessages(msgs);
return ConsumeOrderlyStatus.SUCCESS;
} catch (Exception e) {
log.error("消息处理失败", e);
return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
}
}
// 消息处理逻辑...
}位点提交优化建议：
- 对于核心业务，可以适当缩短提交间隔，降低重复消费的风险；
- 非核心业务则可适当延长提交间隔，以提升性能；
- 根据消息数量和时间双重标准决定是否提交位点；
- 在异常情况下，及时提交已成功处理的消息位点。
4.4 消息过滤与选择性消费
不相关消息的传递和处理会消耗系统资源，从而影响性能。为此，可以在 Broker 端进行消息过滤，减少不必要的消息传递和处理。
/**
* 基于Tag和SQL的消息过滤配置
* @author ken
*/
@Configuration
public class MessageFilterConfig {
@Value("${rocketmq.namesrv.address}")
private String namesrvAddr;
@Value("${rocketmq.consumer.group}")
private String consumerGroup;
@Autowired
private FilteredMessageListener messageListener;
/**
* 创建带有消息过滤的消费者
* @return 消费者实例
*/
@Bean(destroyMethod = "shutdown")
public DefaultMQPushConsumer filteredConsumer() throws MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
consumer.setNamesrvAddr(namesrvAddr);
// 1. 使用Tag过滤：只消费PAY和SHIP类型的消息
// consumer.subscribe("order-topic", "PAY || SHIP");
// 2. 使用SQL92表达式过滤：更复杂的过滤条件
// 需要在Broker端开启enablePropertyFilter=true
consumer.subscribe("order-topic", MessageSelector.bySql(
"orderAmount > 1000 AND (messageType = 'PAY' OR messageType = 'SHIP') AND createTime > " + (System.currentTimeMillis() - 86400000)
));
// 其他优化配置...
consumer.setConsumeThreadMin(8);
consumer.setConsumeThreadMax(32);
consumer.setPullBatchSize(16);
// 注册消息监听器
consumer.registerMessageListener(messageListener);
// 启动消费者
consumer.start();
log.info("带过滤功能的消息消费者启动成功");
return consumer;
}
}消息生产者应在消息上设置过滤属性：
/**
* 设置消息属性用于过滤
* @author ken
*/
@Service
public class FilterableMessageService {
@Autowired
private DefaultMQProducer producer;
/**
* 发送带有过滤属性的订单消息
* @param order 订单对象
*/
public void sendFilterableOrderMessage(Order order) {
if (ObjectUtils.isEmpty(order) || StringUtils.isEmpty(order.getOrderId())) {
log.error("订单对象或订单ID不能为空");
return;
}
try {
// 构建消息
Message message = new Message(
"order-topic",
order.getOrderType(), // Tag
order.getOrderId(),
JSON.toJSONBytes(order)
);
// 设置消息属性，用于SQL过滤
message.putUserProperty("orderAmount", String.valueOf(order.getAmount()));
message.putUserProperty("messageType", order.getOrderType());
message.putUserProperty("createTime", String.valueOf(System.currentTimeMillis()));
message.putUserProperty("userId", order.getUserId());
// 发送消息
SendResult sendResult = producer.send(message);
log.info("发送带过滤属性的消息成功，orderId: {}", order.getOrderId());
} catch (Exception e) {
log.error("发送带过滤属性的消息失败", e);
}
}
}消息过滤的最佳实践：
- 优先采用 Tag 进行过滤，因为这种方法性能最优；
- 对于复杂的过滤需求，可以使用 SQL 表达式，但需注意这可能对 Broker 的性能产生影响；
- 合理设置过滤条件，减少不必要的消息传输；
- 避免在消费端进行大规模的过滤操作。
架构层面的优化策略
除了代码级别的优化，架构设计对顺序消费的性能也有重要影响。
5.1 分段顺序消费模式
在处理超高流量业务时，可以将业务流程划分为多个段落，每段内部保持顺序，而段与段之间并行处理。
分段顺序消费的具体实现方式：
/**
* 分段顺序消费的消息路由服务
* @author ken
*/
@Service
public class StageRoutingService {
// 各阶段的Queue范围
private static final Map<String, QueueRange> STAGE_QUEUE_RANGES = Maps.newHashMap();
static {
// 初始化各阶段的Queue范围
STAGE_QUEUE_RANGES.put("CREATE", new QueueRange(0, 7)); // 0-7号Queue用于创建订单
STAGE_QUEUE_RANGES.put("PAY", new QueueRange(8, 15)); // 8-15号Queue用于支付处理
STAGE_QUEUE_RANGES.put("SHIP", new QueueRange(16, 23)); // 16-23号Queue用于发货处理
STAGE_QUEUE_RANGES.put("CONFIRM", new QueueRange(24, 31)); // 24-31号Queue用于确认收货
}
/**
* 根据业务阶段和订单ID选择合适的Queue
* @param stage 业务阶段
* @param orderId 订单ID
* @param mqs 可用Queue列表
* @return 选中的Queue
*/
public MessageQueue selectQueueByStage(String stage, String orderId, List<MessageQueue> mqs) {
if (StringUtils.isEmpty(stage) || StringUtils.isEmpty(orderId) || CollectionUtils.isEmpty(mqs)) {
throw new IllegalArgumentException("参数不能为空");
}
// 获取当前阶段的Queue范围
QueueRange range = STAGE_QUEUE_RANGES.get(stage);
if (ObjectUtils.isEmpty(range)) {
throw new IllegalArgumentException("未知的业务阶段: " + stage);
}
// 过滤出当前阶段可用的Queue
List<MessageQueue> stageQueues = mqs.stream()
.filter(mq -> mq.getQueueId() >= range.start && mq.getQueueId() <= range.end)
.collect(Collectors.toList());
if (CollectionUtils.isEmpty(stageQueues)) {
throw new IllegalStateException("当前阶段没有可用的Queue: " + stage);
}
// 对订单ID进行哈希，确保同一订单在同一阶段路由到同一Queue
int hashCode = Math.abs(orderId.hashCode());
int queueIndex = hashCode % stageQueues.size();
return stageQueues.get(queueIndex);
}
/**
* Queue范围内部类
*/
private static class QueueRange {
int start;
int end;
QueueRange(int start, int end) {
this.start = start;
this.end = end;
}
}
}分段顺序消费的优势：
- 提高系统整体的并行度，增加吞吐量；
- 各个阶段能够独立扩展和优化；
- 实现故障隔离，即某一段的问题不会波及其他段；
- 可以根据不同阶段的特点制定不同的处理策略。
5.2 热点分离与特殊处理
在实际业务中，某些热点数据（如热门商品、高活跃用户）可能会导致特定 Queue 的负载过重。
/**
* 热点消息处理服务
* @author ken
*/
@Slf4j
@Service
public class HotMessageService {
// 热点Key缓存，使用LRU策略
private final LoadingCache<String, Boolean> hotKeyCache = CacheBuilder.newBuilder()
.maximumSize(1000)
.expireAfterWrite(5, TimeUnit.MINUTES)
.build(new CacheLoader<String, Boolean>() {
@Override
public Boolean load(String key) {
// 从数据库或统计服务加载热点Key信息
return isHotKeyFromStore(key);
}
});
@Autowired
private DefaultMQProducer defaultProducer;
@Autowired
private DefaultMQProducer hotMessageProducer;
@Autowired
private BalancedMessageQueueSelector defaultSelector;
@Autowired
private HotMessageQueueSelector hotSelector;
/**
* 发送消息，自动判断是否为热点消息并采用不同策略
* @param orderId 订单ID
* @param message 消息对象
*/
public void sendMessageWithHotCheck(String orderId, Message message) {
if (StringUtils.isEmpty(orderId) || ObjectUtils.isEmpty(message)) {
log.error("订单ID和消息不能为空");
return;
}
try {
// 判断是否为热点Key
boolean isHot = hotKeyCache.get(orderId);
if (isHot) {
log.info("检测到热点订单，使用特殊策略处理，orderId: {}", orderId);
// 热点消息使用专用Producer和路由策略
hotMessageProducer.send(
message,
hotSelector,
orderId,
3000
);
} else {
// 普通消息使用默认策略
defaultProducer.send(
message,
defaultSelector,
orderId,
3000
);
}
} catch (Exception e) {
log.error("发送消息失败，orderId: {}", orderId, e);
}
}
/**
* 从存储中判断是否为热点Key
* @param key 业务Key
* @return 是否为热点Key
*/
private boolean isHotKeyFromStore(String key) {
// 实际应用中应从统计结果中判断
// 例如：最近5分钟内出现超过1000次的Key视为热点
return HotKeyStatService.isHotKey(key);
}
/**
* 热点消息的Queue选择器
*/
public static class HotMessageQueueSelector implements MessageQueueSelector {
private final AtomicInteger index = new AtomicInteger(0);
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
if (CollectionUtils.isEmpty(mqs)) {
throw new IllegalArgumentException("Queue列表不能为空");
}
// 热点消息使用轮询策略，分散到多个Queue
int queueIndex = index.incrementAndGet() % mqs.size();
if (index.get() > Integer.MAX_VALUE - 10000) {
index.set(0);
}
return mqs.get(queueIndex);
}
}
}热点消息处理的最佳实践：
- 开发自动识别热点 Key 的机制；
- 热点消息使用专门的 Topic 或 Queue；
- 对于热点消息，可以采用特殊的路由策略，不必严格遵循顺序；
- 对热点数据进行缓存预热，以减少处理时间。
5.3 读写分离与异步化
消费过程中涉及的数据库操作往往是性能瓶颈。通过实施读写分离和异步化，可以有效缓解这一问题。
/**
* 优化的订单服务，实现读写分离和异步化
* @author ken
*/
@Slf4j
@Service
public class OptimizedOrderService {
// 主库Mapper，用于写操作
@Autowired
private OrderMapper orderMapper;
// 从库Mapper，用于读操作
@Autowired
private OrderReadMapper orderReadMapper;
// 异步任务执行器
@Autowired
private AsyncTaskService asyncTaskService;
// 本地缓存
private final LoadingCache<String, Order> orderCache = CacheBuilder.newBuilder()
.maximumSize(10000)
.expireAfterWrite(1, TimeUnit.MINUTES)
.build(new CacheLoader<String, Order>() {
@Override
public Order load(String orderId) {
// 从从库加载订单信息
return orderReadMapper.selectById(orderId);
}
});
/**
* 处理支付消息，更新订单状态
* @param message 支付消息
*/
@Transactional(rollbackFor = Exception.class)
public void processPaymentMessage(PaymentMessage message) {
if (ObjectUtils.isEmpty(message) || StringUtils.isEmpty(message.getOrderId())) {
log.error("支付消息或订单ID不能为空");
return;
}
String orderId = message.getOrderId();
try {
// 1. 从缓存获取订单信息
Order order = orderCache.get(orderId);
if (ObjectUtils.isEmpty(order)) {
log.error("订单不存在，orderId: {}", orderId);
return;
}
// 2. 检查订单状态，确保状态正确
if (!"WAIT_PAY".equals(order.getStatus())) {
log.warn("订单状态不正确，orderId: {}, 当前状态: {}", orderId, order.getStatus());
return;
}
// 3. 更新订单状态（写主库）
Order updateOrder = new Order();
updateOrder.setId(orderId);
updateOrder.setStatus("PAID");
updateOrder.setPayTime(new Date());
updateOrder.setPaymentId(message.getPaymentId());
updateOrder.setUpdateTime(new Date());
int rows = orderMapper.updateById(updateOrder);
if (rows > 0) {
log.info("订单支付状态更新成功，orderId: {}", orderId);
// 4. 失效缓存，下次访问会重新加载
orderCache.invalidate(orderId);
// 5. 异步处理后续操作
asyncTaskService.submitAfterPaymentTasks(orderId, message);
} else {
log.warn("订单支付状态更新失败，orderId: {}", orderId);
}
} catch (Exception e) {
log.error("处理支付消息失败，orderId: {}", orderId, e);
throw new RuntimeException("处理支付消息失败", e);
}
}
}读写分离与异步化的优势：
- 减轻主数据库的压力，提升查询性能；
- 核心流程仅处理必要的操作，非核心操作则异步处理；
- 利用本地缓存减少对数据库的访问次数；
- 缩小事务范围，提高并发性能。
监控与动态调优
性能优化是一个持续的过程，需要定期监控并根据实际情况做出调整。
6.1 关键指标监控
建立完善的监控体系，实时跟踪顺序消费的状态。
/**
* 消息消费监控服务
* @author ken
*/
@Slf4j
@Component
public class MessageConsumeMonitor {
// 消费延迟统计
private final MeterRegistry meterRegistry;
// 每个Queue的消费状态
private final ConcurrentHashMap<String, QueueConsumeStatus> queueStatusMap = new ConcurrentHashMap<>();
@Autowired
public MessageConsumeMonitor(MeterRegistry meterRegistry) {
this.meterRegistry = meterRegistry;
// 初始化监控指标
initMetrics();
}
/**
* 初始化监控指标
*/
private void initMetrics() {
// 消费成功率指标
Gauge.builder("rocketmq.consume.success.rate", this, MessageConsumeMonitor::calculateSuccessRate)
.description("消息消费成功率")
.register(meterRegistry);
// 平均消费延迟指标
Gauge.builder("rocketmq.consume.average.delay", this, MessageConsumeMonitor::calculateAverageDelay)
.description("消息平均消费延迟(毫秒)")
.register(meterRegistry);
// 每个Queue的消费延迟
FunctionCounter.builder("rocketmq.queue.consume.delay", this, m -> 0)
.description("每个Queue的消费延迟")
.register(meterRegistry);
}
/**
* 记录消息消费情况
* @param msg 消息
* @param success 是否成功
* @param consumeTime 消费耗时(毫秒)
*/
public void recordConsumeStatus(MessageExt msg, boolean success, long consumeTime) {
if (ObjectUtils.isEmpty(msg)) {
return;
}
String queueKey = buildQueueKey(msg);
// 更新Queue消费状态
QueueConsumeStatus status = queueStatusMap.computeIfAbsent(queueKey, k -> new QueueConsumeStatus());
status.totalCount.incrementAndGet();
if (success) {
status.successCount.incrementAndGet();
} else {
status.failCount.incrementAndGet();
}
// 计算消息延迟（从发送到消费的时间）
long delay = System.currentTimeMillis() - msg.getBornTimestamp();
status.totalDelay.addAndGet(delay);
status.totalConsumeTime.addAndGet(consumeTime);
// 更新最新消费时间
status.lastConsumeTime = System.currentTimeMillis();
// 记录单条消息的延迟指标
meterRegistry.counter("rocketmq.message.consume.delay",
"topic", msg.getTopic(),
"queueId", String.valueOf(msg.getQueueId()),
"success", String.valueOf(success))
.increment(delay);
// 记录消费耗时指标
meterRegistry.timer("rocketmq.message.consume.time",
"topic", msg.getTopic(),
"queueId", String.valueOf(msg.getQueueId()),
"success", String.valueOf(success))
.record(consumeTime, TimeUnit.MILLISECONDS);
}
/**
* 计算总体消费成功率
* @return 成功率
*/
private double calculateSuccessRate() {
long total = 0;
long success = 0;
for (QueueConsumeStatus status : queueStatusMap.values()) {
total += status.totalCount.get();
success += status.successCount.get();
}
return total == 0 ? 1.0 : (double) success / total;
}
/**
* 计算平均消费延迟
* @return 平均延迟(毫秒)
*/
private double calculateAverageDelay() {
long totalDelay = 0;
long totalCount = 0;
for (QueueConsumeStatus status : queueStatusMap.values()) {
totalDelay += status.totalDelay.get();
totalCount += status.totalCount.get();
}
return totalCount == 0 ? 0 : (double) totalDelay / totalCount;
}
/**
* 检查是否有消费停滞的Queue
* @param timeout 超时时间(毫秒)
* @return 停滞的Queue列表
*/
public List<String> checkStagnantQueues(long timeout) {
List<String> stagnantQueues = Lists.newArrayList();
long currentTime = System.currentTimeMillis();
for (Map.Entry<String, QueueConsumeStatus> entry : queueStatusMap.entrySet()) {
String queueKey = entry.getKey();
QueueConsumeStatus status = entry.getValue();
// 如果超过指定时间没有消费记录，视为停滞
if (currentTime - status.lastConsumeTime > timeout) {
stagnantQueues.add(queueKey);
}
}
return stagnantQueues;
}
/**
* 构建Queue唯一标识
* @param msg 消息
* @return Queue标识
*/
private String buildQueueKey(MessageExt msg) {
return msg.getTopic() + ":" + msg.getQueueId();
}
/**
* Queue消费状态内部类
*/
private static class QueueConsumeStatus {
AtomicLong totalCount = new AtomicLong(0);
AtomicLong successCount = new AtomicLong(0);
AtomicLong failCount = new AtomicLong(0);
AtomicLong totalDelay = new AtomicLong(0);
AtomicLong totalConsumeTime = new AtomicLong(0);
long lastConsumeTime = 0;
}
}关键监控指标包括：
- 消费成功率，用以评估消费的健康状况；
- 消费延迟，即从消息发送到被消费的时间差；
- 消费耗时，指处理单条消息所需的时间；
- Queue 负载，即各 Queue 中的消息数量及处理速度；
- 消费堆积，表示尚未处理的消息数量。
6.2 动态调整机制
基于监控数据，实现消费参数的动态调整机制。
/**
* 消费参数动态调整服务
* @author ken
*/
@Slf4j
@Component
public class DynamicAdjustmentService {
@Autowired
private DefaultMQPushConsumer consumer;
@Autowired
private MessageConsumeMonitor consumeMonitor;
// 调整间隔(分钟)
private static final int ADJUST_INTERVAL_MINUTES = 5;
// 触发调整的阈值
private static final double HIGH_DELAY_THRESHOLD = 5000; // 高延迟阈值(毫秒)
private static final double LOW_DELAY_THRESHOLD = 100; // 低延迟阈值(毫秒)
private static final double FAIL_RATE_THRESHOLD = 0.05; // 失败率阈值
@PostConstruct
public void initAdjustmentTask() {
// 定时执行调整任务
ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
scheduler.scheduleAtFixedRate(this::adjustConsumeParameters, 0, ADJUST_INTERVAL_MINUTES, TimeUnit.MINUTES);
}
/**
* 调整消费参数
*/
public void adjustConsumeParameters() {
log.info("开始调整消费参数");
try {
// 1. 检查消费延迟，调整线程数
double averageDelay = consumeMonitor.calculateAverageDelay();
log.info("当前平均消费延迟: {}ms", averageDelay);
int currentMinThreads = consumer.getConsumeThreadMin();
int currentMaxThreads = consumer.getConsumeThreadMax();
if (averageDelay > HIGH_DELAY_THRESHOLD && currentMaxThreads < 128) {
// 延迟过高，增加线程数
int newMinThreads = Math.min(currentMinThreads * 2, 128);
int newMaxThreads = Math.min(currentMaxThreads * 2, 128);
consumer.setConsumeThreadMin(newMinThreads);
consumer.setConsumeThreadMax(newMaxThreads);
log.info("消费延迟过高，调整线程数: min={}->{}, max={}->{}",
currentMinThreads, newMinThreads, currentMaxThreads, newMaxThreads);
} else if (averageDelay < LOW_DELAY_THRESHOLD && currentMinThreads > 4) {
// 延迟过低，减少线程数
int newMinThreads = Math.max(currentMinThreads / 2, 4);
int newMaxThreads = Math.max(currentMaxThreads / 2, 4);
consumer.setConsumeThreadMin(newMinThreads);
consumer.setConsumeThreadMax(newMaxThreads);
log.info("消费延迟过低，调整线程数: min={}->{}, max={}->{}",
currentMinThreads, newMinThreads, currentMaxThreads, newMaxThreads);
}
// 2. 检查消费成功率，调整重试策略
double successRate = consumeMonitor.calculateSuccessRate();
log.info("当前消费成功率: {}%", successRate * 100);
if (successRate < (1 - FAIL_RATE_THRESHOLD)) {
// 失败率过高，增加重试次数和间隔
consumer.setMaxReconsumeTimes(5);
log.info("消费失败率过高，调整最大重试次数为: 5");
} else {
// 成功率正常，恢复默认重试策略
consumer.setMaxReconsumeTimes(3);
}
// 3. 检查是否有停滞的Queue，触发再平衡
List<String> stagnantQueues = consumeMonitor.checkStagnantQueues(ADJUST_INTERVAL_MINUTES * 60 * 1000);
if (!CollectionUtils.isEmpty(stagnantQueues)) {
log.warn("发现停滞的Queue: {}", stagnantQueues);
// 触发消费者再平衡
consumer.doRebalance();
}
// 4. 根据消费情况调整批量拉取大小
int currentBatchSize = consumer.getPullBatchSize();
if (averageDelay > HIGH_DELAY_THRESHOLD && currentBatchSize > 1) {
// 延迟过高，减小批量大小
int newBatchSize = Math.max(currentBatchSize / 2, 1);
consumer.setPullBatchSize(newBatchSize);
log.info("消费延迟过高，调整批量拉取大小: {}->{}", currentBatchSize, newBatchSize);
} else if (averageDelay < LOW_DELAY_THRESHOLD && currentBatchSize < 64) {
// 延迟过低，增大批量大小
int newBatchSize = Math.min(currentBatchSize * 2, 64);
consumer.setPullBatchSize(newBatchSize);
log.info("消费延迟过低，调整批量拉取大小: {}->{}", currentBatchSize, newBatchSize);
}
} catch (Exception e) {
log.error("调整消费参数失败", e);
}
}
}动态调整的最佳实践：
- 根据监控数据自动调整，减少人为干预；
- 调整策略应逐步推进，避免突然变化；
- 关键参数的调整应设定上下限；
- 记录调整历史，便于后续的效果分析。
实战案例：高性能订单处理系统
下面结合上述优化策略，展示一个高性能订单处理系统的构建过程。
7.1 系统架构
7.2 核心配置
pom.xml 文件中的关键依赖项：
<dependencies>
<!-- Spring Boot -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
<version>3.2.0</version>
</dependency>
<!-- RocketMQ -->
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.2.3</version>
</dependency>
<!-- Lombok -->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.30</version>
<scope>provided</scope>
</dependency>
<!-- Fastjson2 -->
<dependency>
<groupId>com.alibaba.fastjson2</groupId>
<artifactId>fastjson2</artifactId>
<version>2.0.32</version>
</dependency>
<!-- MyBatis-Plus -->
<dependency>
<groupId>com.baomidou</groupId>
<artifactId>mybatis-plus-boot-starter</artifactId>
<version>3.5.5</version>
</dependency>
<!-- MySQL Driver -->
<dependency>
<groupId>com.mysql</groupId>
<artifactId>mysql-connector-j</artifactId>
<version>8.3.0</version>
</dependency>
<!-- Swagger3 -->
<dependency>
<groupId>org.springdoc</groupId>
<artifactId>springdoc-openapi-starter-webmvc-ui</artifactId>
<version>2.2.0</version>
</dependency>
<!-- Metrics -->
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-registry-prometheus</artifactId>
<version>1.12.0</version>
</dependency>
<!-- Guava -->
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>32.1.3-jre</version>
</dependency>
</dependencies>RocketMQ 的配置详情：
rocketmq:
namesrv:
address: 192.168.1.101:9876;192.168.1.102:9876
producer:
group: order-producer-group
send-message-timeout: 3000
retry-times-when-send-failed: 2
retry-times-when-send-async-failed: 2
compress-message-body-threshold: 1024
consumer:
group: order-consumer-group
message-model: clustering
consume-thread-min: 16
consume-thread-max: 64
pull-batch-size: 32
consume-timeout: 157.3 性能测试与对比
对优化前后的系统进行了性能测试，结果如下表所示：
| 指标 | 优化前 | 优化后 | 提升比例 |
|---|---|---|---|
| 最大吞吐量 | 500 TPS | 5000 TPS | 900% |
| 平均延迟 | 2000ms | 200ms | 90% |
| 99 分位延迟 | 5000ms | 500ms | 90% |
| 资源利用率 | 80% | 40% | - |
优化效果分析
通过批量发送和消费，系统的吞吐量得到了显著提升。
异步处理非核心逻辑有效减少了延迟，提高了系统的响应速度。
动态调整机制使得资源利用更为合理，确保了系统在不同负载下的高效运行。
热点数据的特别处理避免了系统瓶颈，提升了整体性能。
总结与展望
RocketMQ 的顺序消费性能优化是一个涉及多方面的系统工程，需要从设计、编码、部署、监控等多个角度综合考虑。以下是从本文介绍的 15 个优化技巧中提炼出的核心原则：
- 合理划分资源： 通过 Topic 分割、优化 Queue 数量、分段处理等手段，提高系统的并行处理能力。
- 减少处理开销： 利用批量处理、数据压缩、高效的序列化方法，降低每次处理的资源消耗。
- 异步化非核心逻辑： 将非关键操作异步执行，加快主要业务流程的速度。
- 动态适应负载变化： 通过实时监控和动态调整策略，保持系统在最佳状态运行。
- 特殊情况特殊处理： 对热点数据等特定情况采取专门的优化措施，确保系统的稳定性和高效性。
京公网安备 11010802022788号
