// 同步代碼
public void syncOrderDetail() {
    Order order = orderDao.getById(1L); // 100ms
    List<Goods> goodsList = goodsDao.getByOrderId(1L); // 150ms
    Address address = userService.getAddress(1L); // 200ms
    // 組裝數(shù)據(jù)返回
}
// 想搞異步，就new Thread
public void asyncBadExample() {
    Order[] order = new Order[1]; // 用數(shù)組存，因?yàn)槟涿麅?nèi)部類要final
    List<Goods>[] goodsList = new List[1];
    Address[] address = new Address[1];
    // 查訂單
    new Thread(() -> order[0] = orderDao.getById(1L)).start();
    // 查商品
    new Thread(() -> goodsList[0] = goodsDao.getByOrderId(1L)).start();
    // 查地址
    new Thread(() -> address[0] = userService.getAddress(1L)).start();
    // 等結(jié)果
    while (order[0] == null || goodsList[0] == null || address[0] == null) {
        // 空循環(huán)等，CPU直接干燒
    }
    // 組裝數(shù)據(jù)
}

// 先搞個(gè)線程池
private ExecutorService executor = Executors.newFixedThreadPool(3);
public void asyncWithFuture() throws ExecutionException, InterruptedException {
    // 提交任務(wù)，返回Future
    Future<Order> orderFuture = executor.submit(() -> orderDao.getById(1L));
    Future<List<Goods>> goodsFuture = executor.submit(() -> goodsDao.getByOrderId(1L));
    Future<Address> addressFuture = executor.submit(() -> userService.getAddress(1L));
    // 拿結(jié)果（這里會阻塞，直到任務(wù)完成）
    Order order = orderFuture.get();
    List<Goods> goodsList = goodsFuture.get();
    Address address = addressFuture.get();
    // 組裝數(shù)據(jù)
}

// 自定義線程池（后面詳細(xì)講怎么配）
private ExecutorService orderExecutor = new ThreadPoolExecutor(
    4, // 核心線程數(shù)
    8, // 最大線程數(shù)
    60L, // 空閑線程存活時(shí)間
    TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(100), // 隊(duì)列
    new ThreadFactoryBuilder().setNameFormat("order-async-%d").build(), // 線程名
    new ThreadPoolExecutor.CallerRunsPolicy() // 拒絕策略
);
public void asyncWithCompletableFuture() {
    // 1. 異步查訂單
    CompletableFuture<Order> orderFuture = CompletableFuture.supplyAsync(() -> {
        System.out.println("線程" + Thread.currentThread().getName() + "：查訂單");
        return orderDao.getById(1L);
    }, orderExecutor);
    // 2. 異步查商品（不用等訂單，直接跑）
    CompletableFuture<List<Goods>> goodsFuture = CompletableFuture.supplyAsync(() -> {
        System.out.println("線程" + Thread.currentThread().getName() + "：查商品");
        return goodsDao.getByOrderId(1L);
    }, orderExecutor);
    // 3. 異步查地址（需要訂單里的userId，所以等訂單查完再跑）
    CompletableFuture<Address> addressFuture = orderFuture.thenCompose(order -> {
        // thenCompose：用前一個(gè)任務(wù)的結(jié)果，發(fā)起新的異步任務(wù)
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("線程" + Thread.currentThread().getName() + "：查地址，userId=" + order.getUserId());
            return userService.getAddress(order.getUserId());
        }, orderExecutor);
    });
    // 4. 等所有任務(wù)完成，組裝結(jié)果（不阻塞主線程，用whenComplete處理結(jié)果）
    CompletableFuture.allOf(orderFuture, goodsFuture, addressFuture)
        .whenComplete((v, e) -> {
            // v是allOf的返回值，因?yàn)閍llOf沒有返回值，所以v是null
            // e是異常，如果三個(gè)任務(wù)中有一個(gè)拋異常，e就是那個(gè)異常
            if (e != null) {
                System.out.println("異步任務(wù)出錯了：" + e.getMessage());
                return;
            }
            // 拿結(jié)果（這里get()不會阻塞，因?yàn)閍llOf已經(jīng)確保任務(wù)完成了）
            try {
                Order order = orderFuture.get();
                List<Goods> goodsList = goodsFuture.get();
                Address address = addressFuture.get();
                // 組裝數(shù)據(jù)
                OrderDetailVO vo = new OrderDetailVO(order, goodsList, address);
                System.out.println("組裝完成：" + vo);
            } catch (Exception ex) {
                // 這里一般不會拋異常，因?yàn)榍懊鎍llOf已經(jīng)處理了
                ex.printStackTrace();
            }
        });
    // 主線程繼續(xù)干別的，不用等上面的任務(wù)完成
    System.out.println("主線程：異步任務(wù)已發(fā)起，我先溜了～");
}

// Future的嵌套寫法，丑哭
Future<Order> orderFuture = executor.submit(() -> orderDao.getById(1L));
Order order = orderFuture.get();
Future<Address> addressFuture = executor.submit(() -> userService.getAddress(order.getUserId()));

// 鏈?zhǔn)秸{(diào)用示例
CompletableFuture<BigDecimal> payAmountFuture = CompletableFuture.supplyAsync(() -> {
    // 1. 查訂單
    System.out.println("查訂單");
    return orderDao.getById(1L);
}, orderExecutor)
.thenApply(order -> {
    // 2. 計(jì)算優(yōu)惠金額（用訂單的金額和用戶等級，同步操作）
    System.out.println("計(jì)算優(yōu)惠金額，訂單金額：" + order.getAmount());
    BigDecimal discount = calculateDiscount(order); // 同步方法，不用開新線程
    order.setDiscount(discount);
    return order;
})
.thenCombine(CompletableFuture.supplyAsync(() -> {
    // 3. 查商品列表（異步，和計(jì)算優(yōu)惠并行）
    System.out.println("查商品列表");
    return goodsDao.getByOrderId(1L);
}, orderExecutor), (order, goodsList) -> {
    // 4. 結(jié)合訂單（帶優(yōu)惠）和商品列表，計(jì)算實(shí)付金額
    System.out.println("計(jì)算實(shí)付金額");
    BigDecimal totalGoodsAmount = goodsList.stream()
        .map(Goods::getPrice)
        .reduce(BigDecimal.ZERO, BigDecimal::add);
    // 實(shí)付 = 商品總金額 - 優(yōu)惠
    return totalGoodsAmount.subtract(order.getDiscount());
})
.exceptionally(ex -> {
    // 5. 異常處理：如果前面任何一步出錯，返回默認(rèn)值0
    System.out.println("計(jì)算實(shí)付金額出錯：" + ex.getMessage());
    return BigDecimal.ZERO;
});
// 拿結(jié)果（如果想阻塞等結(jié)果，可以用get()，也可以用join()，join()不用拋異常）
BigDecimal payAmount = payAmountFuture.join();
System.out.println("最終實(shí)付金額：" + payAmount);

.handle((result, ex) -> {
    if (ex != null) {
        System.out.println("出錯了：" + ex.getMessage());
        return BigDecimal.ZERO;
    }
    return result;
});

CompletableFuture<Address> addressFuture = CompletableFuture.supplyAsync(() -> {
    // 模擬調(diào)用第三方接口超時(shí)
    try {
        Thread.sleep(300); // 睡300ms，超過200ms的超時(shí)時(shí)間
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return userService.getAddress(1L);
}, orderExecutor)
// 超時(shí)控制：200ms超時(shí)，返回默認(rèn)地址
.completeOnTimeout(new Address("默認(rèn)地址：未知"), 200, TimeUnit.MILLISECONDS);
// 或者超時(shí)拋異常
// .orTimeout(200, TimeUnit.MILLISECONDS);

new ThreadPoolExecutor(
    int corePoolSize,        // 核心線程數(shù)
    int maximumPoolSize,     // 最大線程數(shù)
    long keepAliveTime,      // 空閑線程存活時(shí)間
    TimeUnit unit,           // 存活時(shí)間單位
    BlockingQueue<Runnable> workQueue, // 任務(wù)隊(duì)列
    ThreadFactory threadFactory,       // 線程工廠
    RejectedExecutionHandler handler   // 拒絕策略
);

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.concurrent.*;

publicclass ThreadPoolConfig {

    // CPU核心數(shù)
    privatestaticfinalint CPU_CORES = Runtime.getRuntime().availableProcessors();

    // 訂單相關(guān)異步任務(wù)線程池（IO密集型）
    @Bean
    public Executor orderAsyncExecutor() {
        // 線程工廠：給線程起名字
        ThreadFactory threadFactory = new ThreadFactoryBuilder()
                .setNameFormat("order-async-%d") // 線程名：order-async-1, order-async-2...
                .setDaemon(false) // 非守護(hù)線程（守護(hù)線程會隨著主線程退出而退出，這里要避免）
                .build();

        // 線程池參數(shù)
        int corePoolSize = CPU_CORES * 2; // 核心線程數(shù)：CPU*2
        int maximumPoolSize = CPU_CORES * 4; // 最大線程數(shù)：CPU*4
        long keepAliveTime = 60L; // 空閑線程存活時(shí)間：60秒
        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(100); // 任務(wù)隊(duì)列：100個(gè)
        RejectedExecutionHandler rejectedHandler = new ThreadPoolExecutor.CallerRunsPolicy(); // 拒絕策略：調(diào)用者執(zhí)行

        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                corePoolSize,
                maximumPoolSize,
                keepAliveTime,
                TimeUnit.SECONDS,
                workQueue,
                threadFactory,
                rejectedHandler
        );

        // 允許核心線程超時(shí)銷毀（默認(rèn)核心線程不會超時(shí)，這里設(shè)為true，空閑時(shí)可以銷毀，節(jié)省資源）
        executor.allowCoreThreadTimeOut(true);

        return executor;
    }

    // 計(jì)算相關(guān)異步任務(wù)線程池（CPU密集型，比如計(jì)算優(yōu)惠、統(tǒng)計(jì)數(shù)據(jù)）
    @Bean
    public Executor calculateAsyncExecutor() {
        ThreadFactory threadFactory = new ThreadFactoryBuilder()
                .setNameFormat("calculate-async-%d")
                .build();

        int corePoolSize = CPU_CORES + 1; // 核心線程數(shù)：CPU+1
        int maximumPoolSize = CPU_CORES + 1; // 最大線程數(shù)和核心線程數(shù)一致，不需要臨時(shí)線程
        long keepAliveTime = 60L;
        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(50);
        RejectedExecutionHandler rejectedHandler = new ThreadPoolExecutor.CallerRunsPolicy();

        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                corePoolSize,
                maximumPoolSize,
                keepAliveTime,
                TimeUnit.SECONDS,
                workQueue,
                threadFactory,
                rejectedHandler
        );

        return executor;
    }
}

@RestController
@RequestMapping("/order")
publicclassOrderController {

    @Autowired
    private OrderService orderService;

    @Autowired
    private Executor orderAsyncExecutor; // 注入前面配置的訂單異步線程池

    /**
     * 下單接口
     */
    @PostMapping("/create")
    public Result<OrderVO> createOrder(@RequestBody OrderCreateDTO dto) {
        // 1. 參數(shù)校驗(yàn)（同步）
        validateParams(dto);

        // 2. 核心流程：創(chuàng)建訂單（同步，包括庫存扣減、創(chuàng)建訂單記錄）
        Order order = orderService.createOrder(dto);

        // 3. 非核心流程：異步執(zhí)行
        asyncProcessAfterCreate(order);

        // 4. 返回結(jié)果（核心流程完成就返回，不用等異步流程）
        OrderVO vo = convertToVO(order);
        return Result.success(vo);
    }

    /**
     * 參數(shù)校驗(yàn)（同步）
     */
    private void validateParams(OrderCreateDTO dto) {
        if (dto.getUserId() == null) {
            thrownew BusinessException("用戶ID不能為空");
        }
        if (CollectionUtils.isEmpty(dto.getGoodsList())) {
            thrownew BusinessException("商品列表不能為空");
        }
        // 其他校驗(yàn)...
    }

    /**
     * 下單后異步處理（非核心流程）
     */
    private void asyncProcessAfterCreate(Order order) {
        // 3.1 異步記錄日志
        CompletableFuture.runAsync(() -> {
            orderService.recordOrderLog(order.getId(), "訂單創(chuàng)建成功");
        }, orderAsyncExecutor)
        .exceptionally(ex -> {
            // 日志記錄失敗不影響主流程，只打日志
            log.error("記錄訂單日志失敗，orderId={}", order.getId(), ex);
            returnnull;
        });

        // 3.2 異步更新用戶積分（需要訂單金額，所以用supplyAsync）
        CompletableFuture.supplyAsync(() -> {
            // 計(jì)算積分：訂單金額1元=1積分
            int points = order.getAmount().intValue();
            return orderService.updateUserPoints(order.getUserId(), points);
        }, orderAsyncExecutor)
        .exceptionally(ex -> {
            log.error("更新用戶積分失敗，userId={}, orderId={}", order.getUserId(), order.getId(), ex);
            // 積分更新失敗可以重試，這里用定時(shí)任務(wù)重試，或者發(fā)MQ重試
            sendRetryUpdatePointsMQ(order.getUserId(), order.getAmount().intValue());
            returnnull;
        });

        // 3.3 異步發(fā)送通知（需要用戶手機(jī)號，所以先查用戶信息）
        CompletableFuture.supplyAsync(() -> {
            // 查用戶信息（IO密集型）
            return userService.getUserById(order.getUserId());
        }, orderAsyncExecutor)
        .thenCompose(user -> {
            // 用用戶手機(jī)號發(fā)送短信通知（異步）
            return CompletableFuture.runAsync(() -> {
                notifyService.sendSms(user.getPhone(), "您的訂單" + order.getId() + "已創(chuàng)建成功");
            }, orderAsyncExecutor);
        })
        .exceptionally(ex -> {
            log.error("發(fā)送下單通知失敗，orderId={}", order.getId(), ex);
            returnnull;
        });

        // 3.4 異步同步商品銷量
        CompletableFuture.runAsync(() -> {
            for (OrderGoods goods : order.getOrderGoodsList()) {
                goodsService.syncSalesCount(goods.getGoodsId(), goods.getQuantity());
            }
        }, orderAsyncExecutor)
        .exceptionally(ex -> {
            log.error("同步商品銷量失敗，orderId={}", order.getId(), ex);
            returnnull;
        });
    }

    /**
     * 轉(zhuǎn)換VO
     */
    private OrderVO convertToVO(Order order) {
        OrderVO vo = new OrderVO();
        vo.setOrderId(order.getId());
        vo.setUserId(order.getUserId());
        vo.setAmount(order.getAmount());
        vo.setStatus(order.getStatus());
        vo.setCreateTime(order.getCreateTime());
        return vo;
    }

    /**
     * 發(fā)送積分更新重試MQ
     */
    private void sendRetryUpdatePointsMQ(Long userId, int points) {
        RetryUpdatePointsMQ mq = new RetryUpdatePointsMQ();
        mq.setUserId(userId);
        mq.setPoints(points);
        mq.setRetryCount(0); // 重試次數(shù)
        mq.setNextRetryTime(System.currentTimeMillis() + 5 * 60 * 1000); // 5分鐘后重試
        rabbitTemplate.convertAndSend("retry-update-points-exchange", "retry.update.points", mq);
    }
}

@Service
publicclass OrderServiceImpl implements OrderService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Override
    public Order createOrder(OrderCreateDTO dto) {
        // 1. 扣庫存
        stockService.deductStock(dto.getGoodsList());

        // 2. 創(chuàng)建訂單
        Order order = orderMapper.insert(buildOrder(dto));

        // 3. 發(fā)送MQ消息（分布式異步）
        sendOrderCreatedMQ(order);

        return order;
    }

    /**
     * 發(fā)送訂單創(chuàng)建成功的MQ消息
     */
    private void sendOrderCreatedMQ(Order order) {
        OrderCreatedMQ mq = new OrderCreatedMQ();
        mq.setOrderId(order.getId());
        mq.setUserId(order.getUserId());
        mq.setAmount(order.getAmount());
        mq.setCreateTime(new Date());

        try {
            // 發(fā)送消息到order.created隊(duì)列
            rabbitTemplate.convertAndSend(
                "order-exchange", // 交換機(jī)
                "order.created", // 路由鍵
                mq,
                message -> {
                    // 設(shè)置消息過期時(shí)間：30分鐘（超過30分鐘沒處理，就放到死信隊(duì)列）
                    message.getMessageProperties().setExpiration("1800000");
                    // 設(shè)置消息唯一ID，用于冪等處理
                    message.getMessageProperties().setMessageId(order.getId().toString());
                    return message;
                }
            );
            log.info("發(fā)送訂單創(chuàng)建MQ消息成功，orderId={}", order.getId());
        } catch (Exception e) {
            log.error("發(fā)送訂單創(chuàng)建MQ消息失敗，orderId={}", order.getId(), e);
            // 消息發(fā)送失敗，可以記錄到本地表，用定時(shí)任務(wù)重試
            mqRetryService.saveRetryRecord("order.created", mq);
        }
    }
}

@Service
publicclass LogisticsConsumer {

    @Autowired
    private LogisticsService logisticsService;

    /**
     * 監(jiān)聽訂單創(chuàng)建消息，創(chuàng)建物流單
     */
    @RabbitListener(
        queues = "order.created.logistics.queue", // 物流服務(wù)的隊(duì)列
        containerFactory = "rabbitListenerContainerFactory"
    )
    public void handleOrderCreated(Message message, @Payload OrderCreatedMQ mq) {
        String messageId = message.getMessageProperties().getMessageId();
        log.info("收到訂單創(chuàng)建消息，準(zhǔn)備創(chuàng)建物流單，messageId={}, orderId={}", messageId, mq.getOrderId());

        try {
            // 1. 冪等處理：先查有沒有處理過這個(gè)消息
            if (logisticsService.existsByOrderId(mq.getOrderId())) {
                log.info("物流單已存在，跳過處理，orderId={}", mq.getOrderId());
                return;
            }

            // 2. 創(chuàng)建物流單
            logisticsService.createLogisticsOrder(mq);

            // 3. 手動確認(rèn)消息（如果用的是手動確認(rèn)模式）
            Channel channel = message.getMessageProperties().getChannel();
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            log.error("處理訂單創(chuàng)建消息失敗，messageId={}, orderId={}", messageId, mq.getOrderId(), e);
            // 處理失敗，手動拒絕消息，讓MQ重試（重試次數(shù)由MQ配置）
            try {
                Channel channel = message.getMessageProperties().getChannel();
                // requeue=false：不重新放回原隊(duì)列，讓MQ放到死信隊(duì)列
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            } catch (IOException ex) {
                log.error("拒絕消息失敗，messageId={}", messageId, ex);
            }
        }
    }
}