zoukankan      html  css  js  c++  java

  • java Concurrent包学习笔记（一）：ExecutorService

    一、介绍

    ExecutorService是java.util.concurrent包中的一个线程池实现接口。其有两个实现类：

      1）ThreadPoolExecutor：普通线程池通过配置线程池大小，能有效管理线程的调度，在执行大量异步线程时提高程序的性能。

        /**
     * Creates a new {@code ThreadPoolExecutor} with the given initial
     * parameters.
     *
     * @param corePoolSize the number of threads to keep in the pool, even
     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @param maximumPoolSize the maximum number of threads to allow in the
     *        pool
     * @param keepAliveTime when the number of threads is greater than
     *        the core, this is the maximum time that excess idle threads
     *        will wait for new tasks before terminating.
     * @param unit the time unit for the {@code keepAliveTime} argument
     * @param workQueue the queue to use for holding tasks before they are
     *        executed.  This queue will hold only the {@code Runnable}
     *        tasks submitted by the {@code execute} method.
     * @param threadFactory the factory to use when the executor
     *        creates a new thread
     * @param handler the handler to use when execution is blocked
     *        because the thread bounds and queue capacities are reached
     * @throws IllegalArgumentException if one of the following holds:<br>
     *         {@code corePoolSize < 0}<br>
     *         {@code keepAliveTime < 0}<br>
     *         {@code maximumPoolSize <= 0}<br>
     *         {@code maximumPoolSize < corePoolSize}
     * @throws NullPointerException if {@code workQueue}
     *         or {@code threadFactory} or {@code handler} is null
     */
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }
    • corePoolSize：核心线程数，如果运行的线程少于corePoolSize，则创建新线程来执行新任务，即使线程池中的其他线程是空闲的
    • maximumPoolSize：最大线程数，可允许创建的线程数，corePoolSize和maximumPoolSize设置的边界自动调整池大小： 
    • corePoolSize <运行的线程数< maximumPoolSize:仅当队列满时才创建新线程
    • corePoolSize=运行的线程数= maximumPoolSize：创建固定大小的线程池
    • keepAliveTime：如果线程数多于corePoolSize,则这些多余的线程的空闲时间超过keepAliveTime时将被终止
    • unit：keepAliveTime参数的时间单位
    • workQueue：保存任务的阻塞队列，与线程池的大小有关： 
    • 当运行的线程数少于corePoolSize时，在有新任务时直接创建新线程来执行任务而无需再进队列 
    • 当运行的线程数等于或多于corePoolSize，在有新任务添加时则选加入队列，不直接创建线程 
    • 当队列满时，在有新任务时就创建新线程
    • threadFactory：使用ThreadFactory创建新线程，默认使用defaultThreadFactory创建线程
    • handle：定义处理被拒绝任务的策略，默认使用ThreadPoolExecutor.AbortPolicy,任务被拒绝时将抛出RejectExecutorException

      2）ScheduledThreadPoolExecutor：执行延迟任务和周期性任务。

    二、ExecutorService种类

     1、newSingleThreadExecutor

    由数可知，创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，没有被执行的线程先排在等待队列中，而且先放入线程池的先执行

     示例：

    package executorservice.demo;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author boshen
 * @date 2018/12/20
 */
public class SingleThreadExecutorTest {
    class StudentThread implements Runnable{
        private String name;
        StudentThread(String name){
            this.name = name;
        }
        public void run(){
            System.out.println("学生：" + name + " 开始吃饭");
            try {
                Thread.sleep(3000);
                System.out.println("学生：" + name + " 吃完饭了");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args){
        SingleThreadExecutorTest cb = new SingleThreadExecutorTest();
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        executorService.submit(cb.new StudentThread("张三"));
        executorService.submit(cb.new StudentThread("李四"));
        executorService.shutdown();
    }
}
    学生：张三 开始吃饭
学生：张三 吃完饭了
学生：李四 开始吃饭
学生：李四 吃完饭了

     2、newFixedThreadPool

      

     创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待

     示例：

    package executorservice.demo;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author boshen
 * @date 2018/12/20
 */
public class FixedThreadPoolTest {
    class StudentThread implements Runnable{
        private String name;
        StudentThread(String name){
            this.name = name;
        }
        public void run(){
            System.out.println("学生：" + name + " 开始吃饭");
            try {
                Thread.sleep(2000);
                System.out.println("学生：" + name + " 吃完饭了");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args){
        FixedThreadPoolTest cb = new FixedThreadPoolTest();
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        executorService.submit(cb.new StudentThread("张三"));
        executorService.submit(cb.new StudentThread("李四"));
        executorService.submit(cb.new StudentThread("王五"));
        executorService.submit(cb.new StudentThread("马六"));
        executorService.shutdown();
    }
}
    学生：李四 开始吃饭
学生：张三 开始吃饭
学生：李四 吃完饭了
学生：张三 吃完饭了
学生：王五 开始吃饭
学生：马六 开始吃饭
学生：马六 吃完饭了
学生：王五 吃完饭了

     3、newCachedThreadPool

    创建可缓存的线程池，如果线程池中的线程在60秒未被使用就将被移除，在执行新的任务时，当线程池中有之前创建的可用线程就重用可用线程，否则就新建一条线程

     示例：

    package executorservice.demo;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author boshen
 * @date 2018/12/20
 */
public class CachedThreadPoolTest {
    class StudentThread1 implements Runnable{
        private String name;
        StudentThread1(String name){
            this.name = name;
        }
        public void run(){
            System.out.println("学生：" + name + " 开始吃饭,线程名为："+Thread.currentThread().getName());
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    class StudentThread2 implements Runnable{
        private String name;
        StudentThread2(String name){
            this.name = name;
        }
        public void run(){
            System.out.println("学生：" + name + " 开始吃饭,线程名为："+Thread.currentThread().getName());
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args){
        CachedThreadPoolTest cb = new CachedThreadPoolTest();
        ExecutorService executorService = Executors.newCachedThreadPool();
        executorService.submit(cb.new StudentThread1("张三"));
        executorService.submit(cb.new StudentThread1("李四"));
        executorService.submit(cb.new StudentThread2("王五"));
        executorService.submit(cb.new StudentThread2("马六"));
        try {
            Thread.sleep(4000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        executorService.submit(cb.new StudentThread1("赵七"));
        executorService.submit(cb.new StudentThread1("杨八"));
        executorService.shutdown();
    }
}
    学生：张三 开始吃饭,线程名为：pool-1-thread-1
学生：王五 开始吃饭,线程名为：pool-1-thread-3
学生：马六 开始吃饭,线程名为：pool-1-thread-4
学生：李四 开始吃饭,线程名为：pool-1-thread-2
学生：赵七 开始吃饭,线程名为：pool-1-thread-2
学生：杨八 开始吃饭,线程名为：pool-1-thread-1

    由结果可知：

    张三和李四执行时间为2秒，王五和马六执行时间为10秒，提交了前4个线程之后隔了4秒提交赵七和杨八的线程，这时候张三和李四已经执行完了。

    所以张三的线程pool-1-thread-1继续执行杨八，李四的线程pool-1-thread-2继续执行赵七。并没有多创建出来pool-1-thread-5和pool-1-thread-6

     4、newScheduledThreadPool

    创建一个定长线程池，支持定时及周期性任务执行

    • Executors.newScheduledThreadPool(int corePoolSize)，corePoolSize表示线程容量。
    • schedule(Callable/Runnable command,long initialDelay,TimeUnit unit):第一个参数任务，第二个参数表示执行任务前等待的时间，第三参数表示时间单位。
    • scheduleAtFixedRate(Runnable command,long initialDelay,long period,TimeUnit unit)：第一个参数表示周期线执行的任务，第二个参数表示第一次执行前的延迟时间，第三个参数表示任务启动间隔时间，第四个参数表示时间单位。虽然任务类型是Runnable但该方法有返回值ScheduledFuture。可以通过该对象获取线程信息。
    • scheduleWithFixedDelay(Runnable command,long initialDelay,long period,TimeUnit unit)：与scheduleAtFixedRate方法类似，不同的是第三个参数表示前一次结束的时间和下一次任务启动的间隔时间

     示例

    package executorservice.demo;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @author boshen
 * @date 2018/12/20
 */
public class ScheduledThreadPoolTest {
    class StudentThread implements Runnable{
        private String name;
        StudentThread(String name){
            this.name = name;
        }
        public void run(){
            try {
                System.out.println("学生：" + name + " 开始吃饭,线程名为："+Thread.currentThread().getName());
                Thread.sleep(1000);
            }catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }


    public static void main(String[] args){
        ScheduledThreadPoolTest cb = new ScheduledThreadPoolTest();
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(2);
        //当以下线程提交后要2秒后才执行，只执行一次
        executorService.schedule(cb.new StudentThread("张三"),2000, TimeUnit.MILLISECONDS);
        //当以下线程提交后要2秒后才执行，每3秒执行一次,直到调用了executorService.shutdown();
        executorService.scheduleWithFixedDelay(cb.new StudentThread("李四"),2,3,TimeUnit.SECONDS);
        try {
            Thread.sleep(20000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        executorService.shutdown();
    }
}
    学生：李四 开始吃饭,线程名为：pool-1-thread-2
学生：张三 开始吃饭,线程名为：pool-1-thread-1
学生：李四 开始吃饭,线程名为：pool-1-thread-2
学生：李四 开始吃饭,线程名为：pool-1-thread-2
学生：李四 开始吃饭,线程名为：pool-1-thread-2
学生：李四 开始吃饭,线程名为：pool-1-thread-2

    三、ExecutorService的几个方法区别

     1、execute(Runnable),无法取得返回值

        public static void main(String[] args){
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        executorService.execute(new Runnable() {
            public void run() {
                System.out.println("该异步任务无返回值");
            }
        });

        executorService.shutdown();
    }

    2、submit(Runnable),返回一个 Future 对象。这个 Future 对象可以用来检查 Runnable 是否已经执行完毕,但是也无法取得run方法里面想要返回的值因为run方法为void

        public static void main(String[] args){
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        Future future = executorService.submit(new Runnable() {
            public void run() {
                try {
                    Thread.sleep(10000);
                    System.out.println("该任务执行了10秒");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }
        });
        System.out.println("主线程中获取子线程的执行状态：如果返回null表示执行正确完成");
        try {
            System.out.println(future.get());//线程没有执行完之前，会阻塞在这里
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        executorService.shutdown();
    }
    主线程中获取子线程的执行状态：如果返回null表示执行正确完成
该任务执行了10秒
null

    3、submit(Callable),返回一个 Future 对象。这个 Future 对象可以返回线程中call方法里面return的对象

        public static void main(String[] args){
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        Future future = executorService.submit(new Callable() {
            public Object call() throws Exception {
                Thread.sleep(10000);
                System.out.println("该任务执行了10秒");
                return "call 返回的值";
            }
        });
        System.out.println("主线程中获取子线程的结果：");
        try {
            System.out.println(future.get());//线程没有执行完之前，会阻塞在这里
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        executorService.shutdown();
    }
    主线程中获取子线程的结果：
该任务执行了10秒
call 返回的值

    4、invokeAll(Collection<? extends Callable<T>> tasks),参数是加入线程池的所有Callable，返值是List<Future<T>>，表示返回执行后的一系列Callable的执行结果

    package executorservice.demo;

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;

/**
 * @author Administrator
 * @date 2018/12/27
 */
public class InvokeAllTest {
    class StudentThread implements Callable{
        private String name;
        StudentThread(String name){
            this.name = name;
        }
        public Object call() throws Exception{
            System.out.println("学生：" + name + " 开始吃饭,线程名为："+Thread.currentThread().getName());
            return "result:  学生"+name+"吃完饭了";
        }
    }
    public static void main(String[] args){
        InvokeAllTest invokeAllTest = new InvokeAllTest();
        ExecutorService executorService = Executors.newCachedThreadPool();
        List<Callable<String>> callables = new ArrayList<Callable<String>>();
        callables.add(invokeAllTest.new StudentThread("张三"));
        callables.add(invokeAllTest.new StudentThread("李四"));
        callables.add(invokeAllTest.new StudentThread("王五"));
        callables.add(invokeAllTest.new StudentThread("马六"));
        try {
            List<Future<String>> futures = executorService.invokeAll(callables);
            for(Future<String> future:futures){
                System.out.println(future.get());
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        executorService.shutdown();
    }
}
    学生：张三 开始吃饭,线程名为：pool-1-thread-1
学生：王五 开始吃饭,线程名为：pool-1-thread-3
学生：李四 开始吃饭,线程名为：pool-1-thread-2
学生：马六 开始吃饭,线程名为：pool-1-thread-4
result:  学生张三吃完饭了
result:  学生李四吃完饭了
result:  学生王五吃完饭了
result:  学生马六吃完饭了

    四、ExecutorService 关闭

    ExecutorService 的 shutdown() 方法。并不会立即关闭，但它将不再接受新的任务，而且一旦所有线程都完成了当前任务的时候，ExecutorService 将会关闭。在 shutdown() 被调用之前所有提交给 ExecutorService 的任务都被执行。

    如果你想要立即关闭 ExecutorService，你可以调用 shutdownNow() 方法。这样会立即尝试停止所有执行中的任务，并忽略掉那些已提交但尚未开始处理的任务。无法担保执行任务的正确执行。可能它们被停止了，也可能已经执行结束。
  • 相关阅读:
    （零 ） 天猫精灵接入Home Assistant-总说明
    （二 -2） 天猫精灵接入Home Assistant-自动发现Mqtt设备
    （二 -1） 天猫精灵接入Home Assistant-控制Mqtt设备
    如何在关闭ssh连接的情况下，让程序继续运行？
    hass连接设备
    （一 ） 天猫精灵接入Home Assistant-服务器搭建
    （一） 天猫精灵接入Home Assistant- hass对接天猫精灵
    （3）HomeAssistant 连接MQTT
    （2）HomeAssistant 参数配置
    （1）HomeAssistant 安装开始
  • 原文地址：https://www.cnblogs.com/boshen-hzb/p/10184036.html
Copyright © 2011-2022 走看看