java多线程进阶JUC工具集
Posted 烟锁迷城
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了java多线程进阶JUC工具集相关的知识,希望对你有一定的参考价值。
目录
1、阻塞队列
队列是一种线程表,可以一端插入,一端删除,是先进先出的基础结构。
阻塞队列,需要支持两种情况
- 在队列满溢的时候,添加线程需要被阻塞,删除线程需要被唤醒
- 在队列为空的时候,添加线程需要被唤醒,删除线程需要被阻塞
由此可见,其实这就是一个生产者消费者模型。
在JUC之中,有很多的阻塞队列实现,以下是一些通用方法
添加方法
- add:添加元素,如果队列满了,则抛出异常
- offer:添加元素,返回true/false,添加成功,返回true,否则返回false
- put:添加元素,如果队列满了,则一直阻塞线程
- offer(timeout):添加元素,附带超时时间,如果队列满了,则阻塞线程timeout的时长,如果超时还没有,则返回false。
移除方法
- element:移除元素,如果队列为空,抛出异常
- peek:移除元素,返回true/false,移除成功,返回true,否则返回false
- take:移除元素,如果队列为空,则一直阻塞线程
- poll(timeout):移除元素,附带超时时间,如果队列为空,则阻塞线程timeout的时长,如果超时还没有,则返回null。
JUC实现队列举例:
- ArrayBlockQueue:基于数组实现的队列
- LinkedBlockQueue:基于链表实现的队列
- PriorityBlockingQueue:具有优先级的队列
- DelayQueue:允许延时执行的队列
- SynchronousQueue:没有存储结构的队列(用在线程池newCachedThreadPool中)
- LinkedTransferQueue:无界阻塞队列
- LinkedBlockingQueue:基于链表实现的无界阻塞队列
- LinkedBlockingDeque:基于双向链表实现的队列
2、CountDownLatch
countdownlatch是一个同步工具类,它允许一个或多个线程一直等待,直到其他线程的操作执行完毕再执行。
2.1、基础应用
在示例代码中,CountDownLatch countDownLatch = new CountDownLatch(3);指的是倒计时3,countDownLatch.countDown();意为倒计时启动,会将倒计时的设置数减去1,等到倒计时数等于0时,线程将会启动。
public class CountDownLatchDemo
public static void main(String[] args) throws InterruptedException
CountDownLatch countDownLatch = new CountDownLatch(3);
new Thread(()->
countDownLatch.countDown();
).start();
new Thread(()->
countDownLatch.countDown();
).start();
new Thread(()->
countDownLatch.countDown();
).start();
countDownLatch.await();
System.out.println("end");
2.2、应用场景
可以当循环的100个线程进入等待,直到主线程计数减1,等待的100个线程将同时启动
public class CountDownLatchDemo
public static void main(String[] args) throws InterruptedException
CountDownLatch countDownLatch = new CountDownLatch(1);
IntStream.range(1,100).forEach(i->
new Thread(()->
try
countDownLatch.await();
catch (InterruptedException e)
e.printStackTrace();
System.out.println("end");
).start();
);
Thread.sleep(5000);
countDownLatch.countDown();
2.3、原理分析
在CountDownLatch的await方法中,如果线程中断就抛出中断异常,未中断就判断倒计时数是否为0,若为0就返回1,反之返回-1。
若小于0,即倒计时还未结束,执行doAcquireSharedInterruptibly方法。
public void await() throws InterruptedException
sync.acquireSharedInterruptibly(1);
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
protected int tryAcquireShared(int acquires)
return (getState() == 0) ? 1 : -1;
addWaiter方法在AQS中见到过,它会将线程组成一个双向链表,同时循环判断节点的前一个节点是否为头结点,若为头结点,则判断倒计时数是否为0,若为0就返回1,反之返回-1。若大于等于0,即倒计时已经结束,等待结束。
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try
for (;;)
final Node p = node.predecessor();
if (p == head)
int r = tryAcquireShared(arg);
if (r >= 0)
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
finally
if (failed)
cancelAcquire(node);
在CountDownLatch的countDown方法中,获取当前倒计时数,如果不为0,就减去1,使用CAS更新当前倒计时数,若此时倒计时数为0,返回true,执行doReleaseShared方法。
public void countDown()
sync.releaseShared(1);
public final boolean releaseShared(int arg)
if (tryReleaseShared(arg))
doReleaseShared();
return true;
return false;
protected boolean tryReleaseShared(int releases)
for (;;)
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
if (compareAndSetState(c, nextc))
return nextc == 0;
doReleaseShared方法,将双向链表中的全部线程依次唤醒。逻辑当unparkSuccessor方法唤醒头结点的下一个节点,也就是第一个线程之后,CAS将改变线程的状态,随后根据AQS原理,头结点将被删除,第二个节点将变为头结点。如果h节点为头结点,则代表整个链表都已经被唤醒,完成循环唤醒的操作,这就是共享锁的依次唤醒,上一个线程唤醒下一个线程。
private void doReleaseShared()
for (;;)
Node h = head;
if (h != null && h != tail)
int ws = h.waitStatus;
if (ws == Node.SIGNAL)
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h);
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
if (h == head) // loop if head changed
break;
private void unparkSuccessor(Node node)
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
Node s = node.next;
if (s == null || s.waitStatus > 0)
s = null;
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0)
s = t;
if (s != null)
LockSupport.unpark(s.thread);
3、Semaphore
semaphore也就是信号灯,可以控制同时访问的线程个数,通过acquire获取一个许可,如果没有就等待,通过release释放一个许可。
构造semaphore的参数表示当前能够处理的最大线程数量,semaphore.acquire();获取许可,semaphore.release();释放许可,当许可发放数目达到最大数量之后,其他线程都会进入等待,直到这三个线程处理完成,释放许可,其他线程才能开始运行。
public class SemaphoreDemo
static Semaphore semaphore = new Semaphore(3);
public static void main(String[] args)
IntStream.range(1,10).forEach(i->
new OneThread(i).start();
);
private static class OneThread extends Thread
private int number;
public OneThread(int number)
this.number = number;
@Override
public void run()
try
//获得许可
semaphore.acquire();
System.out.println(number);
Thread.sleep(5000);
//释放许可
semaphore.release();
catch (InterruptedException e)
e.printStackTrace();
决定最大线程数量的参数被放入AQS的state中,tryAcquireShared将会减少state的计数数量,大于0代表许可仍有剩余,反之许可不足。
公平锁将先查看是否已经有等待线程,若是有,就不会发放许可,非公平锁就没有这个筛选。
当许可不足,就会执行doAcquireSharedInterruptibly方法,同样构造一个双向链表,线程开始等待。
许可释放,线程会被唤醒,许可数目增加1。
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
//非公平
final int nonfairTryAcquireShared(int acquires)
for (;;)
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
//公平
protected int tryAcquireShared(int acquires)
for (;;)
if (hasQueuedPredecessors())
return -1;
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
4、CyclicBarrier
CyclicBarrier的是可循环使用屏障,它要做的事情是让一组线程达到一个屏障(也可以叫同步点)时被阻塞,直到最后一个线程到达屏障时,屏障才会开门,所有被屏障拦截的线程才会继续工作,需要子任务全部执行完毕时主任务才执行的情况下可以使用,初始化的数字参数是需要等待的子任务数。
在示例中,cyclicBarrier.await();方法让5个子线程全部执行完毕,主线程才执行工作。
需要注意的是
- 如果没有足够多的线程执行await方法,所有的线程都会被阻塞
- await方法可以使用参数:await(long timeout, TimeUnit unit)此为超时时间与超时单位,这样可以避免全部阻塞
- cyclicBarrier.reset();可以重置计数
public class DataDemo extends Thread
private String name;
private CyclicBarrier cyclicBarrier;
public DataDemo(String name, CyclicBarrier barrier)
this.name = name;
this.barrier = barrier;
@Override
public void run()
System.out.println(name);
try
cyclicBarrier.await();
catch (InterruptedException e)
e.printStackTrace();
catch (BrokenBarrierException e)
e.printStackTrace();
具体使用
public class CyclicBarrierDemo extends Thread
@Override
public void run()
System.out.println("指令汇集");
public static void main(String[] args)
CyclicBarrier cyclicBarrier = new CyclicBarrier(5,new CyclicBarrierDemo());
new DataDemo("1",cyclicBarrier).start();
new DataDemo("2",cyclicBarrier).start();
new DataDemo("3",cyclicBarrier).start();
new DataDemo("4",cyclicBarrier).start();
new DataDemo("5",cyclicBarrier).start();
以上是关于java多线程进阶JUC工具集的主要内容,如果未能解决你的问题,请参考以下文章