java高并发编程（五）线程池

2022-05-31 14:12:54

摘自马士兵java并发编程

一、认识Executor、ExecutorService、Callable、Executors

/**

 * 认识Executor

 */

package yxxy.c_026;

import java.util.concurrent.Executor;

public class T01_MyExecutor implements Executor {

    public static void main(String[] args) {

        new T01_MyExecutor().execute(new Runnable(){

            @Override

            public void run() {

                System.out.println("hello executor");

            }

        });

    }

    @Override

    public void execute(Runnable command) {

        //new Thread(command).run();

        command.run();

    }

}

Executor执行器是一个接口，只有一个方法execute执行任务，在java的线程池的框架里边，这个是最顶层的接口；

ExecutorService：从Executor接口继承。

Callable：里面call方法，和Runnable接口很像，设计出来都是被其他线程调用的；但是Runnable接口里面run方法是没有返回值的也不能抛出异常；而call方法有返回值可以抛异常；

Executors: 操作Executor的一个工具类；以及操作ExecutorService，ThreadFactory，Callable等；

二、ThreadPool:　　　　　　

/**

 * 线程池的概念

 */

package yxxy.c_026;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.TimeUnit;

public class T05_ThreadPool {

    public static void main(String[] args) throws InterruptedException {

        ExecutorService service = Executors.newFixedThreadPool(5); //execute submit

        for (int i = 0; i < 6; i++) {

            service.execute(() -> {

                try {

                    TimeUnit.MILLISECONDS.sleep(500);

                } catch (InterruptedException e) {

                    e.printStackTrace();

                }

                System.out.println(Thread.currentThread().getName());

            });

        }

        System.out.println(service);

        service.shutdown();

        System.out.println(service.isTerminated());

        System.out.println(service.isShutdown());

        System.out.println(service);

        TimeUnit.SECONDS.sleep(5);

        System.out.println(service.isTerminated());

        System.out.println(service.isShutdown());

        System.out.println(service);

    }

}

console:

java.util.concurrent.ThreadPoolExecutor@53d8d10a[Running, pool size = 5, active threads = 5, queued tasks = 1, completed tasks = 0]

false

true

java.util.concurrent.ThreadPoolExecutor@53d8d10a[Shutting down, pool size = 5, active threads = 5, queued tasks = 1, completed tasks = 0]

pool-1-thread-1

pool-1-thread-3

pool-1-thread-2

pool-1-thread-5

pool-1-thread-4

pool-1-thread-1

true

true

java.util.concurrent.ThreadPoolExecutor@53d8d10a[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 6]

创建了一个线程池，扔了5个线程，接下来要执行6个任务，扔进去线程池里面就启一个线程帮你执行一个，因为这里最多就起5个线程，接下来扔第6个任务的时候，不好意思，它排队了，排在线程池所维护的一个任务队列里面，任务队列大多数使用的都是BlockingQueue，这是线程池的概念；

有什么好处？好处在于如果这个任务执行完了，这个线程不会消失，它执行完任务空闲下来了，如果有新的任务来的时候，直接交给这个线程来运行就行了，不需要新启动线程；从这个概念上讲，如果你的任务和线程池线程数量控制的比较好的情况下，你不需要启动新的线程就能执行很多很多的任务，效率会比较高，并发性好；

service.shutdown()：关闭线程池，shutdown是正常的关闭，它会等所有的任务都执行完才会关闭掉；还有一个是shutdownNow，二话不说直接就给关了，不管线程有没有执行完；

service.isTerminated(): 代表的是这里所有执行的任务是不是都执行完了。isShutdown()为true，注意它关了但并不代表它执行完了，只是代表正在关闭的过程之中(注意打印Shutting down)

打印5个线程名字，而且第一个线程执行完了之后，第6个任务来了，第1个线程继续执行，不会有线程6；

当所有线程全部执行完毕之后，线程池的状态为Terminated，表示正常结束，complete tasks=6

线程池里面维护了很多线程，等着你往里扔任务，而扔任务的时候它可以维护着一个任务列表，还没有被执行的任务列表，同样的它还维护着另外一个队列，complete tasks，结束的任务队列，任务执行结束扔到这个队列里，所以，一个线程池维护着两个队列；

三、Future

/**

 * 认识future

 */

package yxxy.c_026;

import java.util.concurrent.ExecutionException;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.Future;

import java.util.concurrent.FutureTask;

import java.util.concurrent.TimeUnit;

public class T06_Future {

    public static void main(String[] args) throws InterruptedException, ExecutionException {

        /*FutureTask<Integer> task = new FutureTask<Integer>(new Callable<Integer>(){

            @Override

            public Integer call() throws Exception {

                TimeUnit.MILLISECONDS.sleep(3000);

                return 1000;

            }

        });*/

        FutureTask<Integer> task = new FutureTask<>(()->{

            TimeUnit.MILLISECONDS.sleep(3000);

            return 1000;

        });

        new Thread(task).start();

        System.out.println(task.get()); //阻塞

        //*******************************

        ExecutorService service = Executors.newFixedThreadPool(5);

        Future<Integer> f = service.submit(()->{

            TimeUnit.MILLISECONDS.sleep(5000);

            return 1;

        });

        System.out.println(f.isDone());

        System.out.println(f.get());

        System.out.println(f.isDone());

    }

}

1000

false

1

true

Future: ExecutorService里面有submit方法，它的返回值是Future类型，因为你扔一个任务进去需要执行一段时间，未来的某一个时间点上，任务执行完了产生给你一个结果，这个Future代表的就是那个Callable的返回值；

四、并行计算的例子：　　　　　　

/**

 * 线程池的概念

 * nasa

 */

package yxxy.c_026;

import java.util.ArrayList;

import java.util.List;

import java.util.concurrent.Callable;

import java.util.concurrent.ExecutionException;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.Future;

public class T07_ParallelComputing {

    public static void main(String[] args) throws InterruptedException, ExecutionException {

        long start = System.currentTimeMillis();

        List<Integer> results = getPrime(1, 200000);

        long end = System.currentTimeMillis();

        System.out.println(end - start);

        final int cpuCoreNum = 4;

        ExecutorService service = Executors.newFixedThreadPool(cpuCoreNum);

        MyTask t1 = new MyTask(1, 80000); //1-5 5-10 10-15 15-20

        MyTask t2 = new MyTask(80001, 130000);

        MyTask t3 = new MyTask(130001, 170000);

        MyTask t4 = new MyTask(170001, 200000);

        Future<List<Integer>> f1 = service.submit(t1);

        Future<List<Integer>> f2 = service.submit(t2);

        Future<List<Integer>> f3 = service.submit(t3);

        Future<List<Integer>> f4 = service.submit(t4);

        start = System.currentTimeMillis();

        f1.get();

        f2.get();

        f3.get();

        f4.get();

        end = System.currentTimeMillis();

        System.out.println(end - start);

    }

    static class MyTask implements Callable<List<Integer>> {

        int startPos, endPos;

        MyTask(int s, int e) {

            this.startPos = s;

            this.endPos = e;

        }

        @Override

        public List<Integer> call() throws Exception {

            List<Integer> r = getPrime(startPos, endPos);

            return r;

        }

    }

    //判断是否是质数

    static boolean isPrime(int num) {

        for(int i=2; i<=num/2; i++) {

            if(num % i == 0) return false;

        }

        return true;

    }

    static List<Integer> getPrime(int start, int end) {

        List<Integer> results = new ArrayList<>();

        for(int i=start; i<=end; i++) {

            if(isPrime(i)) results.add(i);

        }

        return results;

    }

}

console：

2280

818

第二种方式使用了一个线程池，一般线程池有多少个线程，数量多少合适是需要调整的，大多数情况下cpu有几个核至少就应该起多少个线程，可以多起一个但不能少于cpu核数，将20万分成了4段；

这里为什么不将20万平均分呢？

五、CachedThreadPool

package yxxy.c_026;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.TimeUnit;

public class T08_CachedPool {

    public static void main(String[] args) throws InterruptedException {

        ExecutorService service = Executors.newCachedThreadPool();

        System.out.println(service);

        for (int i = 0; i < 2; i++) {

            service.execute(() -> {

                try {

                    TimeUnit.MILLISECONDS.sleep(500);

                } catch (InterruptedException e) {

                    e.printStackTrace();

                }

                System.out.println(Thread.currentThread().getName());

            });

        }

        System.out.println(service);

        TimeUnit.SECONDS.sleep(80); //cachedthreadPool里面的线程空闲状态默认60s后销毁，这里保险起见

        System.out.println(service);

    }

}

console：

java.util.concurrent.ThreadPoolExecutor@7852e922[Running, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0]

java.util.concurrent.ThreadPoolExecutor@7852e922[Running, pool size = 2, active threads = 2, queued tasks = 0, completed tasks = 0]

pool-1-thread-2

pool-1-thread-1

java.util.concurrent.ThreadPoolExecutor@7852e922[Running, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 2]

FixedThreadPool为固定个数的线程池；

CachedThreadPool：刚开始一个线程都没有，来一个任务就起一个线程，假设起了两个线程A，B，如果来了第三个任务，这时候恰好线程B任务执行完了，线程池里面有空闲的，这时候直接让线程池里空闲的线程B来执行；最多起多少个线程？你的系统能支撑多少个为止；默认的情况下，只要一个线程空闲的状态超过60s，这个线程就自动的销毁了，alivetime=60s；这个值也可以自己指定。

六、SingleThreadExecutor

package yxxy.c_026;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

public class T09_SingleThreadPool {

    public static void main(String[] args) {

        ExecutorService service = Executors.newSingleThreadExecutor();

        for(int i=0; i<5; i++) {

            final int j = i;

            service.execute(()->{

                System.out.println(j + " " + Thread.currentThread().getName());

            });

        }

    }

}

console：

0 pool-1-thread-1

1 pool-1-thread-1

2 pool-1-thread-1

3 pool-1-thread-1

4 pool-1-thread-1

SingleThreadExecutor：线程池里就1个线程；扔5个任务，也永远只有1个线程执行；

它能保证任务前后一定是顺序执行，先扔的任务一定先执行完；只有等第一个任务执行完才执行第二个任务。

七、ScheduledThreadPool

package yxxy.c_026;

import java.util.Random;

import java.util.concurrent.Executors;

import java.util.concurrent.ScheduledExecutorService;

import java.util.concurrent.TimeUnit;

public class T10_ScheduledPool {

    public static void main(String[] args) {

        ScheduledExecutorService service = Executors.newScheduledThreadPool(4);

        service.scheduleAtFixedRate(()->{

            try {

                TimeUnit.MILLISECONDS.sleep(new Random().nextInt(1000));

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            System.out.println(Thread.currentThread().getName());

        }, 0, 500, TimeUnit.MILLISECONDS);

    }

}

ScheduledThreadPool：执行定时的任务，定时器线程池，一般可以用来替代timer，而且它里面的线程是可以复用的，第一个线程执行完了之后，任务来了如果第一个线程是空闲的，还可以拿第一个线程来执行。而Timer每次都是new一个新的线程。

scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit),第1个参数是任务，第1个任务马上执行，每隔500毫秒这个任务重复执行。

八、WorkStealingPool

/**

 *

 */

package yxxy.c_026;

import java.io.IOException;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.TimeUnit;

public class T11_WorkStealingPool {

    public static void main(String[] args) throws IOException {

        ExecutorService service = Executors.newWorkStealingPool();

        int count = Runtime.getRuntime().availableProcessors();    //看cpu多少核的;如果是4核，默认就帮你起4个线程

        System.out.println(count);    

        service.execute(new R(1000));

        for(int i=0; i<count; i++){

            service.execute(new R(2000));

        }

        //由于产生的是精灵线程（守护线程、后台线程），主线程不阻塞的话，看不到输出

        System.in.read();

    }

    static class R implements Runnable {

        int time;

        R(int t) {

            this.time = t;

        }

        @Override

        public void run() {

            try {

                TimeUnit.MILLISECONDS.sleep(time);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            System.out.println(time  + " " + Thread.currentThread().getName());

        }

    }

}

console：

8

1000 ForkJoinPool-1-worker-1

2000 ForkJoinPool-1-worker-2

2000 ForkJoinPool-1-worker-0

2000 ForkJoinPool-1-worker-5

2000 ForkJoinPool-1-worker-3

2000 ForkJoinPool-1-worker-6

2000 ForkJoinPool-1-worker-7

2000 ForkJoinPool-1-worker-4

2000 ForkJoinPool-1-worker-1

WorkStealingPool：工作窃取，假设有3个线程A、B、C在运行，workStealing可以简单这么认为，每个线程都维护自己的一个队列，线程A的队列里头积累了5个任务，线程B的队列里1个任务，C的队列里2个任务；那么当线程B执行完任务之后，他会去别的线程池所维护的队列里面把任务偷过来继续执行，主动的找活干。

本质上是使用ForkJoinPool来实现的：

public static ExecutorService newWorkStealingPool() {

        return new ForkJoinPool

            (Runtime.getRuntime().availableProcessors(),

             ForkJoinPool.defaultForkJoinWorkerThreadFactory,

             null, true);

    }

例子解释：cpu多少核默认的起多少个线程，（这里是8），前面几个任务都扔给1-8个线程了，第9个任务来的时候在那里等着了，谁会去执行它呢？先执行完任务的这个线程会去执行。第1个线程只睡1s钟，首先执行完，所以第9个任务一定是第一个线程1去运行它，他会主动的把任务拿过去运行。

workStealing的线程是精灵线程，daemon线程，特点就是主线程main方法一旦结束了，它后台可能还在运行，但是你是看不到它任务输出的；这里Syetem.in.read()让主函数阻塞才能看到输出。debug的时候能看到Daemon Thread[ForkJoinPool-1-worker-1]。为什么用精灵线程？它是在后台不断的运行的，只要虚拟机不退出，这个线程就不会退出，你有任务来了之后，它永远会主动去拿。

workStealing用于什么场景：就说任务分配的不是很均匀，有的线程维护的任务队列比较长，有些线程执行完任务就结束了不太合适，所以他执行完了之后可以去别的线程维护的队列里去偷任务，这样效率更高。

九、ForkJoinPool　　　　

ForkJoinPool： forkjoin的意思就是如果有一个难以完成的大任务，需要计算量特别大，时间特别长，可以把大任务切分成一个个小任务，如果小任务还是太大，它还可以继续分，至于分成多少你可以自己指定，... 分完之后，把结果进行合并，最后合并到一起join一起，产生一个总的结果。而里面任务的切分你可以自己指定，线程的启动根据你任务切分的规则，由ForkJoinPool这个线程池自己来维护。

例子1：

package yxxy.c_026;

import java.io.IOException;

import java.util.Arrays;

import java.util.Random;

import java.util.concurrent.ForkJoinPool;

import java.util.concurrent.RecursiveAction;

import java.util.concurrent.RecursiveTask;

public class T12_ForkJoinPool {

    static int[] nums = new int[1000000];

    static final int MAX_NUM = 50000;

    static Random r = new Random();

    static {

        for(int i=0; i<nums.length; i++) {

            nums[i] = r.nextInt(100);

        }

        System.out.println(Arrays.stream(nums).sum()); //stream api

    }

    static class AddTask extends RecursiveAction { 

        int start, end;

        AddTask(int s, int e) {

            start = s;

            end = e;

        }

        @Override

        protected void compute() {

            if(end-start <= MAX_NUM) {

                long sum = 0L;

                for(int i=start; i<end; i++) sum += nums[i];

                System.out.println("from:" + start + " to:" + end + " = " + sum);

            } else {

                int middle = start + (end-start)/2;

                AddTask subTask1 = new AddTask(start, middle);

                AddTask subTask2 = new AddTask(middle, end);

                subTask1.fork();

                subTask2.fork();

            }

        }

    }

    public static void main(String[] args) throws IOException {

        ForkJoinPool fjp = new ForkJoinPool();

        AddTask task = new AddTask(0, nums.length);

        fjp.execute(task);

        System.in.read();

    }

}

console：

49494882

from:906250 to:937500 = 1545274

from:968750 to:1000000 = 1537201

from:593750 to:625000 = 1548289

from:718750 to:750000 = 1546396

from:468750 to:500000 = 1550373

from:843750 to:875000 = 1543421

from:218750 to:250000 = 1549856

from:93750 to:125000 = 1548384

from:562500 to:593750 = 1541814

from:812500 to:843750 = 1547885

from:187500 to:218750 = 1546831

from:687500 to:718750 = 1554064

from:437500 to:468750 = 1547434

from:937500 to:968750 = 1547676

from:875000 to:906250 = 1551839

from:62500 to:93750 = 1548576

from:531250 to:562500 = 1550943

from:656250 to:687500 = 1544991

from:156250 to:187500 = 1548367

from:406250 to:437500 = 1539881

from:125000 to:156250 = 1548128

from:500000 to:531250 = 1545229

from:781250 to:812500 = 1544296

from:625000 to:656250 = 1545283

from:375000 to:406250 = 1553931

from:31250 to:62500 = 1544024

from:750000 to:781250 = 1543573

from:343750 to:375000 = 1546407

from:0 to:31250 = 1539743

from:281250 to:312500 = 1549470

from:312500 to:343750 = 1552190

from:250000 to:281250 = 1543113

例子解释：

对数组中100万个数求和计算，第一种方式是普通的将所有数加在一起（for循环）；

第二种方式使用ForkJoinPool计算，分而治之，它里面执行的任务必须是ForkJoinTask，这个任务可以自动进行切分，一般用的时候从RecursiveAction或RecursiveTask继承，RecursiveTask递归任务，因为它切分任务还可以在切分。RecursiveAction没有返回值，RecursiveTask有返回值。

例子2：

package yxxy.c_026;

import java.io.IOException;

import java.util.Arrays;

import java.util.Random;

import java.util.concurrent.ForkJoinPool;

import java.util.concurrent.RecursiveAction;

import java.util.concurrent.RecursiveTask;

public class T12_ForkJoinPool {

    static int[] nums = new int[1000000];

    static final int MAX_NUM = 50000;

    static Random r = new Random();

    static {

        for(int i=0; i<nums.length; i++) {

            nums[i] = r.nextInt(100);

        }

        System.out.println(Arrays.stream(nums).sum()); //stream api

    }

    static class AddTask extends RecursiveTask<Long> {

        int start, end;

        AddTask(int s, int e) {

            start = s;

            end = e;

        }

        @Override

        protected Long compute() {

            if(end-start <= MAX_NUM) {

                long sum = 0L;

                for(int i=start; i<end; i++) sum += nums[i];

                return sum;

            } 

            int middle = start + (end-start)/2;

            AddTask subTask1 = new AddTask(start, middle);

            AddTask subTask2 = new AddTask(middle, end);

            subTask1.fork();

            subTask2.fork();

            return subTask1.join() + subTask2.join();

        }

    }

    public static void main(String[] args) throws IOException {

        ForkJoinPool fjp = new ForkJoinPool();

        AddTask task = new AddTask(0, nums.length);

        fjp.execute(task);

        long result = task.join();

        System.out.println(result);

    }

}

console：

49498457

49498457

和例子1差不多，唯一的区别是有返回值了，RecursiveTask<V>中的V泛型就是返回值类型。

long result = task.join()，因为join本身就是阻塞的，只有等所有的都执行完了，最后才得出总的执行结果。所以不需要System.in.read了；

十、自定义线程池 ThreadPoolExecutor

ThreadPoolExecutor：大多数的线程池的实现背后调用的都是ThreadPoolExecutor（前面6种就ForkJoinPool不是），它是线程池通用的一个类，可以自定义线程池；

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)

corePoolSize：核心的线程池里的线程数，自己指定。

maximumPoolSize：最多这个线程池里装多少个线程；

keepAliveTime：线程呆多久没有任务传给它就会消失；

unit：和上面统一指定的；

blockingQueue：真正的装任务的容器，往往都是用blockingQueue；阻塞式的；任务来了就扔进去，什么时候用到了都可以取。

例如，fixedThreadPool的实现是：

public static ExecutorService newFixedThreadPool(int nThreads) {

        return new ThreadPoolExecutor(nThreads, nThreads,

                                      0L, TimeUnit.MILLISECONDS,

                                      new LinkedBlockingQueue<Runnable>());

    }

十一、parallel stream　　

package yxxy.c_026;

import java.util.ArrayList;

import java.util.List;

import java.util.Random;

public class T14_ParallelStreamAPI {

    public static void main(String[] args) {

        List<Integer> nums = new ArrayList<>();

        Random r = new Random();

        for(int i=0; i<10000; i++) nums.add(1000000 + r.nextInt(1000000));

        //System.out.println(nums);

        long start = System.currentTimeMillis();

        nums.forEach(v->isPrime(v));

        long end = System.currentTimeMillis();

        System.out.println(end - start);

        //使用parallel stream api

        start = System.currentTimeMillis();

        nums.parallelStream().forEach(T14_ParallelStreamAPI::isPrime);

        end = System.currentTimeMillis();

        System.out.println(end - start);

    }

    static boolean isPrime(int num) {

        for(int i=2; i<=num/2; i++) {

            if(num % i == 0) return false;

        }

        return true;

    }

}

console：

1526

337

paralleStream(): 运用多线程，如果把这1万个数看成是数据流，我们用多线程去访问里面的数，共同来做计算，默认使用多线程。

---------------

码农公寓

相关文章