阻塞队列和线程池

1. 阻塞队列

1）支持阻塞的插入方法：当队列满时，队列会阻塞插入元素的线程，直到队列不满。
2）支持阻塞的移除方法：在队列为空时，获取元素的线程会等待队列变为非空。
    在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序整体处理数据的速度。
    生产者和消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通信，而是通过阻塞队列来进行通信，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。
    阻塞队列常用于生产者和消费者的场景，生产者是向队列里添加元素的线程，消费者是从队列里取元素的线程。阻塞队列就是生产者用来存放元素、消费者用来获取元素的容器。

    抛出异常： 当队列满时，如果再往队列里插入元素，会抛出IllegalStateException（“Queuefull”）异常。当队列空时，从队列里获取元素会抛出NoSuchElementException异常。
    返回特殊值： 当往队列插入元素时，会返回元素是否插入成功，成功返回true。如果是移除方法，则是从队列里取出一个元素，如果没有则返回null。
    一直阻塞： 当阻塞队列满时，如果生产者线程往队列里put元素，队列会一直阻塞生产者线程，直到队列可用或者响应中断退出。当队列空时，如果消费者线程从队列里take元素，队列会阻塞住消费者线程，直到队列不为空。
    超时退出： 当阻塞队列满时，如果生产者线程往队列里插入元素，队列会阻塞生产者线程一段时间，如果超过了指定的时间，生产者线程就会退出。

1）常用阻塞队列

• ArrayBlockingQueue：一个由数组结构组成的有界阻塞队列。
• LinkedBlockingQueue：一个由链表结构组成的有界阻塞队列。
• PriorityBlockingQueue：一个支持优先级排序的无界阻塞队列。
• DelayQueue：一个使用优先级队列实现的无界阻塞队列。 SynchronousQueue：一个不存储元素的阻塞队列。
• LinkedTransferQueue：一个由链表结构组成的无界阻塞队列。
• LinkedBlockingDeque：一个由链表结构组成的双向阻塞队列。

以上的阻塞队列都实现了BlockingQueue接口，也都是线程安全的。

有界无界

    有限队列就是长度有限，满了以后生产者会阻塞，无界队列就是里面能放无数的东西而不会因为队列长度限制被阻塞，当然空间限制来源于系统资源的限制，如果处理不及时，导致队列越来越大越来越大，超出一定的限制致使内存超限，操作系统或者JVM帮你解决烦恼，直接 OOM kill 了。
    无界也会阻塞，为何？因为阻塞不仅仅体现在生产者放入元素时会阻塞，消费者拿取元素时，如果没有元素，同样也会阻塞。

ArrayBlockingQueue

    是一个用数组实现的有界阻塞队列。此队列按照先进先出（FIFO）的原则对元素进行排序。默认情况下不保证线程公平的访问队列，所谓公平访问队列是指阻塞的线程，可以按照阻塞的先后顺序访问队列，即先阻塞线程先访问队列。非公平性是对先等待的线程是非公平的，当队列可用时，阻塞的线程都可以争夺访问队列的资格，有可能先阻塞的线程最后才访问队列。初始化时有参数可以设置

LinkedBlockingQueue

    是一个用链表实现的有界阻塞队列。此队列的默认和最大长度为Integer.MAX_VALUE。此队列按照先进先出的原则对元素进行排序。

Array实现和Linked实现的区别:

1. 队列中锁的实现不同
   ArrayBlockingQueue实现的队列中的锁是没有分离的，即生产和消费用的是同一个锁；
   LinkedBlockingQueue实现的队列中的锁是分离的，即生产用的是putLock，消费是takeLock
2. 在生产或消费时操作不同
   ArrayBlockingQueue实现的队列中在生产和消费的时候，是直接将枚举对象插入或移除的；
   LinkedBlockingQueue实现的队列中在生产和消费的时候，需要把枚举对象转换为Node进行插入或移除，会影响性能
3. 队列大小初始化方式不同
   ArrayBlockingQueue实现的队列中必须指定队列的大小；
   LinkedBlockingQueue实现的队列中可以不指定队列的大小，但是默认是Integer.MAX_VALUE

PriorityBlockingQueue

    PriorityBlockingQueue是一个支持优先级的无界阻塞队列。默认情况下元素采取自然顺序升序排列。也可以自定义类实现compareTo()方法来指定元素排序规则，或者初始化PriorityBlockingQueue时，指定构造参数Comparator来对元素进行排序。需要注意的是不能保证同优先级元素的顺序。

DelayQueue

    是一个支持延时获取元素的无界阻塞队列。队列使用PriorityQueue来实现。队列中的元素必须实现Delayed接口，在创建元素时可以指定多久才能从队列中获取当前元素。只有在延迟期满时才能从队列中提取元素。
    DelayQueue非常有用，可以将DelayQueue运用在以下应用场景。
    缓存系统的设计：可以用DelayQueue保存缓存元素的有效期，使用一个线程循环查询DelayQueue，一旦能从DelayQueue中获取元素时，表示缓存有效期到了。

SynchronousQueue

    是一个不存储元素的阻塞队列。每一个put操作必须等待一个take操作，否则不能继续添加元素。SynchronousQueue可以看成是一个传球手，负责把生产者线程处理的数据直接传递给消费者线程。队列本身并不存储任何元素，非常适合传递性场景。SynchronousQueue的吞吐量高于LinkedBlockingQueue和ArrayBlockingQueue。

LinkedTransferQueue

    多了tryTransfer和transfer方法，
（1）transfer方法
    如果当前有消费者正在等待接收元素（消费者使用take()方法或带时间限制的poll()方法时），transfer方法可以把生产者传入的元素立刻transfer（传输）给消费者。如果没有消费者在等待接收元素，transfer方法会将元素存放在队列的tail节点，并等到该元素被消费者消费了才返回。
（2）tryTransfer方法
    tryTransfer方法是用来试探生产者传入的元素是否能直接传给消费者。如果没有消费者等待接收元素，则返回false。和transfer方法的区别是tryTransfer方法无论消费者是否接收，方法立即返回，而transfer方法是必须等到消费者消费了才返回。

LinkedBlockingDeque

    LinkedBlockingDeque是一个由链表结构组成的双向阻塞队列。所谓双向队列指的是可以从队列的两端插入和移出元素。双向队列因为多了一个操作队列的入口，在多线程同时入队时，也就减少了一半的竞争。
    多了addFirst、addLast、offerFirst、offerLast、peekFirst和peekLast等方法，以First单词结尾的方法，表示插入、获取（peek）或移除双端队列的第一个元素。以Last单词结尾的方法，表示插入、获取或移除双端队列的最后一个元素。另外，插入方法add等同于addLast，移除方法remove等效于removeFirst。但是take方法却等同于takeFirst，不知道是不是JDK的bug，使用时还是用带有First和Last后缀的方法更清楚。在初始化LinkedBlockingDeque时可以设置容量防止其过度膨胀。另外，双向阻塞队列可以运用在“工作窃取”模式中。

2）延时队列测试

// 订单实体类
public class Order {
    private final String orderNo; // 订单编号
    private final double orderMoney; // 订单金额

    public Order(String orderNo, double orderMoney){
        this.orderNo = orderNo;
        this.orderMoney = orderMoney;
    }

    public String getOrderNo(){
        return orderNo;
    }

    public double getOrderMoney(){
        return orderMoney;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

// 存放的队列的元素
public class ItemVo<T> implements Delayed {
    private long activeTime; // 到期时间,但传入的数值代表过期的时长，传入单位毫秒
    private T data; // 业务数据 泛型

    public ItemVo(long expirationTime, T data){
        this.activeTime = expirationTime*1000 + System.currentTimeMillis();
        this.data = data;
    }

    public long getActiveTime(){
        return activeTime;
    }

    public T getData(){
        return data;
    }


    /***
     * 这个方法返回到激活日期的剩余时间，时间单位由单位参数指定
     */
    @Override
    public long getDelay(TimeUnit unit) {
        long d = unit.convert(this.activeTime - System.currentTimeMillis(), unit);
        return d;
    }

    /**
     * Delayed接口继承了Comparable接口，按剩余时间排序，实际计算考虑精度为纳秒数
     */
    @Override
    public int compareTo(Delayed o) {
        long d = getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS);
        if (d==0){
            return 0;
        }else {
            if (d<0){
                return -1;
            }else {
                return 1;
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

// 将订单推入队列
public class PutOrder implements Runnable {
    private DelayQueue<ItemVo<Order>> queue;

    public PutOrder(DelayQueue<ItemVo<Order>> queue){
        this.queue = queue;
    }

    @Override
    public void run() {
        // 5s后到期
        Order order1 = new Order("10001", 998);
        ItemVo<Order> itemTb = new ItemVo<>(5, order1);
        queue.offer(itemTb);
        System.out.println("订单5秒后超时："+order1.getOrderNo()+";" + order1.getOrderMoney());

        //8秒后到期
        Order order2 = new Order("10002",1998);
        ItemVo<Order> itemJd = new ItemVo<Order>(8, order2);
        queue.offer(itemJd);
        System.out.println("订单8秒后超时："+order2.getOrderNo()+";" + order2.getOrderMoney());
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

// 取出到期的订单
public class FetchOrder implements  Runnable {
    private DelayQueue<ItemVo<Order>> queue;
    public FetchOrder(DelayQueue<ItemVo<Order>> queue){
        this.queue = queue;
    }

    @Override
    public void run() {
        while (true){
            try {
                ItemVo<Order> item = queue.take();
                Order order = item.getData();
                System.out.println("Get From Queue:"+"data=" +order.getOrderNo()+";"+order.getOrderMoney());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

// 延时队列测试
public class TestQueue {
    public static void main(String[] args) throws InterruptedException {
        DelayQueue<ItemVo<Order>> queue = new DelayQueue<>();

        new Thread(new PutOrder(queue)).start();
        new Thread(new FetchOrder(queue)).start();

        for (int i = 1; i <= 15; i++) {
            Thread.sleep(500);
            System.out.println(i*500);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

2. 线程池

    Java中的线程池是运用场景最多的并发框架，几乎所有需要异步或并发执行任务的程序都可以使用线程池。
合理地使用线程池好处：
    1) 降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
    2) 提高响应速度。当任务到达时，任务可以不需要等到线程创建就能立即执行。假设一个服务器完成一项任务所需时间为：T1 创建线程时间，T2 在线程中执行任务的时间，T3 销毁线程时间。 如果：T1 + T3 远大于 T2，则可以采用线程池，以提高服务器性能。线程池技术正是关注如何缩短或调整T1,T3时间的技术，从而提高服务器程序性能的。它把T1，T3分别安排在服务器程序的启动和结束的时间段或者一些空闲的时间段，这样在服务器程序处理客户请求时，不会有T1，T3的开销了。
    3) 提高线程的可管理性。线程是稀缺资源，如果无限制地创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一分配、调优和监控。

1) ThreadPoolExecutor 的类关系

    Executor是一个接口，它是Executor框架的基础，它将任务的提交与任务的执行分离开来。
    ExecutorService接口继承了Executor，在其上做了一些shutdown()、submit()的扩展，可以说是真正的线程池接口；
    AbstractExecutorService抽象类实现了ExecutorService接口中的大部分方法；
    ThreadPoolExecutor是线程池的核心实现类，用来执行被提交的任务。
    ScheduledExecutorService接口继承了ExecutorService接口，提供了带"周期执行"功能ExecutorService；
    ScheduledThreadPoolExecutor是一个实现类，可以在给定的延迟后运行命令，或者定期执行命令。ScheduledThreadPoolExecutor比Timer更灵活，功能更强大。

2) 线程池的创建各个参数含义

public ThreadPoolExecutor(
	int corePoolSize,
	int maximumPoolSize,
	long keepAliveTime,
	TimeUnit unit,
	BlockingQueue<Runnable> workQueue,
	ThreadFactory threadFactory,
	RejectedExecutionHandler handler
)
1
2
3
4
5
6
7
8
9

corePoolSize

    线程池中的核心线程数，当提交一个任务时，线程池创建一个新线程执行任务，直到当前线程数等于corePoolSize；
    如果当前线程数为corePoolSize，继续提交的任务被保存到阻塞队列中，等待被执行；
    如果执行了线程池的prestartAllCoreThreads()方法，线程池会提前创建并启动所有核心线程。

maximumPoolSize

    线程池中允许的最大线程数。如果当前阻塞队列满了，且继续提交任务，则创建新的线程执行任务，前提是当前线程数小于maximumPoolSize

keepAliveTime

    线程空闲时的存活时间，即当线程没有任务执行时，继续存活的时间。默认情况下，该参数只在线程数大于corePoolSize时才有用

TimeUnit

    keepAliveTime的时间单位

workQueue

    workQueue必须是BlockingQueue阻塞队列。当线程池中的线程数超过它的corePoolSize的时候，线程会进入阻塞队列进行阻塞等待。通过workQueue，线程池实现了阻塞功能。
    一般来说，我们应该尽量使用有界队列，因为使用无界队列作为工作队列会对线程池带来如下影响。
        1）当线程池中的线程数达到corePoolSize后，新任务将在无界队列中等待，因此线程池中的线程数不会超过corePoolSize。
        2）由于1，使用无界队列时maximumPoolSize将是一个无效参数。
        3）由于1和2，使用无界队列时keepAliveTime将是一个无效参数。
        4）更重要的，使用无界queue可能会耗尽系统资源，有界队列则有助于防止资源耗尽，同时即使使用有界队列，也要尽量控制队列的大小在一个合适的范围。

threadFactory

    创建线程的工厂，通过自定义的线程工厂可以给每个新建的线程设置一个具有识别度的线程名，当然还可以更加自由的对线程做更多的设置，比如设置所有的线程为守护线程。
    Executors静态工厂里默认的threadFactory，线程的命名规则是“pool-数字-thread-数字”。

RejectedExecutionHandler

    线程池的饱和策略，当阻塞队列满了，且没有空闲的工作线程，如果继续提交任务，必须采取一种策略处理该任务，线程池提供了4种策略：
    （1）AbortPolicy：直接抛出异常，默认策略；
    （2）CallerRunsPolicy：用调用者所在的线程来执行任务；
    （3）DiscardOldestPolicy：丢弃阻塞队列中靠最前的任务，并执行当前任务；
    （4）DiscardPolicy：直接丢弃任务；
    当然也可以根据应用场景实现RejectedExecutionHandler接口，自定义饱和策略，如记录日志或持久化存储不能处理的任务。

3) 线程池的工作机制

    1）如果当前运行的线程少于corePoolSize，则创建新线程来执行任务（注意，执行这一步骤需要获取全局锁）。
    2）如果运行的线程等于或多于corePoolSize，则将任务加入BlockingQueue。
    3）如果无法将任务加入BlockingQueue（队列已满），则创建新的线程来处理任务。
    4）如果创建新线程将使当前运行的线程超出maximumPoolSize，任务将被拒绝，并调用RejectedExecutionHandler.rejectedExecution()方法。

4) 提交任务

    execute() 方法用于提交不需要返回值的任务，所以无法判断任务是否被线程池执行成功。
    submit() 方法用于提交需要返回值的任务。线程池会返回一个future类型的对象，通过这个future对象可以判断任务是否执行成功，并且可以通过future的get()方法来获取返回值，get()方法会阻塞当前线程直到任务完成，而使用get（long timeout，TimeUnit unit）方法则会阻塞当前线程一段时间后立即返回，这时候有可能任务没有执行完。

5) 关闭线程池

    可以通过调用线程池的shutdown或shutdownNow方法来关闭线程池。它们的原理是遍历线程池中的工作线程，然后逐个调用线程的interrupt方法来中断线程，所以无法响应中断的任务可能永远无法终止。但是它们存在一定的区别，shutdownNow首先将线程池的状态设置成STOP，然后尝试停止所有的正在执行或暂停任务的线程，并返回等待执行任务的列表，而shutdown只是将线程池的状态设置成SHUTDOWN状态，然后中断所有没有正在执行任务的线程。
    只要调用了这两个关闭方法中的任意一个，isShutdown方法就会返回true。当所有的任务都已关闭后，才表示线程池关闭成功，这时调用isTerminaed方法会返回true。至于应该调用哪一种方法来关闭线程池，应该由提交到线程池的任务特性决定，通常调用shutdown方法来关闭线程池，如果任务不一定要执行完，则可以调用shutdownNow方法。

6) 合理地配置线程池

    要想合理地配置线程池，就必须首先分析任务特性，可以从以下几个角度来分析。

• 任务的性质：CPU密集型任务、IO密集型任务 和 混合型任务。
• 任务的优先级：高、中和低。
• 任务的执行时间：长、中和短。
• 任务的依赖性：是否依赖其他系统资源，如数据库连接。

    性质不同的任务可以用不同规模的线程池分开处理。
    CPU密集型任务应配置尽可能小的线程，如配置Ncpu+1个线程的线程池。
    由于IO密集型任务线程并不是一直在执行任务，则应配置尽可能多的线程，如2*Ncpu。
    混合型的任务，如果可以拆分，将其拆分成一个CPU密集型任务和一个IO密集型任务，只要这两个任务执行的时间相差不是太大，那么分解后执行的吞吐量将高于串行执行的吞吐量。如果这两个任务执行时间相差太大，则没必要进行分解。
    可以通过Runtime.getRuntime().availableProcessors()方法获得当前设备的CPU个数。
    优先级不同的任务可以使用优先级队列PriorityBlockingQueue来处理。它可以让优先级高的任务先执行。
    执行时间不同的任务可以交给不同规模的线程池来处理，或者可以使用优先级队列，让执行时间短的任务先执行。
    建议使用有界队列。有界队列能增加系统的稳定性和预警能力，可以根据需要设大一点儿，比如几千。
    如果当时我们设置成无界队列，那么线程池的队列就会越来越多，有可能会撑满内存，导致整个系统不可用，而不只是后台任务出现问题。

public class ThreadPoolTest {
    static class Worker implements Runnable {
        private String taskName;
        private Random r = new Random();

        public Worker(String taskName){
            this.taskName = taskName;
        }

        public String getName(){
            return taskName;
        }

        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName()+" process the task : "+taskName);
            try {
                Thread.sleep(r.nextInt(100)*5);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    static class CallWorker implements Callable<String>{
        private String taskName;
        private Random r = new Random();

        public CallWorker(String taskName){
            this.taskName = taskName;
        }

        public String getName(){
            return taskName;
        }

        @Override
        public String call() throws Exception {
            System.out.println(Thread.currentThread().getName()+" process the task : "+taskName);
            return Thread.currentThread().getName() + ":"+r.nextInt(100)*5;
        }
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService pool = new ThreadPoolExecutor(
                2,
                Runtime.getRuntime().availableProcessors(),
                3,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(10),
                new ThreadPoolExecutor.DiscardOldestPolicy());
//        ExecutorService pool = Executors.newCachedThreadPool();

        for (int i = 0; i < 6; i++) {
            Worker worker = new Worker("worker_"+i);
            pool.execute(worker);
        }
        for (int i = 0; i < 6; i++) {
            CallWorker callWorker = new CallWorker("callWorker_"+i);
            Future<String> result = pool.submit(callWorker);
            System.out.println(result.get());
        }
        pool.shutdown();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

参考享学！