08 · 消息队列与账务核心（RabbitMQ）

本篇讲本项目中「资金变动」是怎么处理的：充值、转账、购买扣款，全部建模成命令（Command）经由 RabbitMQ 异步串行处理，模拟银行核心系统的顺序记账。读完本篇你应该能回答：为什么资金操作要异步、消息队列里「交换机 / 队列 / 绑定」分别是什么、单消费者串行处理如何避免余额错乱、以及 HTTP 层与账务层是怎么分工的。

---

8.1 为什么资金变动要走消息队列（概念）

先看一个不用消息队列的「同步」写法：

POST /api/wallet/transfer
  -> Controller 校验参数
  -> Service 直接：from.balance -= amount; to.balance += amount; 写流水;   ← 都在 HTTP 请求线程里完成
  -> 返回成功

这种方式的问题：

1. 账务是全局热点：余额属于「账户」，所有用户的转账 / 充值 / 购买都改账户表。如果多个请求并发改同一个账户，又不做严格加锁，会出现经典的「超卖 / 负余额 / 少加钱」问题——只读一次余额、判断够、扣减，但两个请求同时读到旧余额就都通过了。
2. HTTP 请求被账务拖长：转账、扣款涉及多张表、还要写流水，全卡在用户请求里，响应慢、连接池易被占满（见 02 篇 2.7 关于「事务内不要做长耗时 IO」的坑）。
3. 缺少削峰与解耦：大促下单洪峰时，账务系统应该按自己的节奏处理，而不是被瞬时请求压垮。

消息队列的思路是：HTTP 层只做「校验参数 + 落单 + 发命令」，真正的资金改动交给后台一个消费者慢慢串行处理。

HTTP 请求线程                          后台消费者线程（单一）
─────────────                         ─────────────────────
校验参数                              从队列取一条 MoneyCommand
扣减库存 / 创建订单                     按命令类型改账户余额
发布 MoneyCommand ──► 队列 ──►        写交易流水
立即返回「PENDING / 已受理」           更新订单状态 PAID / FAILED

概念：这就是生产者-消费者（Producer-Consumer）模型与异步解耦。HTTP 层是「生产者」，只负责把任务（命令）丢进队列；后台「消费者」按队列顺序一条条处理。队列像一根管道，把快慢两端隔开。

本项目把消费者设计成单消费者 + 串行处理，专门模拟「银行核心系统」：所有账务命令排成一条队，挨个顺序记账，从根本上杜绝了并发改同一账户的竞态（见 8.6）。这是教学项目里刻意做的简化与示意——真实银行核心会有更完善的分布式事务 / 幂等方案。

---

8.2 RabbitMQ 的核心概念：交换机 / 队列 / 绑定

RabbitMQ 不是「生产者直接把消息丢给消费者」，中间有一个交换机（Exchange）做路由：

Producer ──publish──► Exchange(交换机)
                         │ 按 routing key 路由
                         ▼
                       Queue(队列)  ◄── Consumer 在这里监听

概念	类比	本项目
Producer（生产者）	发件人	MoneyQueueProducer
Exchange（交换机）	邮筒 / 分拣中心	uas.exchange（direct 类型）
Routing Key（路由键）	收件地址	uas.money
Queue（队列）	信箱（真正存消息的地方）	uas.money.queue
Binding（绑定）	把队列「挂」到交换机上、并声明关心哪个路由键	moneyBinding()
Consumer（消费者）	收件人	MoneyQueueConsumer（@RabbitListener）

为什么要多一个 Exchange？ 直接发队列也行，但 Exchange 提供「一对多路由」的灵活性：一条消息可以按 routing key 被复制分发到多个队列（扇出、主题订阅等）。本项目用最简单的 direct 交换机：routing key 精确匹配，消息只进绑定的那一个队列。账务命令只有一条队列，direct 足够且语义清晰。

---

8.3 配置：RabbitConfig（声明交换机 / 队列 / 绑定）

// backend/src/main/java/com/yang/backend/config/RabbitConfig.java
@Configuration
public class RabbitConfig {

    public static final String EXCHANGE = "uas.exchange";
    public static final String QUEUE = "uas.money.queue";
    public static final String ROUTING = "uas.money";

    @Bean
    public DirectExchange moneyExchange() {
        return new DirectExchange(EXCHANGE, true, false);   // durable=true, autoDelete=false
    }

    @Bean
    public Queue moneyQueue() {
        return new Queue(QUEUE, true);                       // durable=true
    }

    @Bean
    public Binding moneyBinding() {
        return BindingBuilder.bind(moneyQueue()).to(moneyExchange()).with(ROUTING);
    }

    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory,
                                         MessageConverter messageConverter) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(messageConverter);
        return rabbitTemplate;
    }
}

要点：

• DirectExchange(EXCHANGE, true, false)：第二个参数 durable=true 表示交换机持久化——RabbitMQ 重启后这个交换机还在；第三个 autoDelete=false 表示没有消费者连接时也不要自动删除。
• new Queue(QUEUE, true)：队列同样设为持久化。队列持久化 + 消息持久化（默认）意味着即使 RabbitMQ 重启，没处理完的命令也不会丢——这对应「账务不能丢命令」的强要求。
• BindingBuilder.bind(queue).to(exchange).with(ROUTING)：把队列挂到交换机上，并声明「只收 routing key = uas.money 的消息」。一条队列只绑定一个 routing key，direct 交换机的精确匹配正合适。
• Jackson2JsonMessageConverter：本项目消息体是 Java 对象（MoneyCommand），这个转换器用 Jackson 把它序列化 / 反序列化成 JSON 在线路上传输。没有它，RabbitTemplate 默认用 Java 原生序列化（可读性差、跨语言不行）。
• RabbitTemplate：生产者发消息用的「发送客户端」，构造时把 JSON 转换器装进去。

坑：Exchange / Queue / Binding 这三个 Bean 必须在应用启动时被 Spring 创建出来（它们是 @Bean）。如果忘了声明队列，消费者 @RabbitListener 监听一个不存在的队列会报错；如果队列没设 durable=true，RabbitMQ 重启后队列消失、堆积的命令随之丢失。账务场景务必持久化。

---

8.4 消息体：MoneyCommand（record）

消息本身是一个不可变的数据对象，描述「要做什么资金操作、涉及谁、多少钱」：

// backend/src/main/java/com/yang/backend/mq/MoneyCommand.java
public record MoneyCommand(
        String type,        // 命令类型：DEPOSIT / TRANSFER / PURCHASE
        Long fromUserId,    // 转出方（充值/购买可为 null）
        Long toUserId,      // 转入方（转账/充值的收款人；购买可为 null）
        BigDecimal amount,  // 金额
        Long orderId,       // 关联订单（仅 PURCHASE 用）
        String reference    // 业务备注，如 "transfer" / "deposit" / 订单号
) {
    public static final String DEPOSIT = "DEPOSIT";
    public static final String TRANSFER = "TRANSFER";
    public static final String PURCHASE = "PURCHASE";
}

要点：

• 用 record（Java 17）而不是普通 class：消息是「纯数据」，不需要可变状态，record 自动生成 final 字段、getter、全参构造、equals/hashCode/toString，简洁且不可变。
• BigDecimal 而非 double：金额必须用 BigDecimal，double/float 有二进制浮点误差（0.1 + 0.2 ≠ 0.3），会算错钱。数据库里 account.balance / transaction_log.amount 也是 NUMERIC(12,2)，和 BigDecimal 一一对应。
• type 用一个字符串常量区分三种命令，消费者拿到后 switch 分发（见 8.6）。

概念：命令模式（Command Pattern）——把「一个要执行的操作」连同它的参数打包成一个对象，放进队列，之后由别的地方取出来执行。好处是「发命令的人」和「执行命令的人」完全解耦，命令还能排队、重试、持久化。

---

8.5 生产者：MoneyQueueProducer

// backend/src/main/java/com/yang/backend/service/MoneyQueueProducer.java
@Service
public class MoneyQueueProducer {

    private final RabbitTemplate rabbitTemplate;

    public MoneyQueueProducer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    public void publish(MoneyCommand command) {
        rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE, RabbitConfig.ROUTING, command);
    }
}

convertAndSend(exchange, routingKey, object) 一步完成「JSON 序列化 + 按 routing key 发给交换机」。注意它用的 EXCHANGE 和 ROUTING 都是 RabbitConfig 里定义的常量，和消费者监听的队列靠同一个 ROUTING 对应起来——生产者只关心「发到哪个交换机、用什么路由键」，完全不知道队列和消费者长什么样，这正是解耦。

真实调用点（HTTP 层）：

// WalletService.java:84  转账
moneyQueueProducer.publish(new MoneyCommand(
        MoneyCommand.TRANSFER, fromUserId, toUser.getId(), amount, null, "transfer"));

// WalletService.java:93  充值
moneyQueueProducer.publish(new MoneyCommand(
        MoneyCommand.DEPOSIT, null, targetUserId, amount, null, "deposit"));

// OrderService.java:105  下单购买
moneyQueueProducer.publish(new MoneyCommand(
        MoneyCommand.PURCHASE, userId, null, total, order.getId(), String.valueOf(order.getId())));

注意：HTTP 层在发布命令之前，只做「参数校验 + 必要落单」，并未真正改余额。例如 OrderService.checkout 先扣库存、创建 PENDING 订单、清空购物车，最后才 publish(PURCHASE)，然后立刻返回「订单已受理」。余额扣减和订单置 PAID 完全发生在后台消费者里。

---

8.6 消费者：MoneyQueueConsumer（串行账务核心）

这是「银行核心」的真正执行者：

// backend/src/main/java/com/yang/backend/service/MoneyQueueConsumer.java
@Service
public class MoneyQueueConsumer {

    @RabbitListener(queues = RabbitConfig.QUEUE)
    @Transactional
    public void handle(MoneyCommand command) {
        switch (command.type()) {
            case MoneyCommand.DEPOSIT  -> handleDeposit(command);
            case MoneyCommand.TRANSFER -> handleTransfer(command);
            case MoneyCommand.PURCHASE -> handlePurchase(command);
            default -> { }
        }
    }
    // ...
}

8.6.1 注解拆解

• @RabbitListener(queues = RabbitConfig.QUEUE)：声明「我是这个队列的消费者」。Spring AMQP 会起一个监听容器，只要有消息进队列，就调 handle(...)。本项目只有一个消费者实例、且默认单线程处理（没有配并发消费者），所以队列里的命令是一条接一条顺序执行的——这是「串行记账、避免并发改余额」的关键。
• @Transactional：处理方法整体在一个数据库事务里。账务改动（改余额、写流水）要么全成功、要么回滚，保证原子性。若处理中途抛异常，事务回滚，并且（默认配置下）消息会被重新投递、重试——不会「钱扣了但流水没写」。

8.6.2 三种命令的处理逻辑

DEPOSIT（充值）——给目标账户加钱：

private void handleDeposit(MoneyCommand command) {
    Account account = walletService.getOrCreateAccount(command.toUserId());
    account.setBalance(account.getBalance().add(command.amount()));
    accountMapper.updateById(account);
    writeLog(MoneyCommand.DEPOSIT, null, account.getUserId(), command.amount(),
            account.getBalance(), command.reference(), "SUCCESS", null);
}

TRANSFER（转账）——先校验余额，不足则写 FAILED 流水并不做任何改动：

private void handleTransfer(MoneyCommand command) {
    Account from = walletService.getOrCreateAccount(command.fromUserId());
    Account to   = walletService.getOrCreateAccount(command.toUserId());
    if (from.getBalance().compareTo(command.amount()) < 0) {
        writeLog(MoneyCommand.TRANSFER, from.getUserId(), to.getUserId(), command.amount(),
                from.getBalance(), command.reference(), "FAILED", "余额不足");
        return;                                            // 余额不足：不扣不减，只留失败流水
    }
    from.setBalance(from.getBalance().subtract(command.amount()));
    to.setBalance(to.getBalance().add(command.amount()));
    accountMapper.updateById(from);
    accountMapper.updateById(to);
    writeLog(MoneyCommand.TRANSFER, from.getUserId(), to.getUserId(), command.amount(),
            from.getBalance(), command.reference(), "SUCCESS", null);
}

PURCHASE（购买扣款）——扣买家余额，成功置订单 PAID；余额不足则订单置 FAILED、恢复库存、写 FAILED 流水：

private void handlePurchase(MoneyCommand command) {
    Account account = walletService.getOrCreateAccount(command.fromUserId());
    Order order = orderMapper.selectById(command.orderId());
    if (order == null || !"PENDING".equals(order.getStatus())) {
        return;                                            // 订单不存在或已处理过，直接忽略
    }
    if (account.getBalance().compareTo(command.amount()) < 0) {
        order.setStatus("FAILED");
        orderMapper.updateById(order);
        restoreStock(order.getId());                      // 把之前扣掉的库存加回去
        writeLog(MoneyCommand.PURCHASE, account.getUserId(), null, command.amount(),
                account.getBalance(), String.valueOf(order.getId()), "FAILED", "余额不足");
        return;
    }
    account.setBalance(account.getBalance().subtract(command.amount()));
    accountMapper.updateById(account);
    order.setStatus("PAID");
    orderMapper.updateById(order);
    writeLog(MoneyCommand.PURCHASE, account.getUserId(), null, command.amount(),
            account.getBalance(), String.valueOf(order.getId()), "SUCCESS", null);
}

8.6.3 账户懒创建

walletService.getOrCreateAccount(userId)（WalletService.java:39）先在 account 表按 user_id 查，没有就新建一行初始余额 0。这样管理员无需预先给用户开账户——首次转账 / 充值 / 下单时自动开户，账户表与用户主表解耦（account.user_id 是唯一外键，见 schema.sql）。

8.6.4 库存回滚

restoreStock 在 PURCHASE 余额不足时被调用，把订单里的每个商品数量加回 product.stock，避免「扣了库存却没付钱、商品凭空消失」：

private void restoreStock(Long orderId) {
    List<OrderItem> items = orderItemMapper.selectList(
            new LambdaQueryWrapper<OrderItem>().eq(OrderItem::getOrderId, orderId));
    for (OrderItem item : items) {
        Product product = productMapper.selectById(item.getProductId());
        if (product != null) {
            product.setStock(product.getStock() + item.getQuantity());
            productMapper.updateById(product);
        }
    }
}

8.6.5 流水（审计轨迹）

writeLog(...) 把每一笔资金变动写进 transaction_log，记录类型、收付款方、金额、变动后余额、关联单号、状态（SUCCESS/FAILED）与失败原因。前端查「交易流水」就是读这张表（WalletController → WalletService.listTransactions，按 from_user_id 或 to_user_id 匹配当前用户）。

---

8.7 端到端流程：一次转账发生了什么

① 浏览器 POST /api/wallet/transfer {toUsername:"alice", amount:50}
      │
② WalletController → permissionGuard.require("wallet:transfer")   ← 权限校验（见 03 篇）
      │
③ WalletService.transfer(...)
      │   - 校验 amount>0、目标用户存在、不能转给自己、当前余额够
      │   - moneyQueueProducer.publish(new MoneyCommand(TRANSFER, from, to, 50, ...))
      │
④ RabbitTemplate 把命令 JSON 发给 uas.exchange，路由键 uas.money
      │
⑤ 交换机把消息放进 uas.money.queue
      │
⑥ 【HTTP 请求到此返回 200，余额还没变】

──── 异步边界 ────

⑦ 后台 MoneyQueueConsumer.handle 取到命令（串行、单线程）
      │   - @Transactional 开启事务
      │   - getOrCreateAccount(from/to)
      │   - 校验余额 → 扣 from、加 to、两个 updateById
      │   - writeLog(SUCCESS)
⑧ 事务提交，余额和流水落库

用户随后查余额 / 查交易流水 / 查订单状态，看到的是第 ⑦⑧ 步的结果。这就是「先受理、后结算」的异步模型。

---

8.8 为什么「单消费者串行」能避免余额错乱

余额错乱的本质是竞态（Race Condition）：两个转账请求同时读到同一余额 100，各自认为够扣 80，结果都扣成功，余额变成 20 而不是正确的 -60/拒绝。

把消费者设成单线程、串行处理队列后，命令被强制排成一条线：

队列: [TRANSFER A→B 80] [TRANSFER C→D 50] [PURCHASE 30] ...
               │
           消费者一次只取一个，处理完（含事务提交）才取下一个

同一时刻只有一个事务在改 account 表，后续命令读到的是前一个已提交的结果，天然没有并发冲突。这是用「吞吐量换正确性」的取舍：账务必须准，速度可以慢一点。

扩展了解：真实高并发账务不会只靠单消费者。常见方案有：① 按 user_id 取模分多个队列，让同一用户的命令落到同一队列串行；② 在 SQL 里用 UPDATE account SET balance = balance - ? WHERE id = ? AND balance >= ?（乐观/条件更新），数据库行锁保证原子；③ 引入分布式事务 / Saga 模式。本项目用单消费者串行，是最直观、最不易出错的示意实现。

坑：@Transactional（Spring 声明式事务）只在被 Spring 代理调用的 Bean 方法上生效。这里 handle 是被 RabbitMQ 监听容器（框架）调用的，属于"外部调用"，所以事务正常生效；但切忌在消费者内部用 this.xxx() 自调用其他方法去指望它们各自有事务（见 02 篇 2.7 的坑①）。本项目各 handleXxx 内部直接操作 mapper，事务覆盖整段逻辑。

---

8.9 幂等与可重试：被重复投递怎么办

RabbitMQ 在消费者处理失败（抛异常、未 ack）时会重新投递消息，可能导致同一条命令被处理两次。本项目做了部分防护：

• PURCHASE 用订单状态防重：handlePurchase 开头检查 order.getStatus()，只有 PENDING 才处理；一旦置成 PAID/FAILED，再次收到同一命令会因状态不符直接 return，不会重复扣款。
• DEPOSIT / TRANSFER 目前没有显式幂等键：若同一条充值命令被重复投递，会加两次钱。教学项目可接受，但生产环境必须加幂等保护（例如给每条命令生成唯一 commandId，处理前先查 transaction_log 是否已有同 commandId 的成功流水，有则跳过）。

概念：幂等（Idempotent）——同一个操作执行一次和执行多次结果相同。消息队列场景下「至少一次投递」是常态，消费逻辑必须自己保证幂等，否则重投就会重复副作用。

---

8.10 本地与容器配置

本地开发需要 RabbitMQ（docker compose up -d postgres redis rabbitmq）。后端连接信息来自 application.yml 的 spring.rabbitmq.*，可用环境变量覆盖：

RABBITMQ_HOST=localhost
RABBITMQ_PORT=5672
RABBITMQ_USERNAME=guest
RABBITMQ_PASSWORD=guest

docker-compose.yml 里 rabbitmq 服务用镜像 rabbitmq:3-management-alpine，并把管理界面映射到宿主机 15672，可用它直观查看队列堆积、消费者数量、消息速率：

docker compose exec rabbitmq rabbitmqctl list_queues name messages consumers

若发现 messages 一直不归零，说明消费者没起来或处理慢；若 consumers = 0，说明应用没成功连上 RabbitMQ（检查 RABBITMQ_HOST 是否写成了 localhost 而实际该写容器服务名 rabbitmq）。

---

8.11 常见坑汇总

1. 队列没声明 / 没持久化：忘记 @Bean 声明 Queue/Exchange，或 durable=false，重启后队列消失、命令丢失。账务必须持久化。
2. 生产者与消费者 routing key 对不上：生产者发 uas.money，但绑定用的另一个 key，消息进不了队列、静默丢失。两者必须引用同一个常量。
3. 消息体不是 JSON / 没装 JSON 转换器：没配 Jackson2JsonMessageConverter 时，MoneyCommand 会用 Java 原生序列化，反序列化失败、消费报错。
4. 消费者里做长耗时 IO：账务事务一直占着数据库连接，队列积压、吞吐量暴跌。保持消费者轻量，只做数据库改动。
5. 并发消费者导致余额竞态：若给 @RabbitListener 配了多并发消费者，同一账户命令可能被并行处理，余额算错。本项目刻意单消费者串行。
6. double 算金额：用 double 表示金额会有浮点误差，务必用 BigDecimal（本项目全程 BigDecimal + NUMERIC(12,2)）。
7. DEPOSIT/TRANSFER 缺幂等保护：命令被重投会重复加钱（见 8.9）。生产必须加 commandId 去重。
8. 忘了先落单再发命令：购买必须先在 HTTP 层创建 PENDING 订单，消费者才能据 orderId 改状态；若只发命令没落单，消费者 selectById 会拿到 null 直接忽略。

---

8.12 练习

1. 看配置：打开 RabbitConfig.java，确认 EXCHANGE / QUEUE / ROUTING 三个常量，想清楚「生产者只认 EXCHANGE+ROUTING，消费者只认 QUEUE」如何解耦。
2. 跟一次转账：用 admin 给某用户 wallet:deposit 充值，再让用户 wallet:transfer 给另一个人。分别观察 account 表和 transaction_log 表的变化，理解 HTTP 层只发命令、消费者才改余额。
3. 观察异步：下单购买后立刻查订单，可能还是 PENDING；稍等再查变成 PAID/FAILED——体会「先受理、后结算」。余额不足时确认库存被 restoreStock 加回。
4. 故意制造失败：停掉 rabbitmq 容器后下单，订单会留在 PENDING，队列开始堆积；重启 RabbitMQ 后消费者继续处理——验证队列持久化的价值。
5. 加幂等键（进阶）：给 MoneyCommand 加一个 commandId 字段，handleDeposit 处理前先查 transaction_log 是否已有同 commandId 的 SUCCESS 流水，有则跳过。思考这如何防止消息重投导致的重复加钱。
6. 改造为并行（进阶，谨慎）：把 MoneyCommand 按 fromUserId 拆成多个队列，让「不同用户」的命令并行、「同一用户」的命令串行。对比单消费者方案在吞吐与正确性上的取舍。

---

8.13 一句话回顾

资金变动（充值 / 转账 / 购买）被建模成 MoneyCommand 命令，由 MoneyQueueProducer 发到 RabbitMQ 的 uas.exchange → uas.money.queue；单个 MoneyQueueConsumer 用 @RabbitListener + @Transactional 串行消费，模拟银行核心顺序记账，从根本上避免并发改余额的竞态。HTTP 层只做「校验 + 落单 + 发命令」并立即返回，结果通过查余额、订单状态与交易流水反映——这就是「先受理、后结算」的异步解耦。