// 順序流 vs 并行流
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);

// 1. 順序流（默認(rèn)）
int sumSequential = numbers.stream()
    .reduce(0, Integer::sum);

// 2. 并行流
int sumParallel = numbers.parallelStream()
    .reduce(0, Integer::sum);

// 3. 將順序流轉(zhuǎn)為并行流
int sumParallel2 = numbers.stream()
    .parallel()
    .reduce(0, Integer::sum);

// 并行流背后是這個(gè)框架
ForkJoinPool commonPool = ForkJoinPool.commonPool();
System.out.println("并行度: " + commonPool.getParallelism()); // 通常是CPU核心數(shù)-1

線程A：│任務(wù)1│任務(wù)2│任務(wù)3│任務(wù)4│
線程B：│任務(wù)5│任務(wù)6│          │ ← 空閑
              ↓
線程B從線程A的隊(duì)尾"偷取"任務(wù)3來(lái)執(zhí)行

// 并行流自動(dòng)將數(shù)據(jù)拆分為多個(gè)子任務(wù)
Spliterator<T> spliterator = list.spliterator();

// 查看分割特性
int characteristics = spliterator.characteristics();
// SIZED: 知道確切大小
// SUBSIZED: 分割后的大小也確切
// ORDERED: 有順序要求

// 方式1：從集合創(chuàng)建
List<String> list = Arrays.asList("a", "b", "c");
Stream<String> parallelStream = list.parallelStream();

// 方式2：將順序流轉(zhuǎn)為并行流
Stream<String> parallelStream2 = list.stream().parallel();

// 方式3：從數(shù)組創(chuàng)建
String[] array = {"a", "b", "c"};
Stream<String> parallelStream3 = Arrays.stream(array).parallel();

List<String> words = Arrays.asList("hello", "world", "java", "stream");

// 順序處理
List<String> upperCaseSequential = words.stream()
    .map(String::toUpperCase)
    .collect(Collectors.toList());

// 并行處理
List<String> upperCaseParallel = words.parallelStream()
    .map(String::toUpperCase)
    .collect(Collectors.toList());

List<Integer> numbers = IntStream.range(1, 1000000).boxed().collect(Collectors.toList());

// 并行過(guò)濾偶數(shù)
List<Integer> evenNumbers = numbers.parallelStream()
    .filter(n -> n % 2 == 0)
    .collect(Collectors.toList());

// 計(jì)算1到1000000的和（并行版本更快）
long sum = LongStream.rangeClosed(1, 1000000)
    .parallel()
    .reduce(0, Long::sum);

private List<Map<String, String>> getInterlockResultWen13or24Level1PointValueParallel(
    Map<String, String> map, 
    List<IntermediateTable> list
) {
    // 1. 建立查找表（提升效率）
    Map<String, String> codeToTagNumber = list.stream()
        .collect(Collectors.toMap(
            IntermediateTable::getInterlockWen13or24Level1Point,
            IntermediateTable::getTagNumber,
            (v1, v2) -> v1  // 如果有重復(fù)key，取第一個(gè)
        ));
    
    // 2. 使用并行流處理
    return map.entrySet()
        .parallelStream()  // ← 關(guān)鍵！開(kāi)啟并行
        .map(entry -> {
            String code = entry.getKey();
            String name = entry.getValue();
            String tagNumber = codeToTagNumber.get(code);
            
            // 3. 執(zhí)行IO操作（HTTP請(qǐng)求）
            String value = tagNumber != null ? getSupOSValue(tagNumber) : "0";
            
            // 4. 構(gòu)建結(jié)果
            Map<String, String> resMap = new LinkedHashMap<>();
            resMap.put("name", name);
            resMap.put("value", value);
            return resMap;
        })
        .collect(Collectors.toList());  // 收集結(jié)果
}

public class ParallelStreamBenchmark {
    
    public static void main(String[] args) {
        // 模擬24次HTTP請(qǐng)求（用睡眠模擬）
        List<Integer> requestIds = IntStream.range(1, 25).boxed().collect(Collectors.toList());
        
        // 串行執(zhí)行
        long start = System.currentTimeMillis();
        List<String> sequentialResults = requestIds.stream()
            .map(id -> mockHttpRequest(id))
            .collect(Collectors.toList());
        long sequentialTime = System.currentTimeMillis() - start;
        
        // 并行執(zhí)行
        start = System.currentTimeMillis();
        List<String> parallelResults = requestIds.parallelStream()
            .map(id -> mockHttpRequest(id))
            .collect(Collectors.toList());
        long parallelTime = System.currentTimeMillis() - start;
        
        System.out.println("串行耗時(shí): " + sequentialTime + "ms");
        System.out.println("并行耗時(shí): " + parallelTime + "ms");
        System.out.println("加速比: " + (double)sequentialTime/parallelTime);
    }
    
    private static String mockHttpRequest(Integer id) {
        try {
            // 模擬HTTP請(qǐng)求延遲（100-500ms隨機(jī)）
            Thread.sleep(100 + (long)(Math.random() * 400));
            return "Response_" + id;
        } catch (InterruptedException e) {
            return "Error";
        }
    }
}

// ? 適合的場(chǎng)景
- 數(shù)據(jù)量大（>1000個(gè)元素）
- 每個(gè)元素的處理是獨(dú)立的
- 處理操作是CPU密集型或IO密集型
- 沒(méi)有順序要求（或者可以使用unordered()放棄順序）

// ? 不適合的場(chǎng)景
- 數(shù)據(jù)量小（<100個(gè)元素）
- 處理操作簡(jiǎn)單（如簡(jiǎn)單的數(shù)學(xué)運(yùn)算）
- 需要嚴(yán)格保持順序
- 操作有共享狀態(tài)（線程不安全）

// ? 錯(cuò)誤示例：共享可變狀態(tài)
List<Integer> sharedList = new ArrayList<>();
numbers.parallelStream()
    .forEach(sharedList::add);  // 并發(fā)修改異常！

// ? 正確做法：使用線程安全收集器
List<Integer> safeList = numbers.parallelStream()
    .collect(Collectors.toList());
    
// ? 或者使用同步集合
List<Integer> synchronizedList = Collections.synchronizedList(new ArrayList<>());
numbers.parallelStream()
    .forEach(synchronizedList::add);

// 1. 使用基礎(chǔ)類(lèi)型流避免裝箱拆箱
IntStream.range(1, 1000000)
    .parallel()
    .sum();

// 2. 使用unordered()放棄順序約束（可能提升性能）
list.parallelStream()
    .unordered()
    .filter(...)
    .collect(Collectors.toList());

// 3. 合并操作（combiner）必須是可結(jié)合的
// reduce操作需要：a ⊕ b ⊕ c = a ⊕ (b ⊕ c)

// 方法1：修改系統(tǒng)屬性
System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "8");

// 方法2：使用自定義ForkJoinPool
ForkJoinPool customPool = new ForkJoinPool(4);
List<String> results = customPool.submit(() -> 
    list.parallelStream()
        .map(item -> process(item))
        .collect(Collectors.toList())
).get();

private List<Map<String, String>> getInterlockResultWen13or24Level1PointValueOptimized(
    Map<String, String> map, 
    List<IntermediateTable> list
) {
    // 1. 建立索引
    Map<String, String> codeToTag = list.stream()
        .collect(Collectors.toMap(
            IntermediateTable::getInterlockWen13or24Level1Point,
            IntermediateTable::getTagNumber,
            (v1, v2) -> v1
        ));
    
    // 2. 預(yù)計(jì)算需要請(qǐng)求的tagNumbers
    List<String> tagNumbers = map.keySet().stream()
        .map(codeToTag::get)
        .filter(Objects::nonNull)
        .distinct()
        .collect(Collectors.toList());
    
    // 3. 并行獲取所有值
    Map<String, String> valueCache = tagNumbers.parallelStream()
        .collect(Collectors.toMap(
            tag -> tag,
            this::getSupOSValue  // 這是你的HTTP請(qǐng)求方法
        ));
    
    // 4. 構(gòu)建結(jié)果
    return map.entrySet().stream()
        .map(entry -> {
            String code = entry.getKey();
            String tag = codeToTag.get(code);
            String value = tag != null ? valueCache.getOrDefault(tag, "0") : "0";
            
            Map<String, String> result = new LinkedHashMap<>();
            result.put("name", entry.getValue());
            result.put("value", value);
            return result;
        })
        .collect(Collectors.toList());
}

// 1. 查看并行流使用的線程
List<String> results = list.parallelStream()
    .peek(item -> System.out.println(Thread.currentThread().getName()))
    .map(...)
    .collect(Collectors.toList());

// 2. 監(jiān)控執(zhí)行時(shí)間
long start = System.nanoTime();
List<String> results = list.parallelStream()
    .map(...)
    .collect(Collectors.toList());
long duration = System.nanoTime() - start;

// 3. 使用JMH進(jìn)行基準(zhǔn)測(cè)試（推薦）
// 添加依賴(lài)：org.openjdk.jmh

// 陷阱1：副作用操作
List<String> list = Arrays.asList("a", "b", "c");
StringBuilder builder = new StringBuilder();
list.parallelStream().forEach(builder::append); // ? 線程不安全

// 陷阱2：有狀態(tài)lambda
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
List<Integer> results = new ArrayList<>();
numbers.parallelStream()
    .filter(n -> {
        results.add(n); // ? 在lambda中修改外部狀態(tài)
        return n % 2 == 0;
    })
    .collect(Collectors.toList());

// 陷阱3：順序依賴(lài)
Optional<Integer> first = numbers.parallelStream()
    .filter(n -> n > 3)
    .findFirst(); // ?? 并行下findFirst可能變慢