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锁及其原理