java多线程进阶JUC工具集

Posted 烟锁迷城

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了java多线程进阶JUC工具集相关的知识,希望对你有一定的参考价值。

目录

1、阻塞队列

2、CountDownLatch

2.1、基础应用

2.2、应用场景

2.3、原理分析

3、Semaphore

4、CyclicBarrier


1、阻塞队列

队列是一种线程表,可以一端插入,一端删除,是先进先出的基础结构。

阻塞队列,需要支持两种情况

  1. 在队列满溢的时候,添加线程需要被阻塞,删除线程需要被唤醒
  2. 在队列为空的时候,添加线程需要被唤醒,删除线程需要被阻塞

由此可见,其实这就是一个生产者消费者模型。

在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个子线程全部执行完毕,主线程才执行工作。
需要注意的是

  1. 如果没有足够多的线程执行await方法,所有的线程都会被阻塞
  2. await方法可以使用参数:await(long timeout, TimeUnit unit)此为超时时间与超时单位,这样可以避免全部阻塞
  3. 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工具集的主要内容,如果未能解决你的问题,请参考以下文章

多线程进阶=;JUC编程

多线程进阶=>JUC并发编程

多线程进阶(JUC-1)准备知识

JUC系列同步工具类之CyclicBarrier

java多线程进阶LOCK锁及其原理

java多线程进阶LOCK锁及其原理