[Java知识库]Java并发-同步器原理

CountDownLatch

在我们实际做开发的时候可能会遇到这样的需求，主线程需要等待所有子线程完成任务之后再执行操作，我们可以用join来实现这个功能，但是join不够灵活，主线程只能等子线程完全执行完毕才能从join返回，而不能在子线程某个位置就执行返回。
为了解决这个问题，JDK开发组提供了CountDownLatch这个类，这样的需求通过CountDownLatch来实现会更加优雅、灵活。

CountDownLatch的使用

public static final CountDownLatch countDownLatch = new CountDownLatch(2);

public static void main(String[] args) throws InterruptedException {
    Thread threadOne = new Thread(new Runnable() {
        @Override
        public void run() {
            System.out.println("threadOne start");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("threadOne end");
            countDownLatch.countDown();
        }
    });

    Thread threadTwo = new Thread(new Runnable() {
        @Override
        public void run() {
            System.out.println("threadTwo start");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("threadTwo end");
            countDownLatch.countDown();
        }
    });
    threadOne.start();
    threadTwo.start();
    countDownLatch.await();
    System.out.println("main end");
}

运行结果：

threadOne start
threadTwo start
threadOne end
threadTwo end
main end

原理解析

基于AQS

CountDownLatch有一个Sync静态内部类继承了AQS，我们调用CountDownLatch的构造函数的时候会生成这个Sync的一个实例对象，我们设置的参数count其实就是设置AQS的state属性。

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

private static final class Sync extends AbstractQueuedSynchronizer {
    private static final long serialVersionUID = 4982264981922014374L;

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

    int getCount() {
        return getState();
    }

    // 执行CountDownLatch的await方法最后会跳转到这个方法，state为0返回1，不为0就返回-1。
    protected int tryAcquireShared(int acquires) {
        return (getState() == 0) ? 1 : -1;
    }

    // 执行CountDownLatch的countDown方法最终会跳转到这个方法，该方法通过CAS + 自旋的方式对count进行修改，如果cas成功并且state == 0，就会返回true。
    protected boolean tryReleaseShared(int releases) {
        for (;;) {
            int c = getState();
            // 如果修改之前的state已经为0了，直接返回，这里可以防止countDown方法一直调用，导致state变为负数，state最低为0。
            if (c == 0)
                return false;
            int nextc = c-1;
            // CAS修改state，让state减一，如果减完之后state为0，就返回true。
            if (compareAndSetState(c, nextc))
                return nextc == 0;
        }
    }
}

await

调用await方法的线程，如果在CountDownLatch的count不为0的时候会阻塞，知道count为0的时候才会被唤醒。

public void await() throws InterruptedException {
    // 调用AQS的acquireSharedInterruptibly方法
    sync.acquireSharedInterruptibly(1);
}

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    // 调用Sync的tryAcquireShared方法，该方法根据state是否为0，会返回1，或者-1
    // 返回1时，直接返回方法，执行任何操作
    // 返回-1时，表示state不为0，那么就把当前线程加到AQS队列中，同时阻塞
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

countDown

调用countDown方法，CountDownLatch的count就会减一，当count为0的时候就会去唤醒所有调用了await阻塞的线程。

public void countDown() {
    // 调用AQS的releaseShared方法
    sync.releaseShared(1);
}

public final boolean releaseShared(int arg) {
    // 调用Sync的tryReleaseShared方法，如果为true，表示最后一个线程调用countDown之后，count为0了，这时候要唤醒所有调用了await被阻塞的线程。
    if (tryReleaseShared(arg)) {
        // 唤醒队列中所有阻塞的线程
        doReleaseShared();
        return true;
    }
    return false;
}

private void doReleaseShared() {
    for (;;) {
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            if (ws == Node.SIGNAL) {
                // 设置waitStatus
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;
                // 唤醒线程
                unparkSuccessor(h);
            }
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;
        }
        if (h == head)
            break;
    }
}

CountDownLatch的计数器是一次性的，也就是等到计数器值变为0后，再调用CountDownLatch的await和 countdown方法都会立刻返回。

CyclicBarrier

CyclicBarrier是回环屏障的意思，它可以让一组线程全部达到一个状态后再全部同时执行。之所以叫作屏障是因为线程调用await方法后就会被阻塞，这个阻塞点就称为屏障点，等所有线程都调用了await方法后，线程们就会冲破屏障，继续向下运行。同时CyclicBarrier的计数器是可以重置。

CyclicBarrier的使用

线程在调用了CyclicBarrier的await，如果当前没有足够的线程调用了await，就会阻塞直到有足够的线程调用了await才会返回，并且会重置计数器。

public static final CyclicBarrier cyclicBarrier = new CyclicBarrier(2);

public static void main(String[] args) throws InterruptedException {
    Thread threadOne = new Thread(new Runnable() {
        @Override
        public void run() {
            try {
                System.out.println("threadOne step1");
                cyclicBarrier.await();
                System.out.println("threadOne step2");
                cyclicBarrier.await();
                System.out.println("threadOne step3");
                cyclicBarrier.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    });

    Thread threadTwo = new Thread(new Runnable() {
        @Override
        public void run() {
            try {
                System.out.println("threadTwo step1");
                cyclicBarrier.await();
                System.out.println("threadTwo step2");
                cyclicBarrier.await();
                System.out.println("threadTwo step3");
                cyclicBarrier.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    });
    threadOne.start();
    threadTwo.start();

    threadOne.join();
    threadTwo.join();

    System.out.println("main end");
}

运行结果如下：

threadOne step1
threadTwo step1
threadTwo step2
threadOne step2
threadOne step3
threadTwo step3
main end

CyclicBarrier的原理

基于锁的内部原理

CyclicBarrier有以下的属性，其内部原理是通过锁来完成的。

// 线程await的时候要获得这个锁才可以执行操作
private final ReentrantLock lock = new ReentrantLock();
// 线程await，如果没有足够的线程到达屏障就会通过Condition来阻塞
private final Condition trip = lock.newCondition();
// 初始化的时候parties和count是一样的，等count为0的时候，CyclicBarrier会利用parties来重置count（计数器）。
private final int parties;
// 计数器，每当有线程调用await的时候，count就会减一，当count为0的时候就会唤醒之前阻塞的线程。
private int count;
// 构造CyclicBarrier的时候可以传入一个barrierCommand，在count为0的时候就会调用里面的run方法。
private final Runnable barrierCommand;
// Generation里面有一个boolean类型的broke，broke为true的时候就可以中断CyclicBarrier。
private Generation generation = new Generation();

CyclicBarrier的构造函数

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

// parties和count是相等的
public CyclicBarrier(int parties, Runnable barrierAction) {
    if (parties <= 0) throw new IllegalArgumentException();
    this.parties = parties;
    this.count = parties;
    this.barrierCommand = barrierAction;
}

dowait

无论是await()还是await(long timeout, TimeUnit unit)最终都会调用到dowait方法。

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

public int await(long timeout, TimeUnit unit)
    throws InterruptedException,
           BrokenBarrierException,
           TimeoutException {
    return dowait(true, unit.toNanos(timeout));
}

// 参数timed为false时，nacos就没用了
// 参数timed为true时，nacos是过期的时间
private int dowait(boolean timed, long nanos)
    throws InterruptedException, BrokenBarrierException,
           TimeoutException {
    final ReentrantLock lock = this.lock;
    // 加锁
    lock.lock();
    try {
        // 判断Generation的broke是否为true，如果是true就抛出异常
        final Generation g = generation;

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

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

        // count减一
        int index = --count;
        // 如果count为0，会执行下面的逻辑
        if (index == 0) {
            boolean ranAction = false;
            try {
                // 调用barrierCommand的run方法，这个是我们在构造函数中传入的
                final Runnable command = barrierCommand;
                if (command != null)
                    command.run();
                ranAction = true;
                // 这里面会唤醒所有阻塞的线程，同时重置计数器
                nextGeneration();
                return 0;
            } finally {
                if (!ranAction)
                    breakBarrier();
            }
        }

        // 减完之后，count不为0就会执行下面的逻辑
        for (;;) {
            try {
                // 如果不是带超时时间的就直接用condition的await阻塞
                if (!timed)
                    trip.await();
                else if (nanos > 0L)
                    // 带超时时间的用condition的awaitNanos阻塞
                    nanos = trip.awaitNanos(nanos);
            } catch (InterruptedException ie) {
                if (g == generation && ! g.broken) {
                    breakBarrier();
                    throw ie;
                } else {
                    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();
    }
}

private void nextGeneration() {
    // 唤醒所有通过Condition阻塞的线程
    trip.signalAll();
    // 重置计数器
    count = parties;
    generation = new Generation();
}

Semaphore

Semaphore是线程同步的辅助类，可以控制当前访问自身的线程个数,并提供了同步机制。使用Semaphore可以控制同时访问资源的线程个数，例如通过 Semaphore 限流。

Semaphore的使用

private final static int POOL_SIZE = 10;

private final Semaphore useful,useless;//useful表示可用的数据库连接，useless表示已用的数据库连接

public DBPoolSemaphore() {
    this. useful = new Semaphore(POOL_SIZE);
    this.useless = new Semaphore(0);
}

//存放数据库连接的容器
private static LinkedList<Connection> pool = new LinkedList<Connection>();
//初始化池
static {
    for (int i = 0; i < POOL_SIZE; i++) {
        pool.addLast(SqlConnectImpl.fetchConnection());
    }
}

/*归还连接*/
public void returnConnect(Connection connection) throws InterruptedException {
    if(connection!=null) {
        System.out.println("当前有"+useful.getQueueLength()+"个线程等待数据库连接！！"
                +"可用连接数:"+useful.availablePermits());
        useless.acquire();
        synchronized (pool) {
            pool.addLast(connection);
        }
        useful.release();
    }
}

/*从池子拿连接*/
public Connection takeConnect() throws InterruptedException {
    useful.acquire();
    Connection conn;
    synchronized (pool) {
        conn = pool.removeFirst();
    }
    useless.release();
    return conn;
}