通過線程池方式改造Stream.parallel()并行流

作者：哪吒 2023-10-12 08:29:06

單線程池newSingleThreadExecutor()，只有一個核心線程的線程池，保證任務按FIFO順序一個個執行；固定線程數線程池newFixedThreadPool(10)，固定數量的可復用的線程數，來執行任務。當線程數達到最大核心線程數，則加入隊列等待有空閑線程時再執行。

大家好，我是哪吒。

上一篇簡單聊一聊公平鎖和非公平鎖，parallel并行流，提到了一個IntStream.rangeClosed并行流問題，很多小伙伴，對這個比較陌生，想用線程池的方式改造一下。

一、IntStream.rangeClosed并行流

@Data
public class LockTest1 {
    public static void main(String[] args) {
        IntStream.rangeClosed(1, 100000).parallel().forEach(i -> new LockTest1().increase());
        System.out.println(time);
    }

    private static int time = 0;

    private static Object lock = new Object();
    public void increase() {
        synchronized (lock) {
            time++;
        }
    }
}

二、線程池方式改造

不會那些新特性，還是原始的香啊，寫起代碼，得心應手。

1、創建線程池

這時候，有些小伙伴，又陷入了選擇恐懼癥。用哪個線程池比較好呢？

簡單回顧一下：

單線程池newSingleThreadExecutor()，只有一個核心線程的線程池，保證任務按FIFO順序一個個執行；
固定線程數線程池newFixedThreadPool(10)，固定數量的可復用的線程數，來執行任務。當線程數達到最大核心線程數，則加入隊列等待有空閑線程時再執行；
可緩存線程池newCachedThreadPool()，創建的都是非核心線程，而且最大線程數為Interge的最大值，空閑線程存活時間是1分鐘。如果有大量耗時的任務，則不適該創建方式，它只適用于生命周期短的任務；
固定線程數newScheduledThreadPool(10)，支持定時和周期性任務newScheduledThreadPool(10)，顧名思義，在固定線程數的前提下，添加了定時任務。

最常用的還是固定線程數線程池newFixedThreadPool(10)。

@Data
public class LockTest2 {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(200);
        for (int i = 0; i < 100000; i++) {
            Thread0926 thread = new Thread0926();
            executorService.execute(thread);
        }
        System.out.println(time);
    }

    private static int time = 0;
    private static Object lock = new Object();
    public void increase() {
        synchronized (lock) {
            time++;
        }
    }
}

2、線程類

public class Thread0926 implements Runnable{
    @Override
    public void run() {
        LockTest2 lockTest = new LockTest2();
        lockTest.increase();
    }
}

3、信心滿滿，走起來

我草，這不對啊，不應該是100000嘛？又把老子整不會了~

三、再次解決并發時i++原子性問題

上一篇測試過，使用synchronized代碼塊是可以解決i++線程安全問題的，這次怎么不好使了？

上面的代碼中，synchronized (lock)鎖住了time++，lock是靜態變量，所以屬于類級別的鎖。但是新建的線程是一個新的類，超出了鎖的范圍，所以失效。

那么，在當前類中，開啟線程，是不是就可以了呢？試一下

public class LockTest4 {
    private static int time = 0;
    private static Object lock = new Object();
    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < 200; i++) {
            Thread thread = new Thread(() -> {
                for (int j = 0; j < 500; j++) {
                    synchronized (lock) {
                        time++;
                    }
                }
            });
            thread.start();
        }
        Thread.sleep(3000);
        System.out.println(time);
    }
}

當然，在synchronized代碼塊中，使用synchronized (LockTest4.class)也是可以的，效果是一樣的。