import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
 * 異步任務(wù)級(jí)聯(lián)調(diào)用示例服務(wù)
 * 使用 Java 17+ 特性 + Spring 注解 + 完整異常處理
 */
@Slf4j
@Service
public class AsyncCascadeService {
    // 自定義線程池（避免使用默認(rèn) ForkJoinPool，提升可控性）
    private static final ExecutorService CUSTOM_EXECUTOR = Executors.newFixedThreadPool(5);
    /**
     * 級(jí)聯(lián)異步任務(wù)示例：
     * 1. supplyAsync：異步執(zhí)行任務(wù)1（生成訂單ID）
     * 2. thenApplyAsync：異步處理任務(wù)1結(jié)果（根據(jù)訂單ID查詢訂單詳情）
     * 3. thenApplyAsync：異步處理任務(wù)2結(jié)果（計(jì)算訂單金額）
     */
    public CompletableFuture<OrderAmountResult> processOrderAsync(Long userId) {
        // 第一個(gè)異步任務(wù)：生成訂單ID（supplyAsync 執(zhí)行有返回值的異步任務(wù)）
        CompletableFuture<String> generateOrderIdFuture = CompletableFuture.supplyAsync(() -> {
            log.info("異步任務(wù)1：生成訂單ID，線程：{}", Thread.currentThread().getName());
            // 模擬業(yè)務(wù)耗時(shí)
            simulateDelay(100);
            return "ORDER_" + userId + "_" + System.currentTimeMillis();
        }, CUSTOM_EXECUTOR) // 指定自定義線程池（推薦，避免默認(rèn)池耗盡）
        .exceptionally(ex -> {
            log.error("任務(wù)1執(zhí)行失敗：", ex);
            throw new RuntimeException("生成訂單ID失敗", ex);
        });
        // 第二個(gè)異步任務(wù)：處理訂單ID，查詢訂單詳情（thenApplyAsync 異步處理前序結(jié)果）
        CompletableFuture<OrderDetail> queryOrderDetailFuture = generateOrderIdFuture
                .thenApplyAsync(orderId -> {
                    log.info("異步任務(wù)2：查詢訂單詳情，訂單ID：{}，線程：{}", orderId, Thread.currentThread().getName());
                    simulateDelay(150);
                    // 模擬查詢結(jié)果
                    return new OrderDetail(orderId, userId, "商品A", 2, 99.9);
                }, CUSTOM_EXECUTOR)
                .exceptionally(ex -> {
                    log.error("任務(wù)2執(zhí)行失?。?, ex);
                    throw new RuntimeException("查詢訂單詳情失敗", ex);
                });
        // 第三個(gè)異步任務(wù)：計(jì)算訂單總金額（級(jí)聯(lián)處理前序結(jié)果）
        return queryOrderDetailFuture
                .thenApplyAsync(orderDetail -> {
                    log.info("異步任務(wù)3：計(jì)算訂單金額，訂單詳情：{}，線程：{}", orderDetail, Thread.currentThread().getName());
                    simulateDelay(100);
                    double totalAmount = orderDetail.price() * orderDetail.quantity();
                    // 使用 Java 17 record 簡(jiǎn)化數(shù)據(jù)返回
                    return new OrderAmountResult(orderDetail.orderId(), totalAmount, totalAmount * 0.08);
                }, CUSTOM_EXECUTOR)
                .exceptionally(ex -> {
                    log.error("任務(wù)3執(zhí)行失?。?, ex);
                    throw new RuntimeException("計(jì)算訂單金額失敗", ex);
                });
    }
    // 模擬業(yè)務(wù)耗時(shí)
    private void simulateDelay(long millis) {
        try {
            TimeUnit.MILLISECONDS.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("任務(wù)被中斷", e);
        }
    }
    // Java 17 Record：訂單詳情
    public record OrderDetail(String orderId, Long userId, String productName, int quantity, double price) {}
    // Java 17 Record：訂單金額計(jì)算結(jié)果
    public record OrderAmountResult(String orderId, double totalAmount, double tax) {}
    // 資源關(guān)閉：Spring 銷毀時(shí)關(guān)閉線程池（try-with-resources 思想）
    @Override
    protected void finalize() throws Throwable {
        try (ExecutorService executor = CUSTOM_EXECUTOR) {
            executor.shutdown();
            if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
            log.info("異步線程池已關(guān)閉");
        } finally {
            super.finalize();
        }
    }
    // 測(cè)試入口
    public static void main(String[] args) {
        AsyncCascadeService service = new AsyncCascadeService();
        // 執(zhí)行級(jí)聯(lián)異步任務(wù)
        CompletableFuture<OrderAmountResult> resultFuture = service.processOrderAsync(1001L);
        // 阻塞獲取結(jié)果（實(shí)際業(yè)務(wù)中建議使用 thenAccept/thenRun 非阻塞處理）
        resultFuture.whenComplete((result, ex) -> {
            if (ex != null) {
                log.error("級(jí)聯(lián)任務(wù)執(zhí)行失敗", ex);
            } else {
                log.info("級(jí)聯(lián)任務(wù)執(zhí)行完成：{}", result);
            }
            // 關(guān)閉線程池
            CUSTOM_EXECUTOR.shutdown();
        }).join();
    }
}

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.beans.factory.annotation.PreDestroy;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
 * 非阻塞處理 CompletableFuture 結(jié)果示例
 * 核心：thenAccept（消費(fèi)結(jié)果）/thenRun（收尾動(dòng)作）替代 join() 阻塞調(diào)用
 */
@Slf4j
@Service
public class NonBlockingAsyncService {
    // 自定義線程池（Spring 中建議通過(guò) @Bean 配置，此處簡(jiǎn)化）
    private final ExecutorService customExecutor = Executors.newFixedThreadPool(5);
    // 訂單狀態(tài)枚舉（Java 17+ 增強(qiáng)枚舉，可結(jié)合 switch 表達(dá)式）
    public enum OrderStatus { SUCCESS, FAILURE }
    /**
     * 級(jí)聯(lián)異步任務(wù)（非阻塞核心邏輯）
     */
    public void processOrderNonBlocking(Long userId) {
        // 步驟1：異步生成訂單ID（supplyAsync）
        CompletableFuture<String> orderIdFuture = CompletableFuture.supplyAsync(() -> {
            log.info("【異步任務(wù)1】生成訂單ID | 線程：{}", Thread.currentThread().getName());
            simulateDelay(100);
            return "ORDER_" + userId + "_" + System.currentTimeMillis();
        }, customExecutor)
        .exceptionally(ex -> {
            log.error("【任務(wù)1失敗】生成訂單ID異常", ex);
            throw new RuntimeException("生成訂單ID失敗", ex);
        });
        // 步驟2：異步查詢訂單詳情（thenApplyAsync）
        CompletableFuture<OrderDetail> orderDetailFuture = orderIdFuture
                .thenApplyAsync(orderId -> {
                    log.info("【異步任務(wù)2】查詢訂單詳情 | 訂單ID：{} | 線程：{}", orderId, Thread.currentThread().getName());
                    simulateDelay(150);
                    return new OrderDetail(orderId, userId, "Java編程實(shí)戰(zhàn)", 2, 89.9);
                }, customExecutor)
                .exceptionally(ex -> {
                    log.error("【任務(wù)2失敗】查詢訂單詳情異常", ex);
                    throw new RuntimeException("查詢訂單詳情失敗", ex);
                });
        // 步驟3：非阻塞處理最終結(jié)果（核心：thenAccept + thenRun）
        orderDetailFuture
                // 1. whenComplete：統(tǒng)一處理結(jié)果/異常（非阻塞）
                .whenComplete((detail, ex) -> {
                    if (ex != null) {
                        log.error("【級(jí)聯(lián)任務(wù)失敗】訂單處理異常", ex);
                        // 非阻塞處理失敗邏輯（如通知、落庫(kù)）
                        handleOrderFailure(userId, ex.getMessage());
                    } else {
                        log.info("【級(jí)聯(lián)任務(wù)成功】訂單詳情：{}", detail);
                    }
                })
                // 2. thenAccept：消費(fèi)結(jié)果（有返回值時(shí)用，非阻塞）
                .thenAccept(detail -> {
                    log.info("【非阻塞消費(fèi)】計(jì)算訂單金額 | 訂單ID：{}", detail.orderId());
                    double total = detail.quantity() * detail.price();
                    // 模擬：非阻塞更新訂單金額到數(shù)據(jù)庫(kù)
                    updateOrderAmount(detail.orderId(), total);
                })
                // 3. thenRun：無(wú)結(jié)果收尾動(dòng)作（非阻塞）
                .thenRun(() -> {
                    log.info("【非阻塞收尾】訂單處理流程結(jié)束 | 清理臨時(shí)資源/發(fā)送通知");
                    // 模擬：非阻塞發(fā)送處理完成通知
                    sendOrderNotify(userId, OrderStatus.SUCCESS);
                });
        // 關(guān)鍵：當(dāng)前線程（如主線程/Controller線程）不會(huì)阻塞，直接繼續(xù)執(zhí)行后續(xù)邏輯
        log.info("【主線程不阻塞】訂單處理請(qǐng)求已提交，當(dāng)前線程繼續(xù)執(zhí)行其他任務(wù)...");
    }
    // ------------------- 模擬業(yè)務(wù)方法 -------------------
    private void simulateDelay(long millis) {
        try {
            TimeUnit.MILLISECONDS.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("任務(wù)被中斷", e);
        }
    }
    private void handleOrderFailure(Long userId, String errorMsg) {
        log.warn("【失敗處理】用戶{}訂單處理失敗：{}", userId, errorMsg);
        // 實(shí)際場(chǎng)景：非阻塞發(fā)送失敗通知、記錄異常日志等
    }
    private void updateOrderAmount(String orderId, double total) {
        log.info("【更新金額】訂單{}總金額：{}元", orderId, total);
        // 實(shí)際場(chǎng)景：非阻塞調(diào)用DAO更新數(shù)據(jù)庫(kù)
    }
    private void sendOrderNotify(Long userId, OrderStatus status) {
        // Java 17+ switch表達(dá)式（簡(jiǎn)化分支邏輯）
        String msg = switch (status) {
            case SUCCESS -> "用戶" + userId + "訂單處理完成";
            case FAILURE -> "用戶" + userId + "訂單處理失敗";
        };
        log.info("【發(fā)送通知】{}", msg);
    }
    // ------------------- Java 17+ 特性 -------------------
    // Record：簡(jiǎn)化數(shù)據(jù)載體（自動(dòng)生成構(gòu)造器、getter、equals等）
    public record OrderDetail(String orderId, Long userId, String productName, int quantity, double price) {}
    // ------------------- 資源管理（Spring規(guī)范） -------------------
    // Spring 銷毀Bean時(shí)關(guān)閉線程池（try-with-resources 確保資源釋放）
    @PreDestroy
    public void destroy() {
        log.info("【關(guān)閉線程池】開始釋放異步線程池資源...");
        // try-with-resources 自動(dòng)關(guān)閉 ExecutorService（Java 9+ 支持）
        try (ExecutorService executor = customExecutor) {
            executor.shutdown();
            if (!executor.awaitTermination(3, TimeUnit.SECONDS)) {
                executor.shutdownNow();
                log.warn("【強(qiáng)制關(guān)閉】線程池未正常終止，強(qiáng)制關(guān)閉");
            }
        } catch (InterruptedException e) {
            customExecutor.shutdownNow();
            Thread.currentThread().interrupt();
            log.error("【關(guān)閉異?！烤€程池關(guān)閉被中斷", e);
        }
        log.info("【關(guān)閉完成】異步線程池已釋放");
    }
    // ------------------- 測(cè)試入口（非阻塞效果演示） -------------------
    public static void main(String[] args) {
        NonBlockingAsyncService service = new NonBlockingAsyncService();
        // 提交異步任務(wù)（非阻塞）
        service.processOrderNonBlocking(1001L);
        // 主線程繼續(xù)執(zhí)行（不會(huì)等待異步任務(wù)完成）
        log.info("【主線程】已提交訂單處理任務(wù)，開始執(zhí)行其他業(yè)務(wù)邏輯...");
        simulateMainThreadWork();
        // 等待異步任務(wù)完成（僅測(cè)試用，實(shí)際業(yè)務(wù)無(wú)需此步驟）
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
    private static void simulateMainThreadWork() {
        log.info("【主線程】執(zhí)行其他任務(wù)：檢查庫(kù)存、驗(yàn)證用戶權(quán)限...");
        try {
            TimeUnit.MILLISECONDS.sleep(500);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

【主線程不阻塞】訂單處理請(qǐng)求已提交，當(dāng)前線程繼續(xù)執(zhí)行其他任務(wù)...
【主線程】執(zhí)行其他任務(wù)：檢查庫(kù)存、驗(yàn)證用戶權(quán)限...
【異步任務(wù)1】生成訂單ID | 線程：pool-1-thread-1
【異步任務(wù)2】查詢訂單詳情 | 訂單ID：ORDER_1001_17338xxxxxx | 線程：pool-1-thread-2
【級(jí)聯(lián)任務(wù)成功】訂單詳情：OrderDetail[orderId=ORDER_1001_17338xxxxxx, userId=1001, productName=Java編程實(shí)戰(zhàn), quantity=2, price=89.9]
【非阻塞消費(fèi)】計(jì)算訂單金額 | 訂單ID：ORDER_1001_17338xxxxxx
【更新金額】訂單ORDER_1001_17338xxxxxx總金額：179.8元
【非阻塞收尾】訂單處理流程結(jié)束 | 清理臨時(shí)資源/發(fā)送通知
【發(fā)送通知】用戶1001訂單處理完成

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
/**
 * 線程池 + Future 異步示例
 * 核心：提交Callable獲取Future，通過(guò)get()阻塞獲取結(jié)果（可設(shè)置超時(shí)）
 */
public class FutureAsyncDemo {
    // 自定義線程池（Java 17+ 推薦使用Executors靜態(tài)工廠+try-with-resources關(guān)閉）
    private static final ExecutorService EXECUTOR = Executors.newVirtualThreadPerTaskExecutor(); // Java 21+ 虛擬線程池（17+需預(yù)覽）
    // Java 17 Record：異步任務(wù)結(jié)果
    public record TaskResult(Long taskId, String content, long costTimeMs) {}
    // 異步任務(wù)：生成業(yè)務(wù)數(shù)據(jù)
    private static Callable<TaskResult> generateDataTask(Long taskId) {
        return () -> {
            long start = System.currentTimeMillis();
            log("任務(wù){(diào)}開始執(zhí)行，線程：{}", taskId, Thread.currentThread().getName());
            // 模擬耗時(shí)
            TimeUnit.MILLISECONDS.sleep(200);
            long cost = System.currentTimeMillis() - start;
            return new TaskResult(taskId, "異步生成數(shù)據(jù)_" + taskId, cost);
        };
    }
    public static void main(String[] args) {
        // 提交異步任務(wù)
        Future<TaskResult> future1 = EXECUTOR.submit(generateDataTask(1L));
        Future<TaskResult> future2 = EXECUTOR.submit(generateDataTask(2L));
        // 非阻塞提交后，主線程可執(zhí)行其他邏輯
        log("主線程執(zhí)行其他任務(wù)...");
        // 阻塞獲取結(jié)果（可設(shè)置超時(shí)，避免永久阻塞）
        try {
            TaskResult result1 = future1.get(1, TimeUnit.SECONDS); // 超時(shí)時(shí)間1秒
            TaskResult result2 = future2.get(1, TimeUnit.SECONDS);
            log("任務(wù)1結(jié)果：{}", result1);
            log("任務(wù)2結(jié)果：{}", result2);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log("任務(wù)被中斷：{}", e.getMessage());
        } catch (ExecutionException e) {
            log("任務(wù)執(zhí)行失?。簕}", e.getCause().getMessage());
        } catch (TimeoutException e) {
            log("任務(wù)超時(shí)：{}", e.getMessage());
            future1.cancel(true); // 超時(shí)取消任務(wù)
            future2.cancel(true);
        } finally {
            // try-with-resources 關(guān)閉線程池（Java 9+ 支持ExecutorService自動(dòng)關(guān)閉）
            try (ExecutorService executor = EXECUTOR) {
                executor.shutdown();
                if (!executor.awaitTermination(3, TimeUnit.SECONDS)) {
                    executor.shutdownNow();
                    log("線程池強(qiáng)制關(guān)閉");
                }
            } catch (InterruptedException e) {
                EXECUTOR.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
    }
    private static void log(String msg, Object... args) {
        System.out.printf("[%s] %s%n", Thread.currentThread().getName(), String.format(msg, args));
    }
}

import java.time.Duration;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
 * 虛擬線程（VirtualThread）異步示例
 * Java 21+ 正式支持，17+ 需添加 --enable-preview 啟動(dòng)參數(shù)
 */
public class VirtualThreadAsyncDemo {
    // Java 17 Record：用戶數(shù)據(jù)
    public record UserData(Long userId, String username, String email) {}
    // 異步任務(wù)：查詢用戶信息
    private static UserData queryUser(Long userId) {
        try {
            // 模擬IO阻塞（虛擬線程阻塞時(shí)會(huì)釋放載體線程，無(wú)性能損耗）
            TimeUnit.MILLISECONDS.sleep(300);
            log("查詢用戶{}完成，線程：{}", userId, Thread.currentThread().getName());
            return new UserData(userId, "user_" + userId, userId + "@example.com");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("查詢用戶中斷", e);
        }
    }
    public static void main(String[] args) throws InterruptedException {
        // 虛擬線程池（Java 21+ 推薦）
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            // 提交1000個(gè)虛擬線程任務(wù)（輕量，無(wú)性能壓力）
            for (long i = 1; i <= 1000; i++) {
                long userId = i;
                executor.submit(() -> queryUser(userId));
            }
            // 主線程等待任務(wù)完成（僅測(cè)試用）
            executor.awaitTermination(Duration.ofSeconds(5));
            log("所有虛擬線程任務(wù)執(zhí)行完成");
        } // try-with-resources 自動(dòng)關(guān)閉線程池
    }
    private static void log(String msg, Object... args) {
        System.out.printf("[%s] %s%n", Thread.currentThread().getName(), String.format(msg, args));
    }
}

import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.channels.CompletionHandler;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.nio.charset.StandardCharsets;
/**
 * NIO.2 異步文件IO示例
 * 核心：CompletionHandler 回調(diào)處理異步結(jié)果，無(wú)線程阻塞
 */
public class AsyncFileIODemo {
    private static final Path FILE_PATH = Path.of("async_demo.txt");
    public static void main(String[] args) throws InterruptedException {
        // try-with-resources 管理異步文件通道（自動(dòng)關(guān)閉資源）
        try (AsynchronousFileChannel fileChannel = AsynchronousFileChannel.open(
                FILE_PATH,
                StandardOpenOption.READ,
                StandardOpenOption.WRITE,
                StandardOpenOption.CREATE
        )) {
            // 步驟1：異步寫入文件
            String content = "Java 17 異步IO示例：" + System.currentTimeMillis();
            ByteBuffer writeBuffer = ByteBuffer.wrap(content.getBytes(StandardCharsets.UTF_8));
            fileChannel.write(writeBuffer, 0, null, new CompletionHandler<Integer, Void>() {
                @Override
                public void completed(Integer bytesWritten, Void attachment) {
                    log("異步寫入完成，寫入字節(jié)數(shù)：{}", bytesWritten);
                    // 寫入完成后異步讀取文件
                    readFileAsync(fileChannel);
                }
                @Override
                public void failed(Throwable exc, Void attachment) {
                    log("異步寫入失敗：{}", exc.getMessage());
                }
            });
            // 主線程等待異步操作完成（僅測(cè)試用）
            Thread.sleep(2000);
        } catch (Exception e) {
            log("文件通道操作異常：{}", e.getMessage());
        }
    }
    // 異步讀取文件
    private static void readFileAsync(AsynchronousFileChannel fileChannel) {
        ByteBuffer readBuffer = ByteBuffer.allocate(1024);
        fileChannel.read(readBuffer, 0, null, new CompletionHandler<Integer, Void>() {
            @Override
            public void completed(Integer bytesRead, Void attachment) {
                if (bytesRead == -1) {
                    log("文件讀取完成（無(wú)數(shù)據(jù)）");
                    return;
                }
                readBuffer.flip();
                String content = StandardCharsets.UTF_8.decode(readBuffer).toString();
                log("異步讀取結(jié)果：{}", content);
            }
            @Override
            public void failed(Throwable exc, Void attachment) {
                log("異步讀取失?。簕}", exc.getMessage());
            }
        });
    }
    private static void log(String msg, Object... args) {
        System.out.printf("[%s] %s%n", Thread.currentThread().getName(), String.format(msg, args));
    }
}

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import java.time.Duration;
/**
 * Reactor 響應(yīng)式異步示例（Spring Boot 3.x + Java 17+）
 */
@Slf4j
@Service
public class ReactorAsyncDemo {
    // Java 17 Record：商品數(shù)據(jù)
    public record Product(Long id, String name, double price) {}
    // 異步查詢商品（Mono 表示單值異步結(jié)果）
    public Mono<Product> queryProductAsync(Long productId) {
        return Mono.fromSupplier(() -> {
                    log.info("查詢商品{}，線程：{}", productId, Thread.currentThread().getName());
                    // 模擬IO耗時(shí)
                    try {
                        Thread.sleep(200);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        throw new RuntimeException("查詢中斷", e);
                    }
                    return new Product(productId, "商品_" + productId, 99.9 + productId);
                })
                .subscribeOn(Schedulers.boundedElastic()) // 指定異步線程池
                .timeout(Duration.ofSeconds(1)) // 超時(shí)控制
                .onErrorResume(ex -> { // 異常兜底
                    log.error("查詢商品失?。簕}", ex.getMessage());
                    return Mono.just(new Product(productId, "默認(rèn)商品", 0.0));
                });
    }
    // 測(cè)試入口（Spring Boot 環(huán)境）
    public static void main(String[] args) {
        ReactorAsyncDemo service = new ReactorAsyncDemo();
        // 非阻塞消費(fèi)結(jié)果
        service.queryProductAsync(1001L)
                .doOnNext(product -> log.info("查詢結(jié)果：{}", product)) // 消費(fèi)結(jié)果
                .doOnTerminate(() -> log.info("異步任務(wù)結(jié)束")) // 收尾動(dòng)作
                .subscribe(); // 訂閱（觸發(fā)異步執(zhí)行）
        // 主線程等待（僅測(cè)試用）
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}