中间件：RabbitMQ（消息队列）

1. 简介

RabbitMQ 在本项目中承担资金业务的异步消息总线角色。所有涉及账户余额变动的操作（充值、转账、下单购买）都先以「资金指令」消息发布到 RabbitMQ，再由消费者在数据库事务内统一执行，从而实现业务解耦、削峰填谷与最终一致性。项目使用 rabbitmq:3-management-alpine 镜像（带管理后台，端口 15672）。

2. 引入方式

pom.xml 引入 Spring AMQP 起步依赖：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

3. 配置位置

• 连接配置：application.yml 的 spring.rabbitmq

  rabbitmq:
    host: ${RABBITMQ_HOST:localhost}
    port: ${RABBITMQ_PORT:5672}
    username: ${RABBITMQ_USERNAME:guest}
    password: ${RABBITMQ_PASSWORD:guest}

• 容器部署：docker-compose.yml 中的 rabbitmq 服务，默认账号 uas/uas123456。

4. 在哪里应用

角色	类	说明
配置	config/RabbitConfig.java	声明交换机、队列、绑定、JSON 转换器
消息体	mq/MoneyCommand.java	资金指令（type、fromUserId、toUserId、amount、orderId、reference）
生产者	service/MoneyQueueProducer.java	将 MoneyCommand 发布到队列
消费者	service/MoneyQueueConsumer.java	监听队列，在事务内处理资金变动
调用方	service/WalletService.java、service/OrderService.java	在转账/充值/下单时发布消息

5. 怎么应用（示例）

① 声明基础设施（RabbitConfig） —— 直连交换机 + 持久化队列 + 路由键，消息采用 JSON 序列化：

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); }
@Bean
public Queue moneyQueue() { return new Queue(QUEUE, true); }
@Bean
public Binding moneyBinding() {
    return BindingBuilder.bind(moneyQueue()).to(moneyExchange()).with(ROUTING);
}
@Bean
public MessageConverter jsonMessageConverter() { return new Jackson2JsonMessageConverter(); }

② 发布消息（生产者） —— WalletService 转账与充值、OrderService 下单购买时发布：

// WalletService.transfer
moneyQueueProducer.publish(new MoneyCommand(
        MoneyCommand.TRANSFER, fromUserId, toUser.getId(), amount, null, "transfer"));
// WalletService.deposit
moneyQueueProducer.publish(new MoneyCommand(
        MoneyCommand.DEPOSIT, null, targetUserId, amount, null, "deposit"));
// OrderService 下单
moneyQueueProducer.publish(new MoneyCommand(
        MoneyCommand.PURCHASE, userId, null, total, order.getId(), String.valueOf(order.getId())));

③ 消费消息（消费者） —— @RabbitListener 监听队列，@Transactional 保证资金变动与流水写入在同一事务：

@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 -> { }
    }
}

handleTransfer 在余额充足时同时扣减转出方、增加转入方并写流水；余额不足只记录 FAILED 流水；handlePurchase 在余额不足时还将订单置为 FAILED 并回补库存。

6. 作用是什么

• 异步解耦：接口层只负责「下单/转账意图」入队并立即返回，真正的扣款在后台异步完成，提升接口响应速度。
• 削峰填谷：突发的大量资金请求由队列缓冲，消费者按自身能力匀速处理。
• 最终一致性：消息体为明确的资金指令，配合数据库事务与 transaction_log 流水，保证资金变动可追溯、可核对。
• 可靠性：队列与交换机均持久化（durable=true），降低服务重启导致消息丢失的风险。

7. 相关文件清单

• pom.xml（spring-boot-starter-amqp）
• src/main/resources/application.yml（spring.rabbitmq）
• docker-compose.yml（rabbitmq 服务）
• src/main/java/com/yang/backend/config/RabbitConfig.java
• src/main/java/com/yang/backend/mq/MoneyCommand.java
• src/main/java/com/yang/backend/service/MoneyQueueProducer.java
• src/main/java/com/yang/backend/service/MoneyQueueConsumer.java
• src/main/java/com/yang/backend/service/WalletService.java
• src/main/java/com/yang/backend/service/OrderService.java