并发设计模式实战系列(20)：扇出/扇入模式（Fan-Out/Fan-In）(完结篇)

┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│   Producer   │ ──> │  Workers    │ ──> │  Collector  │
│ (Main Task)  │     │ (并行处理)   │     │ (结果聚合)   │
└─────────────┘     └─────────────┘     └─────────────┘
      │                   ▲                   ▲
      └───────────────────┴───────────────────┘
            Fan-Out                  Fan-In

import java.util.*;
import java.util.concurrent.*;
import java.util.stream.*;

public class FanOutFanInDemo {

    // 模拟CPU密集型任务
    static int processItem(int item) {
        try {
            Thread.sleep(ThreadLocalRandom.current().nextInt(100));
        } catch (InterruptedException e) {}
        return item * item; // 平方计算
    }

    public static void main(String[] args) throws Exception {
        List<Integer> inputs = IntStream.rangeClosed(1, 100).boxed().collect(Collectors.toList());
        
        // 方法1：使用CompletableFuture（自动Fan-In）
        System.out.println("=== CompletableFuture方案 ===");
        long start = System.currentTimeMillis();
        CompletableFuture<Void> future = CompletableFuture.allOf(
            inputs.stream()
                .map(item -> CompletableFuture.supplyAsync(() -> processItem(item)))
                .toArray(CompletableFuture[]::new)
        ).thenApply(ignored -> {
            System.out.println("所有任务完成");
            return null;
        });
        future.get();
        System.out.printf("耗时: %dms\n", System.currentTimeMillis() - start);

        // 方法2：使用ForkJoinPool（工作窃取）
        System.out.println("\n=== ForkJoinPool方案 ===");
        ForkJoinPool pool = new ForkJoinPool(4);
        start = System.currentTimeMillis();
        List<Integer> results = pool.submit(() -> 
            inputs.parallelStream()
                .map(FanOutFanInDemo::processItem)
                .collect(Collectors.toList())
        ).get();
        System.out.println("结果数量: " + results.size());
        System.out.printf("耗时: %dms\n", System.currentTimeMillis() - start);

        // 方法3：手动控制（精确管理）
        System.out.println("\n=== 手动控制方案 ===");
        ExecutorService workers = Executors.newFixedThreadPool(4);
        CompletionService<Integer> completionService = new ExecutorCompletionService<>(workers);
        start = System.currentTimeMillis();
        
        // Fan-Out阶段
        inputs.forEach(item -> completionService.submit(() -> processItem(item)));
        
        // Fan-In阶段
        List<Integer> orderedResults = new ArrayList<>(inputs.size());
        for (int i = 0; i < inputs.size(); i++) {
            try {
                orderedResults.add(completionService.take().get());
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
        System.out.println("结果数量: " + orderedResults.size());
        System.out.printf("耗时: %dms\n", System.currentTimeMillis() - start);
        workers.shutdown();
    }
}

// 1. CompletableFuture默认使用ForkJoinPool.commonPool()
// 可指定自定义线程池：
CompletableFuture.supplyAsync(() -> task, customPool);

// 2. ForkJoinPool参数选择：
// - 并行度通常设为CPU核心数
new ForkJoinPool(Runtime.getRuntime().availableProcessors());

// 3. 手动方案背压控制：
// 使用有界队列+CallerRunsPolicy
new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS,
    new ArrayBlockingQueue<>(100), 
    new ThreadPoolExecutor.CallerRunsPolicy());

// 根据系统负载调整批次大小
int batchSize = Runtime.getRuntime().availableProcessors() * 2;
List<List<Integer>> batches = Lists.partition(inputs, batchSize);

// 为CompletableFuture添加异常处理
future.exceptionally(ex -> {
    System.err.println("任务失败: " + ex.getMessage());
    return null;
});

// IO密集型 + CPU密集型混合
ExecutorService ioPool = Executors.newCachedThreadPool();
ExecutorService cpuPool = Executors.newWorkStealingPool();

CompletableFuture.supplyAsync(() -> queryDB(), ioPool)
    .thenApplyAsync(data -> processData(data), cpuPool);

// ForkJoinPool监控
ForkJoinPool pool = (ForkJoinPool) ForkJoinPool.commonPool();
System.out.println("活跃线程数: " + pool.getActiveThreadCount());
System.out.println("窃取任务数: " + pool.getStealCount());

┌─────────┐     ┌─────────┐     ┌─────────┐  
│ Stage 1  │───> │ Stage 2  │───> │ Stage 3  │  
│ (Fan-Out)│<─── │ (Fan-Out)│<─── │ (Fan-In) │  
└─────────┘     └─────────┘     └─────────┘  

// 使用Phaser协调多阶段  
Phaser phaser = new Phaser(3);  
ExecutorService[] stagePools = {Executors.newFixedThreadPool(4), ...};  

// Stage 1  
stagePools[0].submit(() -> {  
    List<Data> stage1Results = dataList.parallelStream()  
        .map(d -> processStage1(d)).collect(Collectors.toList());  
    phaser.arriveAndAwaitAdvance();  
    return stage1Results;  
});  

// Stage 2同理...  

CompletableFuture<?>[] futures = inputList.stream()  
    .map(input -> {  
        if (needFork(input)) {  
            return CompletableFuture.allOf(  
                taskA(input),  
                taskB(input)  
            );  
        } else {  
            return taskC(input);  
        }  
    }).toArray(CompletableFuture[]::new);  

// Reactor + Fan-Out  
Flux.fromIterable(inputList)  
    .parallel()  
    .runOn(Schedulers.parallel())  
    .map(item -> process(item))  // Fan-Out  
    .sequential()  
    .collectList()               // Fan-In  
    .subscribe(System.out::println);  

┌──────────────┐     ┌──────────────┐     ┌──────────────┐  
│ Global Queue  │───> │ Local Queues │───> │ Workers      │  
│ (优先级队列)    │<─── │ (工作窃取)    │<─── │ (弹性线程池)  │  
└──────────────┘     └──────────────┘     └──────────────┘  

// Fan-Out速率  
Counter.Builder("fanout_requests_total")  
    .labelNames("source")  
    .register(registry);  

// Fan-In延迟直方图  
Histogram.Builder("fanin_duration_seconds")  
    .buckets(0.1, 0.5, 1)  
    .register(registry);  

graph TD  
    A[需要严格顺序?] -->|是| B[选择手动控制方案]  
    A -->|否| C{任务类型?}  
    C -->|CPU密集型| D[ForkJoinPool]  
    C -->|IO密集型| E[CompletableFuture+自定义线程池]  
    C -->|混合型| F[分阶段差异化处理]