并发编程

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.IntStream;

// 1. 定义任务（实现Runnable，无返回值）
class PrintTask implements Runnable {
    private int id;
    public PrintTask(int id) {
        this.id = id;
    }
    @Override
    public void run() {
        // 模拟任务执行（比如处理数据）
        try {
            Thread.sleep(100); // 休眠100ms，模拟耗时
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("任务" + id + "完成，线程：" + Thread.currentThread().getName());
    }
}

public class SingleThreadDemo {
    public static void main(String[] args) {
        // 2. 创建单线程池
        ExecutorService executor = Executors.newSingleThreadExecutor();
        
        // 3. 提交10个任务
        IntStream.range(0, 10).forEach(i -> executor.execute(new PrintTask(i)));
        
        // 4. 关闭线程池（重要：用完必须关，否则程序不会结束）
        executor.shutdown();
    }
}

import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

// 1. 定义有返回值的任务（实现Callable）
class SumTask implements Callable<Integer> {
    private int start;
    private int end;
    public SumTask(int start, int end) {
        this.start = start;
        this.end = end;
    }
    @Override
    public Integer call() {
        int sum = 0;
        for (int i = start; i <= end; i++) {
            sum += i;
        }
        System.out.println("计算" + start + "-" + end + "，线程：" + Thread.currentThread().getName());
        return sum;
    }
}

public class CachedThreadPoolDemo {
    public static void main(String[] args) throws Exception {
        // 2. 创建缓存线程池
        ExecutorService executor = Executors.newCachedThreadPool();
        
        // 3. 提交3个计算任务（分别计算1-100，101-200，201-300的和）
        List<SumTask> tasks = IntStream.range(0, 3)
                .mapToObj(i -> new SumTask(i * 100 + 1, (i + 1) * 100))
                .collect(Collectors.toList());
        
        // 4. 执行所有任务并获取结果（invokeAll会等待所有任务完成）
        List<Future<Integer>> results = executor.invokeAll(tasks);
        
        // 5. 汇总结果
        int total = 0;
        for (Future<Integer> result : results) {
            total += result.get(); // get()会获取返回值，若未完成则等待
        }
        System.out.println("总和：" + total); // 结果应为45150
        
        // 6. 关闭线程池
        executor.shutdown();
    }
}

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

public class BasicUsage {
    public static void main(String[] args) throws Exception {
        System.out.println("主线程开始...");

        // 模拟一个耗时操作，返回一个字符串
        Supplier<String> supplier = () -> {
            try {
                TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(1, 4)); // 模拟1-3秒延迟
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            }
            return "异步操作的结果";
        };

        CompletableFuture<String> future = CompletableFuture.supplyAsync(supplier);

        // 注册回调函数，当异步操作完成时执行
        future.thenAccept(result -> {
            System.out.println("异步操作完成，结果是：" + result);
        });

        System.out.println("主线程继续执行，不等待异步操作...");

        // 为了演示，让主线程等待一下异步操作完成
        future.join(); // 阻塞当前线程直到CompletableFuture完成并获取结果
        // 也可以使用 future.get(); 但它会抛出检查性异常
        // future.get(5, TimeUnit.SECONDS); // 可以设置超时
        System.out.println("主线程结束。");
    }
}

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

public class RunAsyncExample {
    public static void main(String[] args) throws Exception {
        System.out.println("主线程开始...");

        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(2); // 模拟2秒延迟
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            }
            System.out.println("异步操作完成，没有返回结果。");
        });
        System.out.println("主线程任务不需要等待，执行主线的任务。");
        future.join(); // 等待异步操作完成
        System.out.println("主线程结束。");
    }
}

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ThreadLocalRandom;

public class CombineCompletableFutures {

    public static void main(String[] args) throws Exception {
        System.out.println("主线程开始...");

        CompletableFuture<String> fetchUserData = CompletableFuture.supplyAsync(() -> {
            System.out.println("第一步：正在获取用户数据...");
            try {
                TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(1, 3));
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            }
            return "用户ID:123, 名称:张三";
        });

        CompletableFuture<Double> calculateDiscount = fetchUserData.thenApplyAsync(userData -> {
            System.out.println("第二步：根据用户数据计算折扣...");
            // 假设从用户数据中提取ID并计算折扣
            String userId = userData.split(",")[0].split(":")[1];
            try {
                TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(1, 3));
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            }
            return Double.parseDouble(userId) * 0.01; // 模拟折扣
        });

        CompletableFuture<String> applyDiscountAndGenerateInvoice = calculateDiscount.thenApplyAsync(discount -> {
            System.out.println("第三步：应用折扣并生成发票...");
            try {
                TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(1, 3));
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            }
            return "发票已生成，折扣为：" + String.format("%.2f", discount * 100) + "%";
        });

        applyDiscountAndGenerateInvoice.thenAccept(invoice -> {
            System.out.println("最终结果：" + invoice);
        }).join(); // 等待所有链式操作完成

        System.out.println("主线程结束。");
    }
}

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

public class AllOfExample {
    public static void main(String[] args) throws Exception {
        System.out.println("主线程开始...");

        CompletableFuture<String> task1 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            }
            return "任务1完成";
        });

        CompletableFuture<String> task2 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            }
            return "任务2完成";
        });

        CompletableFuture<String> task3 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            }
            return "任务3完成";
        });

        CompletableFuture<Void> allTasks = CompletableFuture.allOf(task1, task2, task3);

        allTasks.thenRun(() -> {
            try {
                // 当所有任务都完成后，才能安全地获取它们的结果
                System.out.println(task1.join());
                System.out.println(task2.join());
                System.out.println(task3.join());
                System.out.println("所有任务都已完成！");
            } catch (Exception e) {
                System.err.println("有任务异常终止: " + e.getMessage());
            }
        }).join();

        System.out.println("主线程结束。");
    }
}

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ThreadLocalRandom;

public class AnyOfExample {
    public static void main(String[] args) throws Exception {
        System.out.println("主线程开始...");

        CompletableFuture<String> fastTask = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(1, 3));
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            }
            return "这是最快的任务！";
        });

        CompletableFuture<String> slowTask = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(4, 6));
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            }
            return "这个任务有点慢。";
        });

        CompletableFuture<Object> anyTask = CompletableFuture.anyOf(fastTask, slowTask);

        anyTask.thenAccept(result -> {
            System.out.println("最先完成的任务结果是: " + result);
        }).join();

        System.out.println("主线程结束。");
    }
}

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ThreadLocalRandom;

public class SimulateLongRunningOperation {

    // 模拟从数据库获取用户信息的服务
    public static CompletableFuture<String> getUserInfo(String userId) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("正在从数据库获取用户 " + userId + " 的信息...");
            try {
                TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(1, 4)); // 模拟DB查询延迟
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            }
            return "用户信息: {id: " + userId + ", name: John Doe}";
        });
    }

    // 模拟调用第三方API获取产品价格的服务
    public static CompletableFuture<Double> getProductPrice(String productId) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("正在从第三方API获取产品 " + productId + " 的价格...");
            try {
                TimeUnit.SECONDS.sleep(ThreadLocalRandom.current().nextInt(2, 5)); // 模拟API调用延迟
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            }
            return 99.99; // 模拟产品价格
        });
    }

    public static void main(String[] args) throws Exception {
        System.out.println("主线程开始...");

        // 并行获取用户信息和产品价格
        CompletableFuture<String> userInfoFuture = getUserInfo("abc");
        CompletableFuture<Double> productPriceFuture = getProductPrice("xyz");

        // 当两者都完成后，组合结果
        CompletableFuture<String> combinedFuture = userInfoFuture.thenCombine(productPriceFuture, (userInfo, price) -> {
            System.out.println("所有异步数据获取完成，正在组合信息...");
            return "用户数据: [" + userInfo + "], 产品价格: [" + price + "]";
        });

        // 打印最终结果
        combinedFuture.thenAccept(finalResult -> {
            System.out.println("最终报告: " + finalResult);
        }).join();

        System.out.println("主线程结束。");
    }
}

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ThreadLocalRandom;

public class ExceptionHandlingExceptionally {
    public static void main(String[] args) throws Exception {
        System.out.println("主线程开始...");

        CompletableFuture<String> futureWithException = CompletableFuture.supplyAsync(() -> {
            System.out.println("模拟一个可能失败的异步操作...");
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            }
            if (ThreadLocalRandom.current().nextBoolean()) {
                System.out.println("操作成功。");
                return "成功结果";
            } else {
                System.err.println("操作失败，抛出异常！");
                throw new RuntimeException("模拟的业务异常：数据获取失败");
            }
        });

        futureWithException
            .exceptionally(ex -> {
                System.err.println("捕获到异常：" + ex.getMessage());
                return "备用结果：操作失败，使用默认值。";
            })
            .thenAccept(result -> {
                System.out.println("处理后的结果：" + result);
            }).join();

        System.out.println("主线程结束。");
    }
}

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ThreadLocalRandom;

public class ExceptionHandlingHandle {
    public static void main(String[] args) throws Exception {
        System.out.println("主线程开始...");

        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("模拟一个可能失败的异步操作...");
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            }
            if (ThreadLocalRandom.current().nextBoolean()) {
                return "操作成功的结果";
            } else {
                throw new RuntimeException("模拟的业务异常：网络连接中断");
            }
        });

        future
            .handle((result, ex) -> {
                if (ex != null) {
                    System.err.println("在 handle 中捕获到异常：" + ex.getMessage());
                    return "错误处理结果：" + ex.getMessage();
                } else {
                    System.out.println("在 handle 中处理成功结果：" + result);
                    return "成功处理结果：" + result;
                }
            })
            .thenAccept(finalResult -> {
                System.out.println("最终消费结果：" + finalResult);
            }).join();

        System.out.println("主线程结束。");
    }
}

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ThreadLocalRandom;

public class ExceptionHandlingWhenComplete {
    public static void main(String[] args) throws Exception {
        System.out.println("主线程开始...");

        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("模拟一个可能失败的异步操作...");
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            }
            if (ThreadLocalRandom.current().nextBoolean()) {
                return "原始的成功结果";
            } else {
                throw new RuntimeException("模拟的业务异常：服务不可用");
            }
        });

        future
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    System.err.println("在 whenComplete 中发现异常：" + ex.getMessage());
                    // 这里可以记录日志、发送通知等，但不会改变 future 的结果
                } else {
                    System.out.println("在 whenComplete 中发现成功结果：" + result);
                }
            })
            .exceptionally(ex -> { // whenComplete 不会消费异常，异常会继续向下传递
                System.err.println("在 exceptionally 中捕获到异常（来自原始 future）：" + ex.getMessage());
                return "备用结果因为原始操作失败";
            })
            .thenAccept(finalResult -> {
                System.out.println("最终消费结果：" + finalResult);
            }).join();

        System.out.println("主线程结束。");
    }
}

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

public class ChainedExceptionHandler {
    public static void main(String[] args) throws Exception {
        System.out.println("主线程开始...");

        CompletableFuture<String> step1 = CompletableFuture.supplyAsync(() -> {
            System.out.println("步骤1：开始...");
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            }
            // 模拟在第一步抛出异常
            if (true) { // 总是抛出异常
                 throw new RuntimeException("步骤1：发生错误！");
            }
            return "数据A";
        });

        CompletableFuture<String> step2 = step1.thenApply(dataA -> {
            System.out.println("步骤2：接收到 " + dataA + "，开始处理...");
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            }
            return dataA + " -> 数据B";
        });

        CompletableFuture<String> step3 = step2.thenApply(dataB -> {
            System.out.println("步骤3：接收到 " + dataB + "，开始处理...");
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            }
            return dataB + " -> 数据C";
        });

        step3
            .exceptionally(ex -> {
                System.err.println("在链末端捕获到异常：" + ex.getMessage());
                return "错误！链式操作中止。";
            })
            .thenAccept(finalResult -> {
                System.out.println("最终结果：" + finalResult);
            }).join();

        System.out.println("主线程结束。");
    }
}

import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ExecutionException;

public class CheckedExceptionHandling {

    // 模拟一个会抛出检查性异常的方法
    public static String readFileContent() throws IOException {
        System.out.println("正在模拟读取文件...");
        if (System.currentTimeMillis() % 2 == 0) { // 模拟文件不存在或权限问题
            throw new IOException("文件不存在或无法访问！");
        }
        return "文件内容：Hello CompletableFuture!";
    }

    public static void main(String[] args) throws Exception {
        System.out.println("主线程开始...");

        CompletableFuture<String> fileReadFuture = CompletableFuture.supplyAsync(() -> {
            try {
                return readFileContent(); // 调用可能抛出检查性异常的方法
            } catch (IOException e) {
                // 必须在这里捕获检查性异常，并将其包装为非检查性异常
                // 否则编译会失败
                throw new RuntimeException("读取文件时发生错误: " + e.getMessage(), e);
            }
        });

        fileReadFuture
            .exceptionally(ex -> {
                Throwable cause = ex.getCause(); // 获取原始异常
                if (cause instanceof IOException) {
                    System.err.println("捕获到 IOException：" + cause.getMessage());
                } else {
                    System.err.println("捕获到其他异常：" + ex.getMessage());
                }
                return "文件读取失败，使用默认内容。";
            })
            .thenAccept(content -> {
                System.out.println("处理后的文件内容：" + content);
            }).join();

        // 另一种直接处理异常的方式是使用 get() 或 join()，但它们会抛出 ExecutionException/CompletionException
        try {
            // 注意：这里需要 try-catch ExecutionException
            String result = fileReadFuture.get();
            System.out.println("通过 get() 获取的结果：" + result);
        } catch (ExecutionException e) {
            System.err.println("通过 get() 捕获到 ExecutionException：" + e.getCause().getMessage());
        }


        System.out.println("主线程结束。");
    }
}