多来买：Lua 库存与 RocketMQ 事务消息实战#

本篇范围：事务消息发送、Redis Lua 原子扣库存、本地事务监听器、Broker 回查与业务结果映射
事实来源：当前 PromoServiceImpl、PromoTransactionProducer、RedisStockOper 及 MQ 常量
验证口径：代码已确认；未连接 Redis 或 RocketMQ，未验证消息提交、回查和重试行为

本篇聚焦秒杀链路最核心的协调段：Producer 先发 Half Message，再把 Redis Lua 扣库存作为生产者侧“本地事务”，最后依据库存结果提交或回滚消息；当结果不确定时，Broker 根据 transactionId 对 Redis 标记进行回查。

事务消息主链路#

sequenceDiagram participant Promo as "promo-service" participant Producer as "事务消息 Producer" participant Broker as "RocketMQ Broker" participant Redis as "Redis Lua / 状态 Bucket" Promo->>Producer: sendMessageInTransaction Producer->>Broker: Half Message Broker-->>Producer: 执行本地事务 Producer->>Redis: Lua 检查并扣库存 alt 库存充足 Producer->>Redis: transactionId = success Producer-->>Broker: COMMIT_MESSAGE Broker-->>Promo: 发送结果可观察为成功 else 库存不足 Producer->>Redis: transactionId = fail Producer-->>Broker: ROLLBACK_MESSAGE else 结果不确定 Producer-->>Broker: UNKNOW Broker->>Producer: checkLocalTransaction Producer->>Redis: 查询 transactionId 状态 end

阅读目标#

说明 RocketMQ Half Message、提交、回滚和回查四个阶段。
说明这里的“本地事务”是 Redis Lua，而不是 MySQL 事务。
说明 Lua 为什么能原子完成库存检查与扣减。
解释 transactionId Bucket 怎样支持 Broker 回查。
区分“事务消息提交成功”与“订单消费者已经落库”。

第七步：发送 RocketMQ 事务消息#

代码路径：duolaimall-promo/promo-service/src/main/java/com/cskaoyan/mall/promo/service/impl/PromoServiceImpl.java
关键方法：submitOrderInTransaction

1
Long skuId = orderInfoParam
2
        .getOrderDetailList()
3
        .get(0)
4
        .getSkuId();
5

6
SeckillGoodsDTO goods =
7
        getSeckillGoodsDTO(skuId);
8

9
if (goods == null) {
10
    throw new BusinessException(
11
            SeckillCodeEnum.SECKILL_ILLEGAL);
12
}
13

14
HashMap<String, Object> localParams =
15
        new HashMap<>();
16
localParams.put("id", goods.getId());
17
localParams.put("stock", 1);
18
localParams.put("skuId", skuId);
19

20
MqResultEnum result =
21
        promoTransactionProducer
22
                .sendTransactionMessage(
23
                        MqTopicConst
24
                                .PROMO_ORDER_TOPIC,
25
                        orderInfoParam,
26
                        localParams);

消息 body 是 OrderInfoParam JSON，本地事务参数包含 SKU、活动 ID 和扣减数量。活动 ID 在当前 Lua 分支没有使用，属于从数据库扣库存方案保留下来的参数。

发送结果分支#

1
if (MqResultEnum.SEND_FAIL
2
        .equals(result)) {
3
    RSet<Long> set =
4
            redissonClient.getSet(
5
                    RedisConst
6
                            .PROMO_USER_ORDERED_FLAG
7
                    + orderInfoParam.getUserId());
8

9
    set.remove(skuId);
10

11
    throw new BusinessException(
12
            SeckillCodeEnum
13
                    .SECKILL_ORDER_TRY_AGAIN);
14
}
15

16
if (MqResultEnum.LOCAL_TRANSACTION_FAIL
17
        .equals(result)) {
18
    throw new BusinessException(
19
            SeckillCodeEnum.SECKILL_FINISH);
20
}
21

22
if (MqResultEnum.LOCAL_TRANSACTION_EXCEPTION
23
        .equals(result)) {
24
    throw new BusinessException(
25
            SeckillCodeEnum
26
                    .SECKILL_ORDER_TRY_AGAIN);
27
}

只有 SEND_FAIL 显式删除用户 Set 标记。库存不足和本地事务未知分支没有删除：

库存确实不足时保留标记影响不大；
结果未知但实际未扣库存时，用户重试会被当成重复抢购。

Controller 在方法正常返回后响应“正在排队”，而不是“下单成功”。

RocketMQ 事务消息的真实语义#

四个阶段#

1
1. Producer 发送 Half Message
2
2. Broker 暂不向消费者投递
3
3. Producer 执行本地事务：Redis Lua 扣库存
4
4. Producer 返回：
5
   COMMIT   -> 消息对订单消费者可见
6
   ROLLBACK -> Broker 删除半消息
7
   UNKNOWN  -> Broker 稍后回查

当前所谓“本地事务”不是 MySQL 事务，而是 Producer 回调里的一次 Redis Lua 操作，以及把结果写入 Redis Bucket。

Producer 初始化#

代码路径：duolaimall-promo/promo-service/src/main/java/com/cskaoyan/mall/promo/mq/PromoTransactionProducer.java
关键方法：init

1
@PostConstruct
2
public void init()
3
        throws MQClientException {
4

5
    transactionMQProducer =
6
            new TransactionMQProducer(
7
                    "tx-promo-service-group");
8

9
    // 注册中心地址省略，不进入文档
10
    transactionMQProducer
11
            .setTransactionListener(
12
                    new TransactionListener() {
13
        // executeLocalTransaction
14
        // checkLocalTransaction
15
    });
16

17
    transactionMQProducer.start();
18
}

生产者会随 Spring Bean 初始化启动。源码中注册中心地址是硬编码；订单消费者使用另一个硬编码值，两端不一致，这是 P0 级配置风险，但文档不回显具体地址。

第八步：Lua 原子检查并扣库存#

代码路径：duolaimall-promo/promo-service/src/main/java/com/cskaoyan/mall/promo/redis/RedisStockOper.java

当前 Lua#

1
local current =
2
    redis.call('hget', KEYS[1], ARGV[1])
3

4
if not current
5
    or tonumber(current)
6
       < tonumber(ARGV[2]) then
7
    return -1
8
end
9

10
return redis.call(
11
    'hincrby',
12
    KEYS[1],
13
    ARGV[1],
14
    -tonumber(ARGV[2]))

参数映射：

Lua 参数	当前值
`KEYS[1]`	`promo:goods:stock`
`ARGV[1]`	`skuId`
`ARGV[2]`	本次扣减数量，当前为 1

为什么不能拆成 Java 三步#

错误的非原子思路：

1
stock = HGET(...)
2
if stock >= 1:
3
    HINCRBY(..., -1)

两个线程可以同时读到 1，都通过判断，然后各扣一次。Lua 在 Redis 服务端一次执行，脚本运行期间不会与另一条命令交错，因此“检查 + 扣减”成为一个原子操作。

加载和执行 SHA#

1
@PostConstruct
2
public void loadScript() {
3
    sha1 = redissonClient
4
            .getScript()
5
            .scriptLoad(STOCK_LUA);
6
}
7

8
public Long decrRedisStock(
9
        Long skuId,
10
        Integer count) {
11
    return redissonClient
12
            .getScript(StringCodec.INSTANCE)
13
            .evalSha(
14
                RScript.Mode.READ_WRITE,
15
                sha1,
16
                RScript.ReturnType.INTEGER,
17
                List.of(
18
                    RedisConst
19
                        .PROMO_SECKILL_GOODS_STOCK),
20
                skuId.toString(),
21
                count.toString());
22
}

脚本启动时加载，调用时通过 SHA 执行。当前没有看到 NOSCRIPT 后自动重新加载并重试的分支；Redis 脚本缓存被清理时，影响需要运行验证。

第九步：事务监听器记录库存结果#

代码路径：duolaimall-promo/promo-service/src/main/java/com/cskaoyan/mall/promo/mq/PromoTransactionProducer.java
关键方法：executeLocalTransaction

扣库存#

1
Long skuId =
2
        (Long) localParams.get("skuId");
3
Integer stock =
4
        (Integer) localParams.get("stock");
5

6
String transactionId =
7
        msg.getTransactionId();
8

9
RBucket<String> stateBucket =
10
        redissonClient.getBucket(
11
                RedisConst
12
                        .PROMO_TRANSACTION_PREFIX
13
                + transactionId);
14

15
Long remainStock =
16
        redisStockOper
17
                .decrRedisStock(
18
                        skuId, stock);

决定提交或回滚#

1
if (remainStock.intValue() < 0) {
2
    stateBucket.set("fail");
3

4
    LocalCacheHelper.put(
5
            skuId.toString(),
6
            LocalStockStatus
7
                    .STOCK_NOT_ENOUGH
8
                    .getNum());
9

10
    return LocalTransactionState
11
            .ROLLBACK_MESSAGE;
12
}
13

14
stateBucket.set("success");
15

16
return LocalTransactionState
17
        .COMMIT_MESSAGE;

Lua 返回：

-1：库存不存在或不足，回滚消息；
0：刚好扣完，提交消息；
大于 0：仍有库存，提交消息。

本地售罄状态更新慢一拍#

代码中“remainStock == 0 时标记售罄”的分支被注释。当前只有下一次请求执行 Lua 并得到 -1 时，才把本地状态改成无库存。

这不会让 Redis 库存变负，因为 Lua 仍会拒绝；但会让每个实例在库存刚好归零后多放过至少一次本地检查，产生额外 Redis/MQ 事务请求。

第十步：Broker 回查本地事务#

1
@Override
2
public LocalTransactionState
3
checkLocalTransaction(MessageExt msg) {
4

5
    String transactionId =
6
            msg.getTransactionId();
7

8
    RBucket<String> bucket =
9
            redissonClient.getBucket(
10
                    RedisConst
11
                            .PROMO_TRANSACTION_PREFIX
12
                    + transactionId);
13

14
    String value = bucket.get();
15

16
    if ("success".equals(value)) {
17
        return LocalTransactionState
18
                .COMMIT_MESSAGE;
19
    }
20

21
    if ("fail".equals(value)) {
22
        return LocalTransactionState
23
                .ROLLBACK_MESSAGE;
24
    }
25

26
    return LocalTransactionState.UNKNOW;
27
}

当 Producer 执行完库存逻辑却没把结果成功告诉 Broker，Broker 会回查：

Redis 为 success：提交；
Redis 为 fail：回滚；
key 不存在或值未知：继续返回 UNKNOW。

事务状态 Bucket 没有 TTL。它便于回查，却也会长期积累；活动清理的 promo:* 删除如果与 Broker 回查重叠，也可能让已知结果变成未知。

第十一步：把事务发送结果映射为业务结果#

代码路径：duolaimall-promo/promo-service/src/main/java/com/cskaoyan/mall/promo/mq/PromoTransactionProducer.java
关键方法：sendTransactionMessage

1
TransactionSendResult sendResult =
2
        transactionMQProducer
3
                .sendMessageInTransaction(
4
                        message,
5
                        localTransactionParam);
6

7
SendStatus sendStatus =
8
        sendResult.getSendStatus();
9

10
LocalTransactionState localState =
11
        sendResult
12
                .getLocalTransactionState();
13

14
if (!SendStatus.SEND_OK
15
        .equals(sendStatus)) {
16
    return MqResultEnum.SEND_FAIL;
17
}
18

19
if (LocalTransactionState
20
        .ROLLBACK_MESSAGE
21
        .equals(localState)) {
22
    return MqResultEnum
23
            .LOCAL_TRANSACTION_FAIL;
24
}
25

26
if (LocalTransactionState
27
        .COMMIT_MESSAGE
28
        .equals(localState)) {
29
    return MqResultEnum
30
            .Local_TRANSACTION_SUCCESS;
31
}
32

33
return MqResultEnum
34
        .LOCAL_TRANSACTION_EXCEPTION;

Java 方法返回“本次发送和本地事务观察结果”，不代表订单消费者已经完成。即使是 Local_TRANSACTION_SUCCESS，订单消息仍在异步投递途中。

本篇限制与风险收束#

Lua 只保护一个 Redis 库存 Hash，不解决用户重复、订单落库和库存回补。
Lua 扣库存与写 transactionId 结果是两次 Redis 操作，中间存在进程退出窗口。
transactionId Bucket 没有 TTL，清理时机依赖活动清理入口。
Producer 与 Consumer 的硬编码地址当前不一致，消息是否到达同一 Broker 只能标记为未验证。
方法返回事务发送结果，只说明消息侧决策，不代表下游订单已经创建。

关键源码导航#

阅读顺序	文件	关键方法
1	`duolaimall-promo/promo-service/src/main/java/com/cskaoyan/mall/promo/service/impl/PromoServiceImpl.java`	`submitOrderInTransaction`
2	`duolaimall-promo/promo-service/src/main/java/com/cskaoyan/mall/promo/mq/PromoTransactionProducer.java`	`executeLocalTransaction`、`checkLocalTransaction`
3	`duolaimall-promo/promo-service/src/main/java/com/cskaoyan/mall/promo/redis/RedisStockOper.java`	`decrRedisStock`
4	`duolaimall-common/common-mq/src/main/java/com/cskaoyan/mall/mq/constant/MqTopicConst.java`	`PROMO_ORDER_TOPIC`
5	`duolaimall-common/common-service/src/main/java/com/cskaoyan/mall/common/constant/RedisConst.java`	事务回查状态 key

4 条短复习点#

Lua 保证 Redis 内“查库存 + 扣库存”不可被其他命令插入。
RocketMQ 事务消息根据生产者侧库存结果决定订单消息是否可见。
Broker 回查依赖 Redis 中 transactionId 对应的业务结果。
消息 COMMIT 仍不等于订单落库，消费端还需要幂等和重试。

本章总结#

Lua 保证 Redis 库存检查与扣减的单实例原子性，RocketMQ 事务消息据此决定订单消息是否可见；两者没有覆盖消费建单和补偿，因此不能表述为全链路强一致。

下一篇：秒杀订单消费、轮询与补偿边界实战。

管理主页分类