RocketMQ 消息驱动

@RestController
@RequestMapping("/order")
public class OrderController {

    @Autowired
    private StreamBridge streamBridge;

    @PostMapping("/create")
    public Result<String> createOrder(@RequestBody Order order) {
        // 1. 本地业务：写入订单数据库
        orderService.save(order);

        // 2. 构建消息
        Message<Order> message = MessageBuilder
                .withPayload(order)
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                .setHeader("orderId", order.getId())
                .build();

        // 3. 发送到 RocketMQ（bindingName 对应配置中的 order-output）
        boolean sent = streamBridge.send("order-output", message);

        // 4. 无论发送成功与否，订单已创建（最终一致性）
        return Result.success("订单创建成功，通知将异步发送");
    }
}

@Component
public class NotifyConsumer {

    @Bean
    public Consumer<Message<Order>> orderInput() {
        return message -> {
            Order order = message.getPayload();
            System.out.println("收到订单支付消息，订单号: " + order.getId());

            // 1. 发送短信
            smsService.send(order.getPhone(),
                "您在飞翔科技的订单 " + order.getId() + " 已支付成功");

            // 2. App 推送
            pushService.push(order.getUserId(),
                "订单支付成功", "订单 " + order.getId() + " 已确认");
        };
    }
}

@RestController
@RequestMapping("/order")
public class OrderController {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;   // 用于事务消息

    @Autowired
    private OrderMapper orderMapper;

    @PostMapping("/create-with-tx")
    public Result<String> createOrderWithTx(@RequestBody Order order) {
        // 构建 RocketMQ Message
        Message<Order> message = MessageBuilder
                .withPayload(order)
                .build();

        // 发送事务消息（重点：走半消息流程）
        // 参数：事务生产者组、Topic:Tag、消息体、业务参数（arg）
        TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction(
            "tx-order-create-group",           // 事务生产者组
            "stock-deduct-topic:deduct",       // Topic:Tag
            message,
            order.getId()                      // 透传给事务监听器的 arg
        );

        if (result.getSendStatus() == SendStatus.SEND_OK) {
            return Result.success("订单创建成功，库存扣减消息已投递");
        } else {
            return Result.error("消息发送失败");
        }
    }
}

@RocketMQTransactionListener(txProducerGroup = "tx-order-create-group")
public class OrderTransactionListener implements RocketMQLocalTransactionListener {

    @Autowired
    private OrderMapper orderMapper;

    /**
     * 半消息发送成功后，执行本地事务
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(
            Message msg, Object arg) {

        Long orderId = (Long) arg;
        try {
            // 执行本地事务：写入订单
            orderMapper.insert(new OrderDO(orderId, "CREATED"));
            System.out.println("订单创建成功，事务消息提交: " + orderId);
            return RocketMQLocalTransactionState.COMMIT;  // 提交 → Consumer 可见
        } catch (Exception e) {
            System.out.println("订单创建失败，事务消息回滚: " + orderId);
            return RocketMQLocalTransactionState.ROLLBACK;  // 回滚 → 消息被丢弃
        }
    }

    /**
     * 当 executeLocalTransaction 未返回（超时/宕机）时，
     * Broker 主动回调此方法查询事务状态
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        // 从消息中提取业务标识
        Long orderId = (Long) msg.getHeaders().get("orderId");

        // 查询数据库确认订单是否已创建
        OrderDO order = orderMapper.selectById(orderId);
        if (order != null && "CREATED".equals(order.getStatus())) {
            System.out.println("回查：订单已创建，提交事务消息: " + orderId);
            return RocketMQLocalTransactionState.COMMIT;
        } else {
            System.out.println("回查：订单未创建，回滚事务消息: " + orderId);
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }
}

@Component
public class StockConsumer {

    @Autowired
    private StockMapper stockMapper;

    @Bean
    public Consumer<Message<Order>> stockInput() {
        return message -> {
            Order order = message.getPayload();
            System.out.println("扣减库存: productId=" + order.getProductId()
                + ", count=" + order.getCount());
            stockMapper.deduct(order.getProductId(), order.getCount());
        };
    }
}

@RestController
@RequestMapping("/order")
public class OrderController {

    @Autowired
    private StreamBridge streamBridge;

    @PostMapping("/create")
    public Result<String> createOrder(@RequestBody Order order) {
        order.setStatus("UNPAID");
        orderService.save(order);

        // 构建延时消息
        Message<Order> delayMsg = MessageBuilder
                .withPayload(order)
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                .setHeader("orderId", order.getId())
                // 关键：设置延时级别 = 16（约 30 分钟后投递）
                .setHeader("DELAY", 16)
                .build();

        streamBridge.send("order-delay-output", delayMsg);
        return Result.success("订单已创建，请在 30 分钟内完成支付");
    }
}

@Component
public class OrderTimeoutConsumer {

    @Autowired
    private OrderMapper orderMapper;

    @Bean
    public Consumer<Message<Order>> orderDelayInput() {
        return message -> {
            Order order = message.getPayload();
            OrderDO orderDO = orderMapper.selectById(order.getId());

            if (orderDO != null && "UNPAID".equals(orderDO.getStatus())) {
                // 超时未支付 → 取消订单
                orderMapper.updateStatus(order.getId(), "CANCELLED");
                System.out.println("订单 " + order.getId() + " 超时未支付，已自动取消");
                // 释放库存...（可再发一条消息给库存服务）
            } else {
                System.out.println("订单 " + order.getId() + " 已支付，无需取消");
            }
        };
    }
}

@RestController
@RequestMapping("/order")
public class OrderController {

    @Autowired
    private StreamBridge streamBridge;

    @PostMapping("/{orderId}/pay")
    public Result<String> payOrder(@PathVariable Long orderId) {
        OrderEvent event = new OrderEvent(orderId, "PAID", "用户已支付");

        // 关键：用 orderId 作为 hashKey，同一订单消息进入同一 Queue
        Message<OrderEvent> message = MessageBuilder
                .withPayload(event)
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                .setHeader("orderId", orderId)
                // RocketMQ 用 TAGS 做 Tag 过滤
                .setHeader("TAGS", "status-change")
                // 关键：顺序消息的哈希键
                .setHeader("rocketmq_KEYS", orderId.toString())
                .build();

        streamBridge.send("order-status-output", message);
        return Result.success("支付成功");
    }

    @PostMapping("/{orderId}/ship")
    public Result<String> shipOrder(@PathVariable Long orderId) {
        OrderEvent event = new OrderEvent(orderId, "SHIPPED", "已发货");

        Message<OrderEvent> message = MessageBuilder
                .withPayload(event)
                .setHeader("orderId", orderId)
                .setHeader("TAGS", "status-change")
                .setHeader("rocketmq_KEYS", orderId.toString())
                .build();

        streamBridge.send("order-status-output", message);
        return Result.success("发货成功");
    }
}

@Component
public class OrderStatusConsumer {

    @Bean
    public Consumer<Message<OrderEvent>> orderStatusInput() {
        return message -> {
            OrderEvent event = message.getPayload();
            String currentStatus = event.getStatus();
            System.out.println("处理订单状态变更: orderId=" + event.getOrderId()
                + ", status=" + currentStatus);

            // 校验状态流转是否合法
            OrderDO order = orderMapper.selectById(event.getOrderId());
            if (!isValidTransition(order.getStatus(), currentStatus)) {
                // 状态不合法 → 跳过该消息并继续消费后续消息
                System.out.println("非法状态流转: " + order.getStatus()
                    + " → " + currentStatus + "，跳过");
                return;   // 返回即 ACK，队列继续
            }

            // 更新订单状态
            orderMapper.updateStatus(event.getOrderId(), currentStatus);
        };
    }

    private boolean isValidTransition(String from, String to) {
        // 合法流转：CREATED → PAID → SHIPPED → COMPLETED
        String validFlow = "CREATED,PAID,SHIPPED,COMPLETED";
        if (from == null) return "CREATED".equals(to);
        return validFlow.indexOf(from) < validFlow.indexOf(to);
    }
}

// ❌ 错误：异常直接抛出，队列阻塞
@Bean
public Consumer<Message<OrderEvent>> orderStatusInput() {
    return message -> {
        process(message.getPayload());  // 可能抛异常
    };
}

// ✅ 正确：catch 异常，告警后继续
@Bean
public Consumer<Message<OrderEvent>> orderStatusInput() {
    return message -> {
        try {
            process(message.getPayload());
        } catch (Exception e) {
            log.error("顺序消息处理失败，人工介入: {}", message.getPayload(), e);
            // 返回即 ACK，不阻塞队列
        }
    };
}

@Bean
public Consumer<Message<Order>> orderInput() {
    return message -> {
        String msgId = (String) message.getHeaders().get("rocketmq_MESSAGE_ID");
        // Redis SETNX 防重
        Boolean lock = redisTemplate.opsForValue()
                .setIfAbsent("msg:consume:" + msgId, "1", Duration.ofHours(1));
        if (Boolean.FALSE.equals(lock)) {
            System.out.println("消息 " + msgId + " 已消费，跳过");
            return;
        }
        // 执行业务逻辑
        processOrder(message.getPayload());
    };
}