Java 阻塞队列，并发工具 - libin372767477的专栏 - CSDN博客
阻塞队列
阻塞队列支持阻塞插入和阻塞移除
- 阻塞插入方法：就是当队列满时，队列会阻塞插入元素的线程，直到队列不满。
- 阻塞移除方法：就是当对为空时，获取元素的线程会等待队列变为非空。
方法/处理方式
抛出异常
返回特殊值
一直阻塞
超时退出
插入方法
add(e)
offer(e)
put(e)
offer(e, time, unit)
移除方法
remove()
poll()
take()
poll(time, unit)
检查方法
element()
peek()
不可用
不可用
- 抛出异常：当队列满时，如果再往队列里面插入元素，会抛出异常。当队列空时，从队列里获取元素也会抛出异常。
- 返回特殊值：当前队列插入元素时，会返回元素是否成功插入，成功返回true。如果时移除方法，则是从队列里取出一个元素，如果没有则返回null。
- 一直阻塞：当阻塞队列满时，如果生产者线程往队列里put元素，队列会一直阻塞生产者线程，直到队列可用或者响应中断退出。当队列空时，如果消费者线程从队列里take元素，队列会阻塞住消费者线程，直到队列不为空。
- 超时退出：当阻塞队列满时，如果生产者线程往队列里插入元素，队列会阻塞生产者线程一段时间，如果超过了指定时间，生产者线程就会退出。
如果是无界阻塞队列，队列不可能会出现满的情况，所以使用put，offer方法永远不会被阻塞，而使用offer方法时，该方法永远返回true。
Java里的阻塞队列
- ArrayBlockingQueue：一个由数组结构组成的有界阻塞队列。
- LinkedBlockingQueue：一个由链表结构组成的有界阻塞队列。
- PriorityBlockingQueue：一个支持优先级排序的无界阻塞队列。
- DelayQueue：一个使用优先级队列实现的无界阻塞队列。
- SynchronousQueue：一个不存储元素的阻塞队列。
- LinkedTransferQueue：一个由链表结构组成的无界阻塞队列
- LinkedBlockingDeque：一个由链表结构组成的双向阻塞队列。
ArrayBlockingQueue
ArrayBlockingQueue是一个用数组实现的有界阻塞队列，此队列按照先进先出（FIFO）的原则对元素进行排序。
默认情况下不保证线程公平的访问队列，可以使用代码创建一个公平的阻塞队列。
ArrayBlockingQueue fairQueue = new ArrayBlockingQueue（1000， true）；
LinkedBlockingQueue
LinkedBlockingQueue是一个用链表实现的有界阻塞队列，此队列的默认和最大长度为Integer.MAX_VALUE，按照先进先出的原则对元素进行排序。
PriorityBlockingQueue
PriorityBlockingQueue是一个支持优先级的无界阻塞队列，默认采用自然顺序升序排序，也可以自定义类实现compareTo()方法来指定元素排序规则，或者在初始化的时候，指定构造参数Comparator来对元素进行排序。不能保证同优先级元素的顺序。
DelayQueue
DelayQueue是一个支持掩饰获取元素的无界阻塞队列。队列使用PriorityBlockingQueue来实现，队列元素必须实现Delayed接口，在创建元素时可以指定多久才能从队列中获取当前元素。只有在延迟期满的时候才能从队列提取元素。
SynchronousQueue
SynchronousQueue是一个不存储元素的阻塞队列，每个put操作必须等待一个take操作。否则不能继续添加元素。
支持公平访问对了。默认是非公平的。
SynchronousQueue可以看成是一个传球手，负责把生产者线程处理的数据直接传递给消费线程。队列本身不存储任何元素，适合传递性场景，SynchronousQueue吞吐量高于ArrayBlockingQueue和LinkedBlockingQueue
LinkedTransferQueue
LinkedTransferQueue是一个由链表结构组成的无界阻塞队列，相对于其他阻塞队列，LinkedTransferQueue多了tryTransfer和transfer方法
- transfer()，如果当前由消费者正在等待接收元素，transfer方法可以吧把生产者传入的元素立刻transfer给消费者，如果没有消费者等待接收元素，transfer方法将元素放在队列的tail节点，并等到该元素被消费者消费了才返回
- tryTransfer()，采用试探生产者传入的元素是否能直接传给消费者，如果没有消费者等待接收元素，则返回false，和transfer方法的区别是tryTransfer方法无论消费者是否接收，方法立即返回，而transfer方法是必须等到消费者消费了才返回。
LinkedBlockingDeque
LinkedBlockingDeque是一个由链表结构组成的双向阻塞队列，
在初始化LinkedBlockingDeque时就可以设置容量防止其过度膨胀。可以运用在“工作窃取”模式中。
Fork/Join框架
Fork/Join框架是Java7提供的一个用于并行执行任务的框架，是一个把大任务分割称若干个小任务，最终汇总每个小任务结果后得到大任务结果的框架。
设计
- 任务分割，首先有个fork类来把大任务分割成子任务，有可能子任务还是很大，所以要不停的分割，知道分割出的子任务足够小
- 执行任务并合并结果，分割的子任务分别放在双端队列里，然后几个启动线程分别从双端队列里获取任务执行，子任务执行完的结果都统一放在一个队列里，启动一个线程从队列里那数据，然后合并这些数据。
Fork/Join使用两个类来完成以上两件事情。
- ForkJoinTask：我们要使用ForkJoin，首先创建一个ForkJoin任务。它提供在任务中执行fork() 和 join()操作的机制。通常情况下，我们不需要直接继承ForkJoinTask类，只需要继承它的子类。
- RecursiveAction：用于没有返回结果的任务。
- RecursiveTask： 用于有返回结果的任务。
2. ForkJoinPool：ForkJoinTask需要通过ForkJoinPool来执行。
任务分割出的子任务会添加到当前工作线程维护的双端队列中，进入队列的头部。当一个工作线程的队列里暂时没有任务时，它会随机从其他工作线程的队列的尾部获取一个任务。
public class ForkJoinCalculator implements Calculator { private ForkJoinPool pool ; public ForkJoinCalculator(){ pool = new ForkJoinPool(); } @Override public long sumUp(long[] numbers) { return pool.invoke(new SumTask(numbers, 0 ,numbers.length)); } private static class SumTask extends RecursiveTask<Long>{ private long[] numbers ; private int form; private int to; public SumTask(long[] numbers, int form, int to){ this.numbers = numbers; this.form = form; this.to = to; } @Override protected Long compute() { if(to - form < 6 ){ long total = 0; for (int i = form; i < to; i++) { total += numbers[i]; } return total; }else{ int middle = ( form + to ) / 2; System.out.println(Thread.currentThread().getName()+":"+middle); SumTask sumLeft = new SumTask(numbers, form, middle); SumTask sumRight = new SumTask(numbers, middle+1, to); sumLeft.fork(); sumRight.fork(); return sumLeft.join() + sumRight.join(); } } } public static void main(String[] args) { //1,2,3,4 ForkJoinCalculator forkJoinCalculator = new ForkJoinCalculator(); long[] parameter = new long[5]; for (int i = 1; i <= 4; i++) { parameter[i-1] = i; } for (long l : parameter) { System.out.println(l); } long l = forkJoinCalculator.sumUp(parameter); System.out.println(l); }}异常处理
ForkJoinTask在出现异常的时候，无法在主线程里面直接捕获异常，所以ForkJoinTask提供了isCompletedAbnormally()方法来检查任务是否已经抛出异常或已经被取消了，
getException方法返回Throwable对象，如果任务被取消了则返回CancellationException。如果任务没有完成或者没有抛出异常则返回null。
Java中的并发工具类
等待多线程完成的CountDownLatch
CountDownLatch允许一个或多个线程等待其他线程完成操作。
public static void main(String[] args) { try { CountDownLatch countDownLatch = new CountDownLatch(2); new Thread(new Runnable() { @Override public void run() { System.out.println(Thread.currentThread().getName()); try { Thread.sleep(5000); } catch (InterruptedException e) { e.printStackTrace(); } countDownLatch.countDown(); System.out.println(Thread.currentThread().getName()+"OK"); } }).start(); new Thread(new Runnable() { @Override public void run() { System.out.println(Thread.currentThread().getName()); try { Thread.sleep(5000); } catch (InterruptedException e) { e.printStackTrace(); } countDownLatch.countDown(); System.out.println(Thread.currentThread().getName()+"OK"); } }).start(); System.out.println("我要等待了"); countDownLatch.await(); System.out.println("我要结束了"); } catch (InterruptedException e) { e.printStackTrace(); } }public static void main(String[] args) { final int countSize = 5; //任务线程 CountDownLatch workThreadCountDownLatch = new CountDownLatch(countSize); //主线程 CountDownLatch mainThreadCountDownLatch = new CountDownLatch(1); try { for (int i = 0; i < countSize; i++) { new Thread(() -> { try { mainThreadCountDownLatch.await(); System.out.println(Thread.currentThread().getName()); Thread.sleep(5000); workThreadCountDownLatch.countDown(); System.out.println(Thread.currentThread().getName() + "OK"); } catch (Exception e) { e.printStackTrace(); } }).start(); } System.out.println("主线程操作一些事情"); Thread.sleep(3000); System.out.println("主线程操作一些事情结束,开始子线程干活，主线程等待"); } catch (InterruptedException e) { e.printStackTrace(); }finally { mainThreadCountDownLatch.countDown(); try { workThreadCountDownLatch.await(); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("结束了"); } }CountDownLatch的构造函数接收一个int类型的参数作为计数器，如果你想等待N个点完成，这里就传入N。
当调用CountDownLatch的countDown方法是，N就会减1，CountDownLatch的await方法会阻塞当前线程，知道N变成0，由于countDown方法可以用在任何地方，所以这里说的N个点，可以是N个线程，也可以是1个线程里面的N个执行步骤。用在多线程时，只需要把这个CountDownLatch的引用传递到线程里即可。
同步屏障 CyclicBarrier
CyclicBarrier要做的事情是，让一组线程到达一个屏障（也可以叫同步点）时被阻塞，知道最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续运行。
CyclicBarrier默认的构造方法时CyclicBarrier(int parties)，其参数表示屏障拦截的线程数量，每个线程调用await方法告诉CyclicBarrier我已经到达了屏障点，然后当前线程被阻塞，
public class ThreadCallbackCyclicBarrier { private static final int THREAD_COUNT=3; private final static CyclicBarrier CYCLIC_BARRIER = new CyclicBarrier(THREAD_COUNT); public static void main(String[] args) { ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT); for (int i = 0; i <THREAD_COUNT; i++) { executorService.execute(new Runnable() { @Override public void run() { System.out.println("我是线程"+Thread.currentThread().getName()+"我们到达旅游地点"); try { CYCLIC_BARRIER.await(); } catch (InterruptedException e) { e.printStackTrace(); } catch (BrokenBarrierException e) { e.printStackTrace(); } System.out.println("我是线程"+Thread.currentThread().getName()+"我们开始骑车"); try { CYCLIC_BARRIER.await(); } catch (InterruptedException e) { e.printStackTrace(); } catch (BrokenBarrierException e) { e.printStackTrace(); } System.out.println("我是线程"+Thread.currentThread().getName()+"我们开始爬山"); try { CYCLIC_BARRIER.await(); } catch (InterruptedException e) { e.printStackTrace(); } catch (BrokenBarrierException e) { e.printStackTrace(); } System.out.println("我是线程"+Thread.currentThread().getName()+"我们回家了"); try { CYCLIC_BARRIER.await(); } catch (InterruptedException e) { e.printStackTrace(); } catch (BrokenBarrierException e) { e.printStackTrace(); } System.out.println("我是线程"+Thread.currentThread().getName()+"我们到家了"); } }); } executorService.shutdown(); }}主线程和子线程的调度是由CPU决定的，
CyclicBarrier还提供了一个更高级的构造函数，CyclicBarrier(int parties, Runnable barrierAction)，用于在线程到达屏障时，优先执行barrierAction,方便处理更复杂的业务场景。
public class ThreadCallbackCyclicBarrierBarrierAction { private static final int THREAD_COUNT=3; private final static CyclicBarrier CYCLIC_BARRIER = new CyclicBarrier(THREAD_COUNT, new barrierAction()); public static void main(String[] args) { ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT); for (int i = 0; i <THREAD_COUNT; i++) { executorService.execute(new Runnable() { @Override public void run() { System.out.println("我是线程"+Thread.currentThread().getName()+"我们到达旅游地点"); try { CYCLIC_BARRIER.await(); } catch (InterruptedException e) { e.printStackTrace(); } catch (BrokenBarrierException e) { e.printStackTrace(); } System.out.println("我是线程"+Thread.currentThread().getName()+"我们开始骑车"); try { CYCLIC_BARRIER.await(); } catch (InterruptedException e) { e.printStackTrace(); } catch (BrokenBarrierException e) { e.printStackTrace(); } System.out.println("我是线程"+Thread.currentThread().getName()+"我们开始爬山"); try { CYCLIC_BARRIER.await(); } catch (InterruptedException e) { e.printStackTrace(); } catch (BrokenBarrierException e) { e.printStackTrace(); } System.out.println("我是线程"+Thread.currentThread().getName()+"我们回家了"); try { CYCLIC_BARRIER.await(); } catch (InterruptedException e) { e.printStackTrace(); } catch (BrokenBarrierException e) { e.printStackTrace(); } System.out.println("我是线程"+Thread.currentThread().getName()+"我们到家了"); } }); } executorService.shutdown(); } static class barrierAction implements Runnable{ @Override public void run() { System.out.println("我是barrierAction"); } }}CyclicBarrier可以用于多线程计算数据，最后合并计算结果的场景，
CyclicBarrier和CountDownLatch的区别
CountDownLatch计数器只能使用一次，而CyclicBarrier的技术去可以使用reset()方法重置，所以CyclicBarrier能处理更为复杂的场景，
Semaphore信号量
Semaphore用来控制同时访问特定资源的线程数量，它通过协调各个线程，以保证合理的使用公共资源。
public class SemaphoreTest { public static void main(String[] args) { ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor( 5, 200, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(1024), new ThreadFactoryBuilder().setNameFormat("demo-pool-%d").build(), new ThreadPoolExecutor.AbortPolicy() ); Semaphore semaphore = new Semaphore(3); for (int i = 0; i < 30; i++) { threadPoolExecutor.execute(new Runnable() { @Override public void run() { try { semaphore.acquire(); System.out.println("拿到信号灯，做了一些事,剩余信号灯数量"+semaphore.availablePermits()); Thread.sleep(1000); semaphore.release(); } catch (InterruptedException e) { e.printStackTrace(); } } }); } threadPoolExecutor.shutdown(); }}线程间交换数据的Exchanger
Exchanger是一个用于线程间协作的工具类，Exchanger用于进行线程间的数据交换。
如果一个线程执行了exchanger()方法，他会一直等待第二个线程也执行exchanger()方法，当两个线程都到达同步点时，这两个线程可以交换数据，将笨线程生产出来的数据传递给对方。
public class ExchangeTest { public static void main(String[] args) { Exchanger<String> exchanger = new Exchanger<String>(); ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor( 5, 200, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(1024), new ThreadFactoryBuilder().setNameFormat("demo-pool-%d").build(), new ThreadPoolExecutor.AbortPolicy() ); for (int i = 0; i < 4; i++) { threadPoolExecutor.execute(new Runnable() { @Override public void run() { String str = Thread.currentThread().getName() + "的数据"; System.out.println("exchange start， " +str); try { String exchange = exchanger.exchange(str); System.out.println(Thread.currentThread().getName()+" exchange ， " +exchange); } catch (InterruptedException e) { e.printStackTrace(); } } }); } threadPoolExecutor.shutdown(); } }Original url: Access
Created at: 2019-10-28 11:43:45
Category: default
Tags: none
未标明原创文章均为采集，版权归作者所有，转载无需和我联系，请注明原出处，南摩阿彌陀佛，知识，不只知道，要得到
最新评论