RocketMQ 5.3.1 完整部署 + SpringCloud Alibaba 集成与消费幂等案例

一、安装部署

1、安装包下载&&官方教程

### 官方地址，下载慢

wget https://dist.apache.org/repos/dist/release/rocketmq/5.3.1/rocketmq-all-5.3.1-bin-release.zip

### 可以手动去以下网址下载

https://mirrors.aliyun.com/apache/rocketmq/5.3.1

### 本文也提供了5.3.1的安装包

2、linux部署

1. 命令

### 解压 unzip -x rocketmq-all-5.3.1-bin-release.zip 进入脚本目录 cd /home/rocketmq-all-5.3.1-bin-release/bin 启动mqnamesrv nohup sh mqnamesrv & 等mqnamesrv 启动完成后再启动Broker nohup sh mqbroker -n localhost:9876 & 停止 先停Broker sh mqshutdown broker 再停NameServer sh mqshutdown namesrv NameServer 日志 tail -f /home/rocketmq-all-5.3.1-bin-release/logs/rocketmqlogs/namesrv.log Broker 日志 tail -f /home/rocketmq-all-5.3.1-bin-release/logs/rocketmqlogs/broker.log

2、broker启动报错解决

1.内存不够

报错如下，即内存不够，默认需要8G内存

Native memory allocation (mmap) failed to map 8589934592 bytes. There is insufficient memory for the Java Runtime Environment to continue. Command Line: -Xms8g -Xmx8g

修改bin/runbroker.sh启动参数

JAVA_OPT="${JAVA_OPT} -server -Xms512m -Xmx512m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m" # 把15g改为1g JAVA_OPT="${JAVA_OPT} -XX:MaxDirectMemorySize=1g"

2.透明大页 THP 开启 always，Broker 直接退

日志开头打印：/sys/kernel/mm/transparent_hugepage/enabled: [always] madvise never

原因:RocketMQ Broker 强制校验透明大页，always 模式会造成内存碎片、刷盘卡顿，启动拦截。

# 临时生效，服务器重启失效 echo madvise | tee /sys/kernel/mm/transparent_hugepage/enabled echo madvise | tee /sys/kernel/mm/transparent_hugepage/defrag # 校验 cat /sys/kernel/mm/transparent_hugepage/enabled # 备选跳过校验（仅本地测试） nohup sh mqbroker -n 127.0.0.1:9876 -Drocketmq.broker.skipTransparentHugePageCheck=true &

3. windows部署

windows部署依旧跟linux用同一个zip包，只是启动的命令不同而已，启动前需要在环境变量中配置好JAVA_HOME。

### 启动namesrv mqnamesrv.cmd 启动broker mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true

二、集成到springcloudAlibaba

博主用的是5.3.1的rocketmq，引入封装好的消费者与生产者依赖，用2.3.6就可以兼容使用。

1、依赖引入

<!--rocketmq的版本管理--> <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-spring-boot-starter</artifactId> <version>2.3.6</version> </dependency>

2、entity

package com.cloud.zhenyu.entity; import lombok.Data; import java.io.Serializable; /** 测试消息实体 */ @Data public class DemoMsg implements Serializable { private static final long serialVersionUID = 1L; private Long id; private String content; private Long timestamp; private String tag; private String hashKey; }

3、producer

package com.cloud.zhenyu.producer; import com.cloud.zhenyu.entity.DemoMsg; import org.apache.rocketmq.client.producer.SendCallback; import org.apache.rocketmq.client.producer.SendResult; 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.Component; /** RocketMQ 消息生产者 */ @Component public class DemoRocketProducer { @Autowired private RocketMQTemplate rocketMQTemplate; // 主题名称统一常量 public static final String TOPIC_DEMO = "topic-demo"; /** 同步发送（可靠，等待Broker返回结果，业务常用） */ public void sendSyncMsg(DemoMsg msg) { Message<DemoMsg> message = MessageBuilder.withPayload(msg).build(); SendResult sendResult = rocketMQTemplate.syncSend(TOPIC_DEMO, message); System.out.println("同步发送结果：" + sendResult); } /** 异步发送（不阻塞主线程，回调接收结果） */ public void sendAsyncMsg(DemoMsg msg) { Message<DemoMsg> message = MessageBuilder.withPayload(msg).build(); rocketMQTemplate.asyncSend(TOPIC_DEMO, message, new SendCallback() { @Override public void onSuccess(SendResult sendResult) { System.out.println("异步发送成功：" + sendResult); } @Override public void onException(Throwable e) { System.err.println("异步发送失败：" + e.getMessage()); // 此处可做重试、落库补偿逻辑 } }); } /** 单向发送（不等待响应，日志类、不在乎是否送达场景） */ public void sendOneWayMsg(DemoMsg msg) { Message<DemoMsg> message = MessageBuilder.withPayload(msg).build(); rocketMQTemplate.sendOneWay(TOPIC_DEMO, message); } /** 带Tag过滤发送（消费者可根据tag过滤消息） */ public void sendMsgWithTag(DemoMsg msg) { // topic:tag 格式 String destination = TOPIC_DEMO + ":" + msg.getTag(); rocketMQTemplate.convertAndSend(destination, msg); } /** 顺序消息（同一个hashKey的消息会进入同一队列，保证有序） */ public void sendOrderMsg(DemoMsg msg) { rocketMQTemplate.syncSendOrderly(TOPIC_DEMO, msg, msg.getHashKey()); } }

4、consumer

package com.cloud.zhenyu.consumer; import com.cloud.zhenyu.entity.DemoMsg; import com.cloud.zhenyu.producer.DemoRocketProducer; import lombok.extern.slf4j.Slf4j; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.springframework.stereotype.Service; /** 普通消息消费者 consumerGroup：消费者组，同一topic不同业务必须不同组 topic：监听主题 selectorExpression：tag过滤，* 接收全部，tag1||tag2 接收指定tag / @Slf4j @Service @RocketMQMessageListener( consumerGroup = "demo-consumer-group", topic = DemoRocketProducer.TOPIC_DEMO, selectorExpression = "" ) public class DemoRocketConsumer implements RocketMQListener<DemoMsg> { /** 收到消息自动执行，抛出异常会自动重试 */ @Override public void onMessage(DemoMsg message) { log.info("消费者收到消息：{}", message); // 业务处理逻辑 // 若抛出异常，RocketMQ会重试投递；正常执行无异常则标记消费成功 } }

5、配置文件

rocketmq: # NameServer 地址，多个用分号分隔 name-server: 127.0.0.1:9876 # 生产者配置 producer: # 生产者组（同业务统一组名） group: demo-producer-group # 发送消息超时时间 send-message-timeout: 3000 # 消费者统一配置可在这里，也可注解单独指定

三、

三、本地测试说明

博主本来是在阿里云部署测试玩的，但是Rocketmq初始内存就要8G，因为囊中羞涩所以只有将启动参数改到512M，但是后续测试发现用本地java连接云服务器的mq会出现消费不了的异常，原因如下：

Broker 注册给 NameServer 的是服务器内网 IP（172.x.x.x），Windows 本地无法访问内网网段。生产者仅连 9876 可发消息；消费者拉取消息需连接 Broker 10911 内网地址，直接失败，表现：发送成功、消费无响应 / 启动抛连接 null 异常。

所以诸多麻烦下，建议自己玩玩的话还是用windows本机把。

四、消费者幂等与并发安全教学案例

1、案例简介

1.1 业务背景

基于水务账单支付业务，用户可在线缴纳水费账单，支付完成后需完成三个核心操作：
1.生成缴费流水记录
2.更新账单状态为【已缴费】
3.多余支付金额自动充值至用户账户余额
业务特性：账单为系统预生成（每期账单自动生成，未支付时已存在唯一账单号），支持多端登录、多人同账号操作、MQ异步消费，存在极高并发数据风险。

1.2 线上高频BUG场景

1.同一账单多人并发重复支付：多人登录同一账号，同时支付同一未缴费账单，生成不同支付单号的MQ消息，导致重复扣款、重复生成流水。
2.用户余额并发覆盖错乱：同一用户同时触发缴费充值、退款扣费等操作，代码查询余额后修改，出现脏写覆盖，资金对账错乱。

2、解决方案

采用「Redis分布式锁前置拦截 + 数据库原子行锁终极兜底」双层架构，兼顾性能与绝对数据安全，适用于所有支付、充值、退费等资金类MQ消费场景。

第一层：Redis分布式锁（性能优化，拦截99%并发）
•锁维度：以账单唯一号 billNo 加锁（核心业务主体）
•锁逻辑：同一账单同一时间仅允许一条MQ消息执行业务
•作用：拦截瞬时并发重复请求，避免大量无效数据库事务，提升消费性能
第二层：数据库原子行锁（终极兜底，杜绝数据BUG）
解决Redis宕机、锁超时、主从切换等极端锁失效场景，从数据库层面保证数据一致性，包含两类原子锁：
1.账单状态行锁：仅未支付账单可更新状态，杜绝重复支付
2.余额原子加减锁：不查询余额，直接数据库字段运算，彻底解决并发覆盖错乱

3、完整执行流程

1.消费者监听支付MQ消息，解析核心参数：账单号billNo、用户ID、多余充值金额
2.根据billNo尝试获取Redis分布式锁，加锁失败则重试消息，终止当前流程
3.加锁成功，开启数据库本地事务
4.执行账单原子状态更新，仅未支付账单更新成功，已支付账单直接终止流程
5.账单更新成功，生成缴费流水记录
6.执行用户余额原子加减操作，处理多余充值金额
7.事务整体提交，业务执行成功，手动ACK确认消息
8.捕获异常实现重试机制，超限消息转入死信队列
9.finally释放分布式锁，避免死锁

4、代码实现

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.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.stereotype.Component; import org.springframework.transaction.support.TransactionTemplate; import com.alibaba.fastjson.JSON; import lombok.extern.slf4j.Slf4j; import javax.annotation.Resource; import java.nio.charset.StandardCharsets; import java.util.List; import java.util.concurrent.TimeUnit; /** 账单支付MQ消费者 解决：多人并发重复支付、MQ重复消费、用户余额并发错乱 架构：Redis账单锁前置拦截 + 数据库原子行锁兜底 / @Slf4j @Component @RocketMQMessageListener( topic = "topic_pay_notify", consumerGroup = "pay-consumer-group", selectorExpression = "" ) public class PayConsumer implements MessageListenerConcurrently, RocketMQPushConsumerLifecycleListener { private DefaultMQPushConsumer consumer; // 分布式锁过期时间（大于业务最大执行时长，防止业务未完成锁过期） private static final long LOCK_EXPIRE_TIME = 30; @Resource private RedisTemplate<String, String> redisTemplate; @Resource private TransactionTemplate transactionTemplate; @Resource private PayFlowService flowService; @Resource private BillService billService; @Resource private UserBalanceService balanceService; /** 初始化开启手动ACK */ @Override public void prepareStart(DefaultMQPushConsumer consumer) { this.consumer = consumer; // 关闭自动提交，业务成功后手动ACK，防止丢消息、重复消费异常 consumer.setManualCommit(true); } @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) { MessageExt messageExt = list.get(0); String billNo = null; try { // 解析MQ消息 String body = new String(messageExt.getBody(), StandardCharsets.UTF_8); PayDTO payDTO = JSON.parseObject(body, PayDTO.class); billNo = payDTO.getBillNo(); String lockKey = "pay:bill:lock:" + billNo; // 1.前置拦截：账单维度分布式锁，防止同一账单并发支付 boolean lockSuccess = redisTemplate.opsForValue() .setIfAbsent(lockKey, "PROCESSING", LOCK_EXPIRE_TIME, TimeUnit.SECONDS); if (!lockSuccess) { log.info("账单并发处理中，稍后重试，billNo={}", billNo); return ConsumeConcurrentlyStatus.RECONSUME_LATER; } // 2.事务执行全套支付业务 Boolean processSuccess = transactionTemplate.execute(status -&gt; { // 2.1 数据库原子更新账单状态：仅未支付账单可更新（行锁防重复支付） int updateRows = billService.updateUnpaidBillStatus(billNo); if (updateRows &lt;= 0) { // 账单已支付，无需重复处理 return false; } // 2.2 生成缴费流水记录 flowService.createPayFlow(payDTO); // 2.3 原子更新用户余额，彻底解决并发错乱 if (payDTO.getSurplusAmount() &gt; 0) { balanceService.addUserBalance(payDTO.getUserId(), payDTO.getSurplusAmount()); } return true; }); // 3.根据业务结果确认消息 if (processSuccess) { consumer.ack(messageExt); log.info("账单支付消费成功，billNo={}", billNo); } else { consumer.ack(messageExt); log.warn("账单已支付，过滤重复MQ消息，billNo={}", billNo); } } catch (Exception e) { // 异常重试+死信队列兜底 int retryTimes = messageExt.getReconsumeTimes(); log.error("账单消费异常，billNo={}，重试次数={}", billNo, retryTimes, e); if (retryTimes &gt;= 3) { // 重试3次失败，转入死信队列，避免阻塞正常队列 consumer.sendToDeadLetter(messageExt); } else { return ConsumeConcurrentlyStatus.RECONSUME_LATER; } } finally { // 释放分布式锁，防止死锁 if (billNo != null) { redisTemplate.delete("pay:bill:lock:" + billNo); } } return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }

5、核心原子SQL

### 防重复支付：账单状态原子更新 UPDATE t_bill SET pay_status = 1, pay_time = NOW() WHERE bill_no = #{billNo} AND pay_status = 0; 防余额错乱：用户余额原子加减（直接字段运算，规避查询赋值导致的并发覆盖问题，支持缴费、退款、充值并发操作） -- 余额增加（多余缴费、账户充值场景） UPDATE t_user SET balance = balance + #{amount} WHERE user_id = #{userId}; -- 余额扣减（水费扣费、退款场景） UPDATE t_user SET balance = balance - #{amount} WHERE user_id = #{userId} AND balance >= #{amount};

6、总结

1. 幂等维度选型核心：资金、订单、账单类业务，禁止用动态生成的流水号（payNo）做幂等，必须用业务固有唯一主键（billNo/orderNo）。
2. 并发分层思想：Redis锁负责高性能前置拦截，数据库原子锁负责最终数据兜底，二者缺一不可。
3. 余额并发铁律：所有金额增减操作，禁止查询后赋值，必须使用数据库原子加减SQL。
4. MQ可靠性规范：核心资金业务必须开启手动ACK，业务成功再提交，异常自动重试，超限转入死信队列。
5. 杜绝伪幂等：放弃「先查询后写入」的写法，无法抵御高并发竞态问题。