目录

简介

  在编写多线程程序时,难免需要对并发流程进行控制,Thread类有join()和yield()等方法,JUC提供了更为灵活的并发工具类,下面就学习这些工具类的用法以及实现。

CountDownLatch

  latch意思是门闩,countdown指从上往下数,CountDownLatch允许一个或多个线程等待其他任务线程完成操作,就像它的字面意思:从大往小数,数到某个值(0)的时候打开门闩。下面是CountDownLatch的api:

    //构造器
    public CountDownLatch(int count);

    //调用await()方法的线程会进入等待状态,它会等待直到count值为0才继续执行
    public void await();
    //和await()类似,只不过等待一定的时间后count值还没变为0的话就会继续执行
    public boolean await(long timeout, TimeUnit unit);
    //计数器减一
    public void countDown()

可以看到通过构造器构造一个计数器,通过调用countDown方法计数减小,await在计数器大于0时线程处于等待状态,通过下面例子可以学会CountDownLatch的用法:

示例

public class LatchTest {
    public static void main(String[] args) {
        //两个线程,计数器传入2
        final CountDownLatch latch = new CountDownLatch(2);

        //这两个线程执行了latch.countDown(),计数器归0,主线程才被唤醒继续执行
        new Thread(() -> {
            try {
                System.out.println("子线程1: "+Thread.currentThread().getName()+"正在执行");
                Thread.sleep(3000);
                System.out.println("子线程1: "+Thread.currentThread().getName()+"执行完毕");
                latch.countDown();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

        new Thread(() -> {
            try {
                System.out.println("子线程2: "+Thread.currentThread().getName()+"正在执行");
                Thread.sleep(3000);
                System.out.println("子线程2: "+Thread.currentThread().getName()+"执行完毕");
                latch.countDown();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

        try {
            System.out.println("等待2个子线程执行完毕...");
            latch.await();
            System.out.println("2个子线程已经执行完毕");
            System.out.println("继续执行主线程");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

运行结果:
多线程学习笔记六之并发工具类CountDownLatch和CyclicBarrier-LMLPHP

实现分析

  CountDownLatch是基于共享锁实现的,内部类Sync继承同步器AQS,重点分析CountDownLatch以下三个方法:

构造方法

  通过构造函数传入的参数count设置同步状态(count必须大于0,否则抛出异常),同步状态在这里并不表示线程获得锁的重入次数,而是表示一个计数器,计数器的大小与任务线程的数目是一致的,

    public CountDownLatch(int count) {
        if (count < 0) throw new IllegalArgumentException("count < 0");
        this.sync = new Sync(count);
    }

    Sync(int count) {
        setState(count);
    }

await()

  调用了await的线程会处于等待状态,直到计数器归0才会被唤醒。await方法调用了Sync父类AQS的acquireSharedInterruptibly方法,acquireSharedInterruptibly首先检查线程有中断,然后调用tryAcquireShared尝试获取共享锁,获取成功返回1,失败返回-1,若失败调用doAcquireSharedInterruptibly将当前线程加入同步队列阻塞住,等待计数器为0唤醒。

    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;
    }

countDown()

  countDown方法将计数器减一,调用了AQS的releaseShared方法,当tryReleaseShared方法返回true执行doReleaseShared方法,这个方法在分析读写锁是介绍过了,就是唤醒同步等列等待获取锁的线程,即唤醒调用了await方法等待计数器归0的线程。

    public void countDown() {
        sync.releaseShared(1);
    }

    public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }
  • tryReleaseShared(int releases)
      通过循环+CAS的方式修改同步状态state,当同步状态为0时返回true;同步状态为0,即表示计数器归0,所有调用了countDown的线程都执行完了,可以唤醒调用await等待的线程了。
    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;
        }
    }

CountDownLatch与Thread.join()

  Thread类的join方法与CountDownLatch作用类似,join方法的实现原理不停检查调用join的线程是否存活,如果存活则让当前线程处于等待状态,当join线程终止后,会唤醒当前线程。CountDownLatch与join相比更灵活,不必非得线程中止只要调用了countDown方法就行了,可以响应中断以及能够设置超时等功能。

CyclicBarrier

  CyclicBarrier是指可循环使用的屏障,它可以让一组线程当他们分别达到了同步点(common barrier point)时被阻塞,直到最后一个线程到达了同步点,屏障才会开门,让所有被屏障屏蔽的线程继续运行。

public class BarrierTest {
    public static void main(String[] args) {
        int size = 4;
        CyclicBarrier barrier  = new CyclicBarrier(size);
        for(int i=0;i<size;i++)
            new Writer(barrier).start();
    }
    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()+" is coming...");
            try {
                //睡眠模拟业务操作
                Thread.sleep(5000);
                System.out.println("线程"+Thread.currentThread().getName()+" is waiting on barrier");
                cyclicBarrier.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }catch(BrokenBarrierException e){
                e.printStackTrace();
            }
        }
    }
}

运行结果:
多线程学习笔记六之并发工具类CountDownLatch和CyclicBarrier-LMLPHP

实现分析

类属性及构造方法

public class CyclicBarrier {

    //CyclicBarrier使用完了可以重置,每使用一次都会有一个新的Generation对象,broken表示当前屏障是否被损坏
    private static class Generation {
        boolean broken = false;
    }

    //重入锁
    private final ReentrantLock lock = new ReentrantLock();
    //condition实现线程等待与唤醒
    private final Condition trip = lock.newCondition();
    //表示线程数,在parties个线程都调用await方法后,barrier才算是被通过(tripped)了。
    private final int parties;
    //通过构造方法设置一个Runnable对象,用来在所有线程都到达barrier时执行。
    private final Runnable barrierCommand;
    /** The current generation */
    private Generation generation = new Generation();

    //count表示还剩下未到达barrier(未调用await)的线程数量
    private int count;

    //构造函数
    public CyclicBarrier(int parties, Runnable barrierAction) {
        if (parties <= 0) throw new IllegalArgumentException();
        this.parties = parties;
        this.count = parties;
        this.barrierCommand = barrierAction;
    }

    public CyclicBarrier(int parties) {
        this(parties, null);
    }

await()

  await重载的两种方法都是调用的doWait方法。

    public int await() throws InterruptedException, BrokenBarrierException {
        try {
            return dowait(false, 0L);
        } catch (TimeoutException toe) {
            throw new Error(toe); // cannot happen
        }
    }

    public int await(long timeout, TimeUnit unit)
        throws InterruptedException,
               BrokenBarrierException,
               TimeoutException {
        return dowait(true, unit.toNanos(timeout));
    }
  • doWait(boolean timed, long nanos)
      doWait是await的核心方法,通过独占锁和Condition对象让线程阻塞等待,具体先判断当前线程是不是最后一个执行await方法的线程,如果不是,调用condition的await方法让线程等待,在这里我们看到首先线程会获得锁,进入同步块,在循环里让线程等待,这里因为当前线程获得了独占锁,它处于同步队列的head头节点之中,当调用了condition.await()方法后,当前线程从同步队列转移到条件队列,释放了独占锁,所以当前线程获取独占锁并不会影响后来的线程获取独占锁,因为当前线程进入阻塞状态已经释放了独占锁,直到被唤醒后才会去争取获得独占锁,到最后会在finally块中显示的释放。
    private int dowait(boolean timed, long nanos)
        throws InterruptedException, BrokenBarrierException,
               TimeoutException {
        //独占锁
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            //Generation对象
            final Generation g = generation;

            //屏障被破坏,抛出异常
            if (g.broken)
                throw new BrokenBarrierException();

            //线程被中断
            if (Thread.interrupted()) {
                breakBarrier();
                throw new InterruptedException();
            }

            int index = --count;
            //最后一个到达同步点的线程
            if (index == 0) {  // tripped
                boolean ranAction = false;
                try {
                    final Runnable command = barrierCommand;
                    if (command != null)
                        command.run();
                    ranAction = true;
                    nextGeneration();
                    return 0;
                } finally {
                    if (!ranAction)
                        breakBarrier();
                }
            }

            // loop until tripped, broken, interrupted, or timed out
            //一直循环直到最后一个线程到达同步点、屏障破损(genneration的broken属性为true)、中断或超时
            for (;;) {
                try {
                    if (!timed)
                        trip.await();
                    else if (nanos > 0L)
                        nanos = trip.awaitNanos(nanos);
                } catch (InterruptedException ie) {
                    //g == generation && !g.broken说明此时当前这一轮还没结束,并且没有其它线程执行过
                    //breakBarrier方法。这种情况会执行breakBarrier置generation的broken标识为true并
                    //唤醒其它线程,之后继续抛出InterruptedException。
                    if (g == generation && ! g.broken) {
                        breakBarrier();
                        throw ie;
                    } else {
                        // 如果g != generation,此时这一轮已经结束,后面返回index作为到达barrier的次序;
                        // 如果g.broken说明之前已经有其它线程执行了breakBarrier方法,后面会抛出
                        //BrokenBarrierException。
                        Thread.currentThread().interrupt();
                    }
                }

                if (g.broken)
                    throw new BrokenBarrierException();

                if (g != generation)
                    return index;

                //超时
                if (timed && nanos <= 0L) {
                    breakBarrier();
                    throw new TimeoutException();
                }
            }
        } finally {
            lock.unlock();
        }
    }
  • breakBarrier()
      损坏当前屏障,会唤醒所有在屏障中的线程,当线程被中断或等待超时会调用
    private void breakBarrier() {
        generation.broken = true;
        count = parties;
        trip.signalAll();
    }
  • nextGeneration()
      nextGeneration方法在所有线程进入屏障后会被调用,即生成下一个版本,所有线程又可以重新进入到屏障中
    private void nextGeneration() {
        // signal completion of last generation
        trip.signalAll();
        // set up next generation
        count = parties;
        generation = new Generation();
    }

CountDownLatch和CyclicBarrier区别

  从功能上说,CountDownLatch允许一个或多个线程等待其他线程完成操作,而CyclicBarrier是让一组线程达到一个公共同步点之后再一起放行;CountDownLatch计数器只能使用一次,CyclicBarrier可以使用reset方法重置用以处理某些复杂的业务场景。

12-27 14:36