【SpringBoot】RocketMQ实战：从基础集成到生产级消息处理

1. 环境准备与基础集成第一次接触RocketMQ时我踩过不少坑。记得有次凌晨三点还在调试生产者连接问题最后发现是NameServer地址少了个端口号。为了让你们少走弯路我把SpringBoot集成RocketMQ的完整流程梳理成可落地的步骤。安装RocketMQ服务端推荐用Docker方式快速启动docker pull rocketmqinc/rocketmq docker run -d --name rmqnamesrv -p 9876:9876 rocketmqinc/rocketmq sh mqnamesrv docker run -d --name rmqbroker -p 10911:10911 -p 10909:10909 \ -e NAMESRV_ADDRlocalhost:9876 rocketmqinc/rocketmq sh mqbroker在SpringBoot项目中添加依赖时要注意版本匹配。最近有个生产事故就是因为用了老版本starter导致消息堆积dependency groupIdorg.apache.rocketmq/groupId artifactIdrocketmq-spring-boot-starter/artifactId version2.2.3/version /dependency配置文件application.yml的坑最多这三个参数必须检查三遍rocketmq: name-server: 127.0.0.1:9876 # 多个用分号隔开 producer: group: order-service-group # 生产组名要唯一 send-message-timeout: 5000 # 线上建议调大2. 消息收发核心实现2.1 生产者最佳实践我封装过一个增强版RocketMQTemplate解决了消息轨迹和异常重试的问题。先看基础发送代码Autowired private RocketMQTemplate rocketMQTemplate; // 同步发送适合重要订单消息 public SendResult sendOrderMessage(Order order) { MessageOrder message MessageBuilder .withPayload(order) .setHeader(KEYS, order.getOrderNo()) .build(); return rocketMQTemplate.syncSend(ORDER_TOPIC:PAY, message); } // 异步发送适合日志类消息 public void asyncSendLog(LogMessage log) { rocketMQTemplate.asyncSend(LOG_TOPIC, log, new SendCallback() { Override public void onSuccess(SendResult sendResult) { metrics.increment(send_success); } Override public void onException(Throwable e) { log.error(消息发送失败, e); } }); }2.2 消费者避坑指南消费者最容易被忽略的是幂等处理。去年双十一我们系统因为没做去重导致重复发货损失惨重Service RocketMQMessageListener( topic ORDER_TOPIC, selectorExpression PAY, consumerGroup order-pay-group ) public class OrderPayConsumer implements RocketMQListenerOrder { Override Transactional public void onMessage(Order order) { if(redis.setnx(order:order.getOrderNo(), 1, 24, HOURS)){ orderService.processPayment(order); // 业务处理 } } }遇到消息堆积时可以调整消费线程数实测20个线程最稳定rocketmq: consumer: consume-thread-max: 20 consume-timeout: 15m # 防止长时间阻塞3. 高级特性实战3.1 顺序消息实现电商订单状态变更必须严格有序我们通过自定义队列选择器实现// 生产者端 rocketMQTemplate.syncSendOrderly( ORDER_STATUS_TOPIC, message, order.getUserId() // 相同用户ID的消息进入同一队列 ); // 消费者端 RocketMQMessageListener( consumeMode ConsumeMode.ORDERLY // 关键配置 ) public class OrderStatusListener implements RocketMQListenerOrder { Override public void onMessage(Order order) { // 处理订单状态 } }3.2 事务消息方案支付场景要用到事务消息这个回调机制很多人没搞明白// 1. 发送半消息 TransactionSendResult result rocketMQTemplate.sendMessageInTransaction( tx-order-group, PAY_TOPIC, message, null ); // 2. 实现事务监听器 RocketMQTransactionListener(txProducerGrouptx-order-group) public class OrderTransactionListener implements RocketMQLocalTransactionListener { Override public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) { try { orderService.createOrder(msg); // 本地事务 return RocketMQLocalTransactionState.COMMIT; } catch(Exception e) { return RocketMQLocalTransactionState.ROLLBACK; } } Override public RocketMQLocalTransactionState checkLocalTransaction(Message msg) { return orderService.checkOrderStatus(msg) ? COMMIT : UNKNOWN; // 15分钟后会再次检查 } }4. 生产环境调优4.1 消息丢失防护我们通过三重保障解决消息丢失问题生产者开启重试机制rocketmq: producer: retry-times-when-send-failed: 3Broker配置同步刷盘性能会下降30%flushDiskTypeSYNC_FLUSH消费者手动ACKRocketMQMessageListener(ackModeMANUAL) public class SafeConsumer implements RocketMQListenerMessageExt { Override public void onMessage(MessageExt message) { try { process(message); consumer.ack(message); } catch(Exception e) { consumer.requeue(message); } } }4.2 性能优化参数经过压测得出的最优配置16核32G环境# Broker端 sendMessageThreadPoolNums32 pullMessageThreadPoolNums32 # 消费者端 pullBatchSize32 # 每次拉取条数 consumeMessageBatchMaxSize10 # 批量消费条数监控指标要重点关注消息堆积量rocketmq_producer_offset消费TPSrocketmq_consumer_tps端到端延迟rocketmq_end_to_end_latency5. 常见问题解决方案5.1 重复消费问题我们最终采用Redis数据库唯一索引的方案// 消费前先查重 if(!redisLock.tryLock(msg:message.getMsgId(), 10, MINUTES)){ return; // 已处理过 } // 数据库唯一约束 ALTER TABLE orders ADD UNIQUE (order_no);5.2 消息堆积处理紧急情况下用这个脚本快速清理积压sh mqadmin resetOffsetByTime \ -n 127.0.0.1:9876 \ -g consumer_group \ -t problem_topic \ -s now长期方案是动态扩容消费者// 根据堆积量自动调整线程数 Scheduled(fixedRate30000) public void adjustThreads() { long backlog getMessageBacklog(); int dynamicThreads Math.min(32, backlog/100 1); container.setConsumeThreadMax(dynamicThreads); }6. 监控与运维6.1 搭建控制台推荐使用官方dashboarddocker run -d --name rmqconsole \ -p 8080:8080 \ -e JAVA_OPTS-Drocketmq.namesrv.addr127.0.0.1:9876 \ apacherocketmq/rocketmq-console:latest6.2 关键告警项我们配置的Prometheus告警规则- alert: RocketMQ消息堆积 expr: rocketmq_producer_offset - rocketmq_consumer_offset 10000 for: 5m - alert: Broker异常 expr: sum(rocketmq_broker_status) by (pod) 07. 进阶实践7.1 消息轨迹追踪在消息头中添加TraceIDMessageBuilder.withPayload(order) .setHeader(traceId, MDC.get(traceId)) .build();通过拦截器实现全链路追踪public class TraceInterceptor implements RocketMQLocalTransactionListener { Override public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) { MDC.put(traceId, msg.getHeaders().get(traceId)); // 业务处理 } }7.2 多环境隔离通过环境变量自动隔离TopicValue(${spring.profiles.active}) private String env; public String buildTopic(String rawTopic) { return env.equals(prod) ? rawTopic : rawTopic _ env; }