通过线程池方式改造Stream.parallel()并行流

news2024/12/25 9:05:18

在这里插入图片描述

一、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);
    }
}