并发编程系列教程（07） - 线程池原理分析（一）-阿里云开发者社区

本文主要讲解并发包相关的内容，示例代码已经上传到Github，有兴趣的同学可以下载来看看：https://github.com/ylw-github/Java-ThreadDemo

1. CountDownLatch（计数器）

CountDownLatch 类位于java.util.concurrent包下，利用它可以实现类似计数器的功能。

比如有一个任务A，它要等待其他4个任务执行完毕之后才能执行，此时就可以利用CountDownLatch来实现这种功能了。CountDownLatch是通过一个计数器来实现的，计数器的初始值为线程的数量。每当一个线程完成了自己的任务后，计数器的值就会减1。当计数器值到达0时，它表示所有的线程已经完成了任务，然后在闭锁上等待的线程就可以恢复执行任务。

import java.util.concurrent.CountDownLatch;
public class CountDownLatchDemo {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(2);
        new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println(Thread.currentThread().getName() + ",子线程开始执行...");
                countDownLatch.countDown();
                System.out.println(Thread.currentThread().getName() + ",子线程结束执行...");
            }
        }).start();
        new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println(Thread.currentThread().getName() + ",子线程开始执行...");
                countDownLatch.countDown();//计数器值每次减去1
                System.out.println(Thread.currentThread().getName() + ",子线程结束执行...");
            }
        }).start();
        countDownLatch.await();// 減去为0,恢复任务继续执行
        System.out.println("两个子线程执行完毕....");
        System.out.println("主线程继续执行.....");
        for (int i = 0; i < 10; i++) {
            System.out.println("main,i:" + i);
        }
    }
}

运行结果：

2. CyclicBarrier（屏障）

CyclicBarrier 初始化时规定一个数目，然后计算调用了CyclicBarrier.await()进入等待的线程数。当线程数达到了这个数目时，所有进入等待状态的线程被唤醒并继续。
CyclicBarrier 就象它名字的意思一样，可看成是个障碍，所有的线程必须到齐后才能一起通过这个障碍。
CyclicBarrier 初始时还可带一个Runnable的参数，此Runnable任务在CyclicBarrier的数目达到后，所有其它线程被唤醒前被执行。

示例代码：

package com.ylw.threadpool;
import java.util.concurrent.CyclicBarrier;
public class CyclicBarrierDemo {
    public static class Writer extends Thread {
        private CyclicBarrier cyclicBarrier;
        public Writer(CyclicBarrier cyclicBarrier){
            this.cyclicBarrier=cyclicBarrier;
        }
        @Override
        public void run() {
            System.out.println("线程" + Thread.currentThread().getName() + ",正在写入数据");
            try {
                Thread.sleep(3000);
            } catch (Exception e) {
                e.printStackTrace();
            }
            System.out.println("线程" + Thread.currentThread().getName() + ",写入数据成功.....");
            try {
                cyclicBarrier.await();
            } catch (Exception e) {
                e.printStackTrace();
            }
            System.out.println("所有线程执行完毕..........");
        }
    }
    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier=new CyclicBarrier(5);
        for (int i = 0; i < 5; i++) {
            Writer writer = new Writer(cyclicBarrier);
            writer.start();
        }
    }
}

运行结果：

3. Semaphore（计数信号量）

Semaphore 是一种基于计数的信号量。它可以设定一个阈值，基于此，多个线程竞争获取许可信号，做自己的申请后归还，超过阈值后，线程申请许可信号将会被阻塞。

Semaphore 可以用来构建一些对象池，资源池之类的，比如数据库连接池，我们也可以创建计数为1的Semaphore，将其作为一种类似互斥锁的机制，这也叫二元信号量，表示两种互斥状态。它的用法如下：

availablePermits函数；//用来获取当前可用的资源数量
wc.acquire(); //申请资源
wc.release();// 释放资源

// 创建一个计数阈值为5的信号量对象  
// 只能5个线程同时访问  
Semaphore semp = new Semaphore(5);  
try {  
   // 申请许可  
   semp.acquire();  
   try {  
       // 业务逻辑  
   } catch (Exception e) {  
   } finally {  
       // 释放许可  
       semp.release();  
   }  
} catch (InterruptedException e) {  
}

案例：

一个厕所只有3个坑位，但是有10个人来上厕所，那怎么办？假设10的人的编号分别为1-10，并且1号先到厕所，10号最后到厕所。那么1-3号来的时候必然有可用坑位，顺利如厕，4号来的时候需要看看前面3人是否有人出来了，如果有人出来，进去，否则等待。同样的道理，4-10号也需要等待正在上厕所的人出来后才能进去，并且谁先进去这得看等待的人是否有素质，是否能遵守先来先上的规则。

import java.util.Random;
import java.util.concurrent.Semaphore;
public class SemaphoreDemo {
    public static class ThradDemo001 extends Thread {
        private String name;
        private Semaphore wc;
        public ThradDemo001(String name, Semaphore wc) {
            this.name = name;
            this.wc = wc;
        }
        @Override
        public void run() {
            // 剩下的资源
            int availablePermits = wc.availablePermits();
            if (availablePermits > 0) {
                System.out.println(name + "天助我也，终于有茅坑了.....");
            } else {
                System.out.println(name + "怎么没有茅坑了...");
            }
            try {
                // 申请资源
                wc.acquire();
            } catch (InterruptedException e) {
            }
            System.out.println(name + "终于上厕所啦.爽啊" + ",剩下厕所:" + wc.availablePermits());
            try {
                Thread.sleep(new Random().nextInt(1000));
            } catch (Exception e) {
                // TODO: handle exception
            }
            System.out.println(name + "厕所上完啦!");
            // 释放资源
            wc.release();
        }
    }
    public static void main(String[] args) {
        Semaphore semaphore = new Semaphore(3);
        for (int i = 1; i <= 10; i++) {
            ThradDemo001 thradDemo001 = new ThradDemo001("第" + i + "个人", semaphore);
            thradDemo001.start();
        }
    }
}

运行结果：

第1个人天助我也，终于有茅坑了.....
第1个人终于上厕所啦.爽啊,剩下厕所:2
第2个人天助我也，终于有茅坑了.....
第2个人终于上厕所啦.爽啊,剩下厕所:1
第3个人天助我也，终于有茅坑了.....
第3个人终于上厕所啦.爽啊,剩下厕所:0
第4个人怎么没有茅坑了...
第5个人怎么没有茅坑了...
第6个人怎么没有茅坑了...
第7个人怎么没有茅坑了...
第8个人怎么没有茅坑了...
第9个人怎么没有茅坑了...
第10个人怎么没有茅坑了...
第1个人厕所上完啦!
第4个人终于上厕所啦.爽啊,剩下厕所:0
第2个人厕所上完啦!
第5个人终于上厕所啦.爽啊,剩下厕所:0
第4个人厕所上完啦!
第6个人终于上厕所啦.爽啊,剩下厕所:0
第3个人厕所上完啦!
第7个人终于上厕所啦.爽啊,剩下厕所:0
第6个人厕所上完啦!
第8个人终于上厕所啦.爽啊,剩下厕所:0
第8个人厕所上完啦!
第9个人终于上厕所啦.爽啊,剩下厕所:0
第5个人厕所上完啦!
第10个人终于上厕所啦.爽啊,剩下厕所:0
第9个人厕所上完啦!
第10个人厕所上完啦!
第7个人厕所上完啦!
Process finished with exit code 0

4. 并发队列

在并发队列上JDK提供了两套实现，一个是以ConcurrentLinkedQueue为代表的高性能非阻塞队列，一个是以BlockingQueue接口为代表的阻塞队列，无论哪种都继承自Queue。

5. 阻塞队列与非阻塞队

阻塞队列与普通队列的区别在于，当队列是空的时，从队列中获取元素的操作将会被阻塞，或者当队列是满时，往队列里添加元素的操作会被阻塞。试图从空的阻塞队列中获取元素的线程将会被阻塞，直到其他的线程往空的队列插入新的元素。同样，试图往已满的阻塞队列中添加新元素的线程同样也会被阻塞，直到其他的线程使队列重新变得空闲起来，如从队列中移除一个或者多个元素，或者完全清空队列。

ArrayDeque, （数组双端队列）
PriorityQueue, （优先级队列）
ConcurrentLinkedQueue, （基于链表的并发队列）
DelayQueue, （延期阻塞队列）（阻塞队列实现了BlockingQueue接口）
ArrayBlockingQueue, （基于数组的并发阻塞队列）
LinkedBlockingQueue, （基于链表的FIFO阻塞队列）
LinkedBlockingDeque, （基于链表的FIFO双端阻塞队列）
PriorityBlockingQueue, （带优先级的无界阻塞队列）
SynchronousQueue （并发同步阻塞队列）

5.1 ConcurrentLinkedDeque

ConcurrentLinkedQueue : 是一个适用于高并发场景下的队列，通过无锁的方式，实现了高并发状态下的高性能，通常ConcurrentLinkedQueue性能好于BlockingQueue.它是一个基于链接节点的无界线程安全队列。该队列的元素遵循先进先出的原则。头是最先加入的，尾是最近加入的，该队列不允许null元素。

ConcurrentLinkedQueue重要方法:

add 和offer() 都是加入元素的方法(在ConcurrentLinkedQueue中这俩个方法没有任何区别)

poll() 和peek() 都是取头元素节点，区别在于前者会删除元素，后者不会。

ConcurrentLinkedDeque q = new ConcurrentLinkedDeque();
q.offer("张三");
q.offer("李四");
q.offer("王五");
q.offer("麻溜");
//从头获取元素,删除该元素
System.out.println(q.poll());
//从头获取元素,不刪除该元素
System.out.println(q.peek());
//获取总长度
System.out.println(q.size());

运行结果：

5.2 BlockingQueue

阻塞队列（BlockingQueue）是一个支持两个附加操作的队列。这两个附加的操作是：

在队列为空时，获取元素的线程会等待队列变为非空。
当队列满时，存储元素的线程会等待队列可用。

阻塞队列常用于生产者和消费者的场景，生产者是往队列里添加元素的线程，消费者是从队列里拿元素的线程。阻塞队列就是生产者存放元素的容器，而消费者也只从容器里拿元素。

BlockingQueue即阻塞队列，从阻塞这个词可以看出，在某些情况下对阻塞队列的访问可能会造成阻塞。被阻塞的情况主要有如下两种：

当队列满了的时候进行入队列操作

当队列空了的时候进行出队列操作

因此，当一个线程试图对一个已经满了的队列进行入队列操作时，它将会被阻塞，除非有另一个线程做了出队列操作；同样，当一个线程试图对一个空队列进行出队列操作时，它将会被阻塞，除非有另一个线程进行了入队列操作。

在Java中，BlockingQueue的接口位于java.util.concurrent 包中(在Java5版本开始提供)，由上面介绍的阻塞队列的特性可知，阻塞队列是线程安全的。

在新增的Concurrent包中，BlockingQueue很好的解决了多线程中，如何高效安全“传输”数据的问题。通过这些高效并且线程安全的队列类，为我们快速搭建高质量的多线程程序带来极大的便利。本文详细介绍了BlockingQueue家庭中的所有成员，包括他们各自的功能以及常见使用场景。

阻塞队列，顾名思义，首先它是一个队列，通过一个共享的队列，可以使得数据由队列的一端输入，从另外一端输出；

常用的队列主要有以下两种：（当然通过不同的实现方式，还可以延伸出很多不同类型的队列，DelayQueue就是其中的一种）

先进先出（FIFO） ：先插入的队列的元素也最先出队列，类似于排队的功能。从某种程度上来说这种队列也体现了一种公平性。
后进先出（LIFO） ：后插入队列的元素最先出队列，这种队列优先处理最近发生的事件。

多线程环境中，通过队列可以很容易实现数据共享，比如经典的“生产者”和“消费者”模型中，通过队列可以很便利地实现两者之间的数据共享。假设我们有若干生产者线程，另外又有若干个消费者线程。如果生产者线程需要把准备好的数据共享给消费者线程，利用队列的方式来传递数据，就可以很方便地解决他们之间的数据共享问题。但如果生产者和消费者在某个时间段内，万一发生数据处理速度不匹配的情况呢？理想情况下，如果生产者产出数据的速度大于消费者消费的速度，并且当生产出来的数据累积到一定程度的时候，那么生产者必须暂停等待一下（阻塞生产者线程），以便等待消费者线程把累积的数据处理完毕，反之亦然。然而，在concurrent包发布以前，在多线程环境下，我们每个程序员都必须去自己控制这些细节，尤其还要兼顾效率和线程安全，而这会给我们的程序带来不小的复杂度。好在此时，强大的concurrent包横空出世了，而他也给我们带来了强大的BlockingQueue。（在多线程领域：所谓阻塞，在某些情况下会挂起线程（即阻塞），一旦条件满足，被挂起的线程又会自动被唤醒）。

5.2.1 ArrayBlockingQueue

ArrayBlockingQueue是一个有边界的阻塞队列，它的内部实现是一个数组。有边界的意思是它的容量是有限的，我们必须在其初始化的时候指定它的容量大小，容量大小一旦指定就不可改变。

ArrayBlockingQueue是以先进先出的方式存储数据，最新插入的对象是尾部，最新移出的对象是头部。下面是一个初始化和使用ArrayBlockingQueue的例子：

ArrayBlockingQueue<String> arrays = new ArrayBlockingQueue<>(3);
arrays.add("李四");
arrays.add("张军");
arrays.add("张军");
// 添加阻塞队列
try {
    arrays.offer("张三", 1, TimeUnit.SECONDS);//张三添加不进去
    for (String item:arrays) {
        System.out.println("item -> "+item);
    }
} catch (InterruptedException e) {
    e.printStackTrace();
}

运行结果：

5.2.2 LinkedBlockingQueue

LinkedBlockingQueue阻塞队列大小的配置是可选的，如果我们初始化时指定一个大小，它就是有边界的，如果不指定，它就是无边界的。说是无边界，其实是采用了默认大小为Integer.MAX_VALUE的容量。它的内部实现是一个链表。

和ArrayBlockingQueue一样，LinkedBlockingQueue 也是以先进先出的方式存储数据，最新插入的对象是尾部，最新移出的对象是头部。下面是一个初始化和使LinkedBlockingQueue的例子：

LinkedBlockingQueue linkedBlockingQueue = new LinkedBlockingQueue(3);
linkedBlockingQueue.add("张三");
linkedBlockingQueue.add("李四");
linkedBlockingQueue.add("李四");
System.out.println(linkedBlockingQueue.size());

5.2.3 PriorityBlockingQueue

PriorityBlockingQueue是一个没有边界的队列，它的排序规则和 java.util.PriorityQueue一样。需要注意，PriorityBlockingQueue中允许插入null对象。

所有插入PriorityBlockingQueue的对象必须实现 java.lang.Comparable接口，队列优先级的排序规则就是按照我们对这个接口的实现来定义的。

另外，我们可以从PriorityBlockingQueue获得一个迭代器Iterator，但这个迭代器并不保证按照优先级顺序进行迭代。

5.2.4 SynchronousQueue

SynchronousQueue队列内部仅允许容纳一个元素。当一个线程插入一个元素后会被阻塞，除非这个元素被另一个线程消费。

6. 使用BlockingQueue模拟生产者与消费者

package com.ylw.threadpool;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class BlockingQueueDemo {
    //生产者线程
    public static class ProducerThread implements Runnable {
        private BlockingQueue<String> blockingQueue;
        private AtomicInteger count = new AtomicInteger();
        private volatile boolean FLAG = true;
        public ProducerThread(BlockingQueue<String> blockingQueue) {
            this.blockingQueue = blockingQueue;
        }
        @Override
        public void run() {
            while (FLAG) {
                String data = count.incrementAndGet() + "";
                try {
                    boolean offer = blockingQueue.offer(data, 2, TimeUnit.SECONDS);
                    if (offer) {
                        System.out.println(Thread.currentThread().getName() + "-> 生产【" + data + "】成功");
                    } else {
                        System.out.println(Thread.currentThread().getName() + "-> 生产【" + data + "】失败");
                    }
                    Thread.sleep(1000);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            System.out.println(Thread.currentThread().getName() + "生产者线程停止...");
        }
        public void stop() {
            this.FLAG = false;
        }
    }
    //消费者线程
    static class ConsumerThread implements Runnable {
        private volatile boolean FLAG = true;
        private BlockingQueue<String> blockingQueue;
        public ConsumerThread(BlockingQueue<String> blockingQueue) {
            this.blockingQueue = blockingQueue;
        }
        @Override
        public void run() {
            while (FLAG) {
                try {
                    String data = blockingQueue.poll(2, TimeUnit.SECONDS);
                    if (data == null || data == "") {
                        FLAG = false;
                        System.out.println("消费者超过2秒时间未获取到消息.");
                        return;
                    }
                    System.out.println("消费者消费 ->【" + data + "】");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
    public static void main(String[] args) {
        BlockingQueue<String> blockingQueue = new LinkedBlockingQueue<>(3);
        ProducerThread producerThread = new ProducerThread(blockingQueue);
        ConsumerThread consumerThread = new ConsumerThread(blockingQueue);
        Thread t1 = new Thread(producerThread);
        Thread t2 = new Thread(consumerThread);
        t1.start();
        t2.start();
        //10秒后 停止线程..
        try {
            Thread.sleep(10 * 1000);
            producerThread.stop();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

并发编程系列教程（07） - 线程池原理分析（一）

1. CountDownLatch（计数器）

2. CyclicBarrier（屏障）

3. Semaphore（计数信号量）

4. 并发队列

5. 阻塞队列与非阻塞队

5.1 ConcurrentLinkedDeque

5.2 BlockingQueue

5.2.1 ArrayBlockingQueue

5.2.2 LinkedBlockingQueue

5.2.3 PriorityBlockingQueue

5.2.4 SynchronousQueue

6. 使用BlockingQueue模拟生产者与消费者

总结

热门文章

最新文章

相关电子书

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

并发编程系列教程（07） - 线程池原理分析（一）

1. CountDownLatch（计数器）

2. CyclicBarrier（屏障）

3. Semaphore（计数信号量）

4. 并发队列

5. 阻塞队列与非阻塞队

5.1 ConcurrentLinkedDeque

5.2 BlockingQueue

5.2.1 ArrayBlockingQueue

5.2.2 LinkedBlockingQueue

5.2.3 PriorityBlockingQueue

5.2.4 SynchronousQueue

6. 使用BlockingQueue模拟生产者与消费者

总结

热门文章

最新文章

相关电子书