J.U.C并发包（1）


AbstractQueuedSynchronizer

AbstractQueuedSynchronizer是JUC并发包中锁的底层支持，AbstractQueuedSynchronizer是抽象同步队列，简称AQS，是实现同步器的基础组件，并发包中锁的实现底层就是使用AQS实现，


  • 从类图的关系可以看到AQS是一个FIFO的双向队列，内部通过节点head 和 tail 记录队首和队尾元素，队列元素类型为Node。其中Node中Thread变量用来存放进入AQS队列里面的线程

  • Node 节点内部SHARED用来标记该线程是获取共享资源时候被阻塞挂起来后放入AQS队列，

  • EXCLUSIVE标记线程是获取独占资源时候被挂起后放入AQS队列;

  • waitStatus记录当前线程等待状态，分别为CANCELLED(线程被取消了)，SIGNAL(线程需要被唤醒)，CONDITION(线程在条件队列里面等待)，PROPAGATE(释放共享资源时候需要通知其他节点）；

  • AQS中维持了一个单一的状态信息state，可以通过getState,setState,compareAndSetState 函数修改其值；对于ReentrantLock 的实现来说，state 可以用来表示当前线程获取锁的可重入次数；

  • pre记录当前节点的前驱节点，next记录当前节点后继节点

  • 调用acquire(int arg)方法获取独占资源，调用release(int arg)方法释放资源；

具体思路：

  • 当多个线程同时调用 lock.lock() 获取锁的时候，同时只有一个线程获取到了该锁，其他线程会被转换为 Node 节点插入到 lock 锁对应的 AQS 阻塞队列里面，并做自旋 CAS 尝试获取锁，前提是head的直接后继；

  • 如果获取到锁的线程又调用了对应的条件变量的 await() 方法，则该线程会释放获取到的锁，并被转换为 Node 节点插入到条件变量对应的条件队列里面；

  • 这时候因为调用 lock.lock() 方法被阻塞到 AQS 队列里面的一个线程会获取到被释放的锁，如果该线程也调用了条件变量的 await（）方法则该线程也会被放入条件变量的条件队列；

  • 当另外一个线程调用了条件变量的 signal() 或者 signalAll() 方法时候，会把条件队列里面的一个或者全部 Node 节点移动到 AQS 的阻塞队列里面，等待时机获取锁。

CountDownLatch

他是一个同步辅助类，可以实现类似阻塞当前线程的功能，使用了给定的计数器进行初始化，该计数器操作是原子操作，同一时刻只能有一个线程操作该计数器。


如上图中，TA线程由于await()方法被阻塞，除非前面的线程调用countDown()方法，当计数器为0，TA就可以继续往下执行。计数器不可重置。 

        private final static int threadCount = 200;

    public static void main(String[] args) throws Exception {

        ExecutorService exec = Executors.newCachedThreadPool();

        final CountDownLatch countDownLatch = new CountDownLatch(threadCount);

        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    test(threadNum);
                } catch (Exception e) {
                    log.error("exception", e);
                } finally {
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await();
        log.info("finish");
        exec.shutdown();
    }

    private static void test(int threadNum) throws Exception {
        Thread.sleep(100);
        log.info("{}", threadNum);
        Thread.sleep(100);
    }

Semaphore

他可以控制某个资源可以被多少个线程同时访问，使用Semaphore管理必须要先获取一个许可，执行完毕后释放一个许可，后面的线程才能继续访问，代码演示： 

@Slf4j
public class SemaphoreExample1 {

    private final static int threadCount = 20;

    public static void main(String[] args) throws Exception {

        ExecutorService exec = Executors.newCachedThreadPool();

        final Semaphore semaphore = new Semaphore(3);

        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    semaphore.acquire(); // 获取一个许可
                    test(threadNum);
                    semaphore.release(); // 释放一个许可
                } catch (Exception e) {
                    log.error("exception", e);
                }
            });
        }
        exec.shutdown();
    }

    private static void test(int threadNum) throws Exception {
        log.info("{}", threadNum);
        Thread.sleep(1000);
    }
}

CyclicBarrier

他是一个同步辅助类，允许一组线程相互等待，直到到达某个公共的屏障点，commonBarrierPoint，CyclicBarrier也具有一个计数器，当计数器达到设置的值，被await()方法阻塞的值会被唤醒，继续执行后续的操作，计数器可以被重置，适合并发情况下需要合并计算的场景。

演示代码如下： 

@Slf4j
public class CyclicBarrierExample1 {
    private static Logger log = LoggerFactory.getLogger(CyclicBarrierExample1.class);

    private static CyclicBarrier barrier = new CyclicBarrier(5);

    public static void main(String[] args) throws Exception {

        ExecutorService executor = Executors.newCachedThreadPool();

        for (int i = 0; i < 10; i++) {
            final int threadNum = i;
            Thread.sleep(1000);
            executor.execute(() -> {
                try {
                    race(threadNum);
                } catch (Exception e) {
                    log.error("exception", e);
                }
            });
        }
        executor.shutdown();
    }

    private static void race(int threadNum) throws Exception {
        Thread.sleep(1000);
        log.info("{} is ready", threadNum);
        barrier.await();
        log.info("{} continue", threadNum);
    }
}

console：



ReentrantLock

ReentrantLock（可重入锁）和Synchronize锁的区别

  • 可重入性：二者都是重入锁

  • 锁的实现：ReentrantLock通过jdk实现，synchronize是通过jvm实现，但注意ReentrantLock需要释放锁，而synchronize不需要释放锁，由jvm管理，也就是说synchronize不会产生死锁，而ReentrantLock可能产生死锁。

  • 性能的区别：synchronize未做优化前，ReentrantLock优于synchronize，但synchronize在经过偏向锁，轻量级锁优化后性能就差不多了

  • 功能区别：代码简洁synchronize优于ReentrantLock，锁的细粒度和灵活度ReentrantLock更好。
ReentrantLock独有功能

  • 可指定公平锁还是非公平锁：公平锁（先等待的就先获取锁）；
  • 提供了condition类，可以分组唤醒需要唤醒的线程；
  • 提供锁的打断机制，lock.lockInterruptibly()。
ReentrantLock是一种自选锁，内部循环使用CAS操作实现加锁

ReentrantReadWriteLock

ReentrantReadWriteLock是读写锁，维护了一对锁，一个读锁，一个写锁，通过实现ReadWriteLock接口实现了readLock()方法和writeLock()方法。适用于多线程情况下的读写操作，但是要注意如果读操作过于频繁可能会导致写锁饥饿。

StampLock

StampLock有三种控制锁的方式：写，读和乐观读，StampLock会生成票据。

  • 乐观读：乐观的认为写入和读取同时发生的概率很少，因此不悲观的使用读取锁定，程序可以查看读取数据时候遭到写入执行的变更之后，大幅提升程序的性能。

  • 乐观锁和悲观锁实例如下： 

public class LockExample4 {

    class Point {
        private double x, y;
        private final StampedLock sl = new StampedLock();

        void move(double deltaX, double deltaY) { // an exclusively locked method
            long stamp = sl.writeLock();
            try {
                x += deltaX;
                y += deltaY;
            } finally {
                sl.unlockWrite(stamp);
            }
        }

        //下面看看乐观读锁案例
        double distanceFromOrigin() { // A read-only method
            long stamp = sl.tryOptimisticRead(); //获得一个乐观读锁
            double currentX = x, currentY = y;  //将两个字段读入本地局部变量
            if (!sl.validate(stamp)) { //检查发出乐观读锁后同时是否有其他写锁发生？
                stamp = sl.readLock();  //如果没有，我们再次获得一个读悲观锁
                try {
                    currentX = x; // 将两个字段读入本地局部变量
                    currentY = y; // 将两个字段读入本地局部变量
                } finally {
                    sl.unlockRead(stamp);
                }
            }
            return Math.sqrt(currentX * currentX + currentY * currentY);
        }

        //下面是悲观读锁案例
        void moveIfAtOrigin(double newX, double newY) { // upgrade
            // Could instead start with optimistic, not read mode
            long stamp = sl.readLock();
            try {
                while (x == 0.0 && y == 0.0) { //循环，检查当前状态是否符合
                    long ws = sl.tryConvertToWriteLock(stamp); //将读锁转为写锁
                    if (ws != 0L) { //这是确认转为写锁是否成功
                        stamp = ws; //如果成功 替换票据
                        x = newX; //进行状态改变
                        y = newY;  //进行状态改变
                        break;
                    } else { //如果不能成功转换为写锁
                        sl.unlockRead(stamp);  //我们显式释放读锁
                        stamp = sl.writeLock();  //显式直接进行写锁 然后再通过循环再试
                    }
                }
            } finally {
                sl.unlock(stamp); //释放读锁或写锁
            }
        }
    }
}