消息中间件第一性原理：为什么需要异步通信？

/**
 * 订单服务 - 同步实现
 * 问题：串行等待，响应时间长，系统强耦合
 */
@Service
@Slf4j
public class OrderServiceSync {

    @Autowired
    private OrderRepository orderRepository;

    @Autowired
    private InventoryServiceClient inventoryService;

    @Autowired
    private PointsServiceClient pointsService;

    @Autowired
    private NotificationServiceClient notificationService;

    @Autowired
    private LogisticsServiceClient logisticsService;

    /**
     * 创建订单 - 同步方式
     * 总耗时：900ms
     */
    @Transactional
    public Order createOrder(OrderRequest request) {
        long startTime = System.currentTimeMillis();

        try {
            // 1. 创建订单（核心业务逻辑，50ms）
            log.info("开始创建订单，用户ID: {}", request.getUserId());
            Order order = buildOrder(request);
            orderRepository.save(order);
            log.info("订单创建成功，订单号: {}", order.getOrderNo());

            // 2. 同步调用库存服务（网络调用，200ms）
            log.info("开始扣减库存");
            long inventoryStart = System.currentTimeMillis();
            try {
                InventoryDeductRequest inventoryRequest = InventoryDeductRequest.builder()
                        .productId(request.getProductId())
                        .quantity(request.getQuantity())
                        .orderId(order.getId())
                        .build();
                inventoryService.deduct(inventoryRequest);
                log.info("库存扣减成功，耗时: {}ms", System.currentTimeMillis() - inventoryStart);
            } catch (FeignException e) {
                log.error("库存扣减失败", e);
                throw new BusinessException("库存不足或服务不可用");
            }

            // 3. 同步调用积分服务（网络调用，150ms）
            log.info("开始增加积分");
            long pointsStart = System.currentTimeMillis();
            try {
                PointsAddRequest pointsRequest = PointsAddRequest.builder()
                        .userId(request.getUserId())
                        .points((int) (order.getAmount() / 10))  // 消费10元得1积分
                        .orderId(order.getId())
                        .build();
                pointsService.add(pointsRequest);
                log.info("积分增加成功，耗时: {}ms", System.currentTimeMillis() - pointsStart);
            } catch (FeignException e) {
                // 积分服务失败不影响订单创建，只记录日志
                // 但用户还是要等待150ms
                log.error("积分增加失败，将稍后重试", e);
            }

            // 4. 同步调用通知服务（网络调用 + 第三方API，300ms）
            log.info("开始发送通知");
            long notificationStart = System.currentTimeMillis();
            try {
                NotificationRequest notificationRequest = NotificationRequest.builder()
                        .userId(request.getUserId())
                        .type(NotificationType.ORDER_CREATED)
                        .content("您的订单" + order.getOrderNo() + "已创建成功")
                        .build();
                notificationService.send(notificationRequest);
                log.info("通知发送成功，耗时: {}ms", System.currentTimeMillis() - notificationStart);
            } catch (FeignException e) {
                // 通知失败不影响订单创建，但用户还是要等待300ms
                log.error("通知发送失败，将稍后重试", e);
            }

            // 5. 同步调用物流服务（网络调用，250ms）
            log.info("开始创建物流单");
            long logisticsStart = System.currentTimeMillis();
            try {
                LogisticsCreateRequest logisticsRequest = LogisticsCreateRequest.builder()
                        .orderId(order.getId())
                        .address(request.getAddress())
                        .build();
                logisticsService.create(logisticsRequest);
                log.info("物流单创建成功，耗时: {}ms", System.currentTimeMillis() - logisticsStart);
            } catch (FeignException e) {
                // 物流失败不影响订单创建，但用户还是要等待250ms
                log.error("物流单创建失败，将稍后重试", e);
            }

            // 记录总耗时
            long totalTime = System.currentTimeMillis() - startTime;
            log.info("订单创建完成，总耗时: {}ms", totalTime);

            return order;

        } catch (Exception e) {
            log.error("订单创建失败", e);
            throw e;
        }
    }

    private Order buildOrder(OrderRequest request) {
        Order order = new Order();
        order.setOrderNo(generateOrderNo());
        order.setUserId(request.getUserId());
        order.setProductId(request.getProductId());
        order.setQuantity(request.getQuantity());
        order.setAmount(calculateAmount(request));
        order.setStatus(OrderStatus.PENDING_PAYMENT);
        order.setCreateTime(LocalDateTime.now());
        return order;
    }

    private String generateOrderNo() {
        return "ORD" + System.currentTimeMillis();
    }

    private BigDecimal calculateAmount(OrderRequest request) {
        // 简化处理，实际应该查询商品价格
        return new BigDecimal("100.00").multiply(new BigDecimal(request.getQuantity()));
    }
}

// 订单服务强依赖4个下游服务
@Autowired
private InventoryServiceClient inventoryService;  // 依赖1

@Autowired
private PointsServiceClient pointsService;  // 依赖2

@Autowired
private NotificationServiceClient notificationService;  // 依赖3

@Autowired
private LogisticsServiceClient logisticsService;  // 依赖4

// 新增需求：订单创建后发放优惠券
@Autowired
private CouponServiceClient couponService;  // 依赖5（需要修改订单服务代码）

@Transactional  // 这个事务只能管理本地数据库
public Order createOrder(OrderRequest request) {
    // 本地事务：订单表写入
    orderRepository.save(order);

    // 远程调用：无法纳入事务
    inventoryService.deduct(...);  // 如果失败，订单已经创建了
    pointsService.add(...);        // 如果失败，库存已经扣了
    notificationService.send(...); // 如果失败，积分已经加了
    logisticsService.create(...);  // 如果失败，通知已经发了

    // 如何保证一致性？
    // - 2PC（两阶段提交）？性能差，不适合微服务
    // - TCC（Try-Confirm-Cancel）？实现复杂
    // - Saga？代码侵入性强
}

/**
 * 订单服务 - 异步实现（使用RabbitMQ）
 * 优势：快速响应，系统解耦，高可用
 */
@Service
@Slf4j
public class OrderServiceAsync {

    @Autowired
    private OrderRepository orderRepository;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 创建订单 - 异步方式
     * 总耗时：20ms（提升45倍）
     */
    @Transactional
    public Order createOrder(OrderRequest request) {
        long startTime = System.currentTimeMillis();

        // 1. 创建订单（核心业务逻辑，50ms）
        log.info("开始创建订单，用户ID: {}", request.getUserId());
        Order order = buildOrder(request);
        orderRepository.save(order);
        log.info("订单创建成功，订单号: {}", order.getOrderNo());

        // 2. 发送库存扣减消息（异步，5ms）
        publishInventoryDeductEvent(order);

        // 3. 发送积分增加消息（异步，5ms）
        publishPointsAddEvent(order);

        // 4. 发送通知消息（异步，5ms）
        publishNotificationEvent(order);

        // 5. 发送物流创建消息（异步，5ms）
        publishLogisticsCreateEvent(order);

        // 记录总耗时
        long totalTime = System.currentTimeMillis() - startTime;
        log.info("订单创建完成，总耗时: {}ms", totalTime);

        return order;
    }

    /**
     * 发布库存扣减事件
     */
    private void publishInventoryDeductEvent(Order order) {
        InventoryDeductEvent event = InventoryDeductEvent.builder()
                .orderId(order.getId())
                .productId(order.getProductId())
                .quantity(order.getQuantity())
                .build();

        rabbitTemplate.convertAndSend(
                "order.exchange",        // Exchange
                "inventory.deduct",      // Routing Key
                event
        );
        log.info("库存扣减消息已发送");
    }

    /**
     * 发布积分增加事件
     */
    private void publishPointsAddEvent(Order order) {
        PointsAddEvent event = PointsAddEvent.builder()
                .userId(order.getUserId())
                .points((int) (order.getAmount().longValue() / 10))
                .orderId(order.getId())
                .build();

        rabbitTemplate.convertAndSend(
                "order.exchange",
                "points.add",
                event
        );
        log.info("积分增加消息已发送");
    }

    /**
     * 发布通知事件
     */
    private void publishNotificationEvent(Order order) {
        OrderNotificationEvent event = OrderNotificationEvent.builder()
                .userId(order.getUserId())
                .orderNo(order.getOrderNo())
                .type(NotificationType.ORDER_CREATED)
                .content("您的订单" + order.getOrderNo() + "已创建成功")
                .build();

        rabbitTemplate.convertAndSend(
                "order.exchange",
                "notification.order",
                event
        );
        log.info("通知消息已发送");
    }

    /**
     * 发布物流创建事件
     */
    private void publishLogisticsCreateEvent(Order order) {
        LogisticsCreateEvent event = LogisticsCreateEvent.builder()
                .orderId(order.getId())
                .address(order.getAddress())
                .build();

        rabbitTemplate.convertAndSend(
                "order.exchange",
                "logistics.create",
                event
        );
        log.info("物流创建消息已发送");
    }

    private Order buildOrder(OrderRequest request) {
        // 同上
    }
}

/**
 * 库存服务 - 消费者
 * 独立部署，异步处理，失败自动重试
 */
@Service
@Slf4j
public class InventoryConsumer {

    @Autowired
    private InventoryService inventoryService;

    /**
     * 监听库存扣减消息
     * 配置：自动ACK、最多重试3次、死信队列
     */
    @RabbitListener(queues = "inventory.deduct.queue")
    public void handleInventoryDeduct(InventoryDeductEvent event) {
        log.info("收到库存扣减消息: {}", event);

        try {
            // 处理库存扣减（200ms）
            inventoryService.deduct(
                    event.getProductId(),
                    event.getQuantity(),
                    event.getOrderId()
            );
            log.info("库存扣减成功");

        } catch (InsufficientInventoryException e) {
            // 库存不足：业务异常，不重试，发送到死信队列
            log.error("库存不足，订单ID: {}", event.getOrderId());
            throw new AmqpRejectAndDontRequeueException("库存不足", e);

        } catch (Exception e) {
            // 其他异常：可能是网络问题，重试
            log.error("库存扣减失败，将重试", e);
            throw e;
        }
    }
}

/**
 * 积分服务 - 消费者
 */
@Service
@Slf4j
public class PointsConsumer {

    @Autowired
    private PointsService pointsService;

    @RabbitListener(queues = "points.add.queue")
    public void handlePointsAdd(PointsAddEvent event) {
        log.info("收到积分增加消息: {}", event);

        try {
            // 处理积分增加（150ms）
            pointsService.add(
                    event.getUserId(),
                    event.getPoints(),
                    event.getOrderId()
            );
            log.info("积分增加成功");

        } catch (Exception e) {
            log.error("积分增加失败，将重试", e);
            throw e;
        }
    }
}

/**
 * 通知服务 - 消费者
 */
@Service
@Slf4j
public class NotificationConsumer {

    @Autowired
    private NotificationService notificationService;

    @RabbitListener(queues = "notification.order.queue")
    public void handleNotification(OrderNotificationEvent event) {
        log.info("收到通知消息: {}", event);

        try {
            // 发送通知（300ms）
            notificationService.send(
                    event.getUserId(),
                    event.getType(),
                    event.getContent()
            );
            log.info("通知发送成功");

        } catch (Exception e) {
            log.error("通知发送失败，将重试", e);
            throw e;
        }
    }
}

/**
 * 物流服务 - 消费者
 */
@Service
@Slf4j
public class LogisticsConsumer {

    @Autowired
    private LogisticsService logisticsService;

    @RabbitListener(queues = "logistics.create.queue")
    public void handleLogisticsCreate(LogisticsCreateEvent event) {
        log.info("收到物流创建消息: {}", event);

        try {
            // 创建物流单（250ms）
            logisticsService.create(
                    event.getOrderId(),
                    event.getAddress()
            );
            log.info("物流单创建成功");

        } catch (Exception e) {
            log.error("物流单创建失败，将重试", e);
            throw e;
        }
    }
}

// 同步方式：需要修改订单服务代码
@Service
public class OrderServiceSync {

    @Autowired
    private InventoryServiceClient inventoryService;

    @Autowired
    private PointsServiceClient pointsService;

    @Autowired
    private NotificationServiceClient notificationService;

    @Autowired
    private LogisticsServiceClient logisticsService;

    // 新增需求：订单创建后发放优惠券
    @Autowired
    private CouponServiceClient couponService;  // ← 需要修改订单服务

    @Transactional
    public Order createOrder(OrderRequest request) {
        Order order = buildOrder(request);
        orderRepository.save(order);

        inventoryService.deduct(...);
        pointsService.add(...);
        notificationService.send(...);
        logisticsService.create(...);
        couponService.issue(...);  // ← 需要修改订单服务代码

        return order;
    }
}

// 异步方式：无需修改订单服务代码
@Service
public class OrderServiceAsync {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Transactional
    public Order createOrder(OrderRequest request) {
        Order order = buildOrder(request);
        orderRepository.save(order);

        // 发布订单创建事件（不关心谁消费）
        OrderCreatedEvent event = new OrderCreatedEvent(order);
        rabbitTemplate.convertAndSend("order.exchange", "order.created", event);

        return order;
    }
}

// 营销服务（新增消费者，无需修改订单服务）
@Service
public class MarketingConsumer {

    @Autowired
    private CouponService couponService;

    // 订阅订单创建事件
    @RabbitListener(queues = "marketing.coupon.queue")
    public void handleOrderCreated(OrderCreatedEvent event) {
        // 发放优惠券
        couponService.issue(event.getUserId(), event.getOrderId());
    }
}

/**
 * RabbitMQ重试配置
 */
@Configuration
public class RabbitMQConfig {

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
            ConnectionFactory connectionFactory) {

        SimpleRabbitListenerContainerFactory factory =
                new SimpleRabbitListenerContainerFactory();

        factory.setConnectionFactory(connectionFactory);

        // 重试策略
        factory.setAdviceChain(
                RetryInterceptorBuilder
                        .stateless()
                        .maxAttempts(3)  // 最多重试3次
                        .backOffOptions(
                                1000,  // 初始间隔1秒
                                2.0,   // 指数退避因子
                                10000  // 最大间隔10秒
                        )
                        .recoverer(new RepublishMessageRecoverer(
                                rabbitTemplate(),
                                "dlx.exchange",  // 死信交换机
                                "dlx.routing.key"  // 死信路由键
                        ))
                        .build()
        );

        return factory;
    }
}

// 使用线程池实现异步
@Service
public class OrderServiceThreadPool {

    private final ExecutorService executorService =
            Executors.newFixedThreadPool(100);

    public Order createOrder(OrderRequest request) {
        Order order = buildOrder(request);
        orderRepository.save(order);

        // 异步执行下游任务
        executorService.submit(() -> {
            inventoryService.deduct(...);
        });

        executorService.submit(() -> {
            pointsService.add(...);
        });

        executorService.submit(() -> {
            notificationService.send(...);
        });

        executorService.submit(() -> {
            logisticsService.create(...);
        });

        return order;
    }
}

// 使用Redis List实现队列
@Service
public class OrderServiceRedis {

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    public Order createOrder(OrderRequest request) {
        Order order = buildOrder(request);
        orderRepository.save(order);

        // 发送消息到Redis
        String message = JSON.toJSONString(new InventoryDeductEvent(...));
        redisTemplate.opsForList().leftPush("inventory:queue", message);

        return order;
    }
}

// 消费者（需要自己写轮询逻辑）
@Service
public class InventoryConsumer {

    @Scheduled(fixedDelay = 100)
    public void consume() {
        String message = redisTemplate.opsForList().rightPop("inventory:queue");
        if (message != null) {
            InventoryDeductEvent event = JSON.parseObject(message, InventoryDeductEvent.class);
            inventoryService.deduct(event.getProductId(), event.getQuantity());
        }
    }
}

// 使用数据库表实现队列
@Service
public class OrderServiceDB {

    @Autowired
    private MessageRepository messageRepository;

    public Order createOrder(OrderRequest request) {
        Order order = buildOrder(request);
        orderRepository.save(order);

        // 插入消息到数据库
        Message message = new Message();
        message.setTopic("inventory.deduct");
        message.setPayload(JSON.toJSONString(new InventoryDeductEvent(...)));
        message.setStatus(MessageStatus.PENDING);
        messageRepository.save(message);

        return order;
    }
}

// 消费者（需要定时轮询）
@Service
public class MessageConsumer {

    @Scheduled(fixedDelay = 1000)
    public void consume() {
        List<Message> messages = messageRepository.findPendingMessages();
        for (Message message : messages) {
            try {
                processMessage(message);
                message.setStatus(MessageStatus.SUCCESS);
            } catch (Exception e) {
                message.setRetryCount(message.getRetryCount() + 1);
                message.setStatus(MessageStatus.FAILED);
            }
            messageRepository.save(message);
        }
    }
}

// 使用RabbitMQ
@Service
public class OrderServiceMQ {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public Order createOrder(OrderRequest request) {
        Order order = buildOrder(request);
        orderRepository.save(order);

        // 发送消息到MQ
        InventoryDeductEvent event = new InventoryDeductEvent(...);
        rabbitTemplate.convertAndSend("order.exchange", "inventory.deduct", event);

        return order;
    }
}

// 消费者（自动推送，无需轮询）
@Service
public class InventoryConsumer {

    @RabbitListener(queues = "inventory.deduct.queue")
    public void handleInventoryDeduct(InventoryDeductEvent event) {
        inventoryService.deduct(event.getProductId(), event.getQuantity());
    }
}