Java并发包APIs总结

Posted 2021-07-20 Shi Peng

一、前言

Java5开始提供了java.util.concurrent，里面包含了绝大多数编写Java高并发代码的技巧和套路，所以不单要会用API，还需要理解concurrent包里每个API的设计方法和目的，这样在自己编写高并发程序时，才能够在需要时采用合适的技巧来解决问题。

二、API接口及核心类

2.1、java.util.concurrent下的接口

BlockingDeque
BlockingQueue
Callable
CompletableFuture.AsynchronousCompletionTask
CompletionService
CompletionStage
ConcurrentMap
ConcurrentNavigableMap
Delayed
Executor
ExecutorService
ForkJoinPool.ForkJoinWorkerThreadFactory
ForkJoinPool.ManagedBlocker
Future
RejectedExecutionHandler
RunnableFuture
RunnableScheduledFuture
ScheduledExecutorService
ScheduledFuture
ThreadFactory
TransferQueue

2.2、java.util.concurrent下的类

AbstractExecutorService
ArrayBlockingQueue
CompletableFuture
ConcurrentHashMap
ConcurrentHashMap.KeySetView
ConcurrentLinkedDeque
ConcurrentLinkedQueue
ConcurrentSkipListMap
ConcurrentSkipListSet
CopyOnWriteArrayList
CopyOnWriteArraySet
CountDownLatch
CountedCompleter
CyclicBarrier
DelayQueue
Exchanger
ExecutorCompletionService
Executors
ForkJoinPool
ForkJoinTask
ForkJoinWorkerThread
FutureTask
LinkedBlockingDeque
LinkedBlockingQueue
LinkedTransferQueue
Phaser
PriorityBlockingQueue
RecursiveAction
RecursiveTask
ScheduledThreadPoolExecutor
Semaphore
SynchronousQueue
ThreadLocalRandom
ThreadPoolExecutor
ThreadPoolExecutor.AbortPolicy
ThreadPoolExecutor.CallerRunsPolicy
ThreadPoolExecutor.DiscardOldestPolicy
ThreadPoolExecutor.DiscardPolicy

三、核心API分析

3.1、集合：Map、List、Array

3.1.1、ConcurrentHashMap

ConcurrentHashMap比非线程安全的HashMap好在线程安全，比线程安全的Hashtable性能更好，但ConcurrentHashMap降低了对读一致性的要求。

JDK1.7：分段锁，分成多个segment, 每个segment是一个HashMap，且segment继承了ReentrantLock。
JDK1.8：放弃了分段锁，使用Node, 再次降低锁粒度。

总结：
1、JDK1.7通过使用Segment来减少锁粒度，一个ConcurrentHashMap被切分成多个Segment, 在put时，需要锁定该segment；在get时，使用volatile来保证可见性；当要全局统计size时，可能会多次尝试计算modcount值，如果每次都相等返回，如果不相等，仍要锁定每个Segment来计算。

2、JDK1.7的ConcurrentHashMap, 当长度过长时，碰撞频繁，性能下降。所以JDK1.8重构了ConcurrentHashMap，改进如下：
1）放弃Segment, 采用Node, 把锁粒度从Segment降低到Node
2）resize时，如果线程1在执行put, 则通过线程2帮助resize
3）采用CAS操作来确保Node的操作的原子性，从而替代了锁。
4）不在用ReentrantLock, 而是直接用synchronized，应该是synchronized做了足够多的优化。

详见：https://blog.csdn.net/shijinghan1126/article/details/86500559

3.1.2、LinkedBlockingQueue

LinkedBlockingQueue是线程安全的阻塞队列，用长度有最大限制的双向链表实现，常用于生产者、消费者模型。

LinkedBlockingQueue的锁：读和写，由2把锁分别控制，分别为读锁和写锁，每把锁都管理head节点和last节点

LinkedBlockingQueue通过对读写锁分离的方式，提升并发吞吐量。

详见：https://blog.csdn.net/shijinghan1126/article/details/86500816

3.1.3、CopyOnWriteArrayList

copy-on-write策略：多线程共享一个资源，当需要修改时，把资源全量copy到另一个地方然后修改，修改完再通过指针切换，指向新的地址。这是一种延迟懒惰策略来提升吞吐量，也是一种读写分离的思想。读时不需要加锁，但在写时需要加锁，以避免copy出多个副本，多个线程写时要排队。

应用场景：读多写少的场景。

缺点：废内存、数据一致性。

详见：https://blog.csdn.net/shijinghan1126/article/details/86500663

3.2、锁

3.2.1、ReentrantReadWriteLock

首先，ReentrantReadWriteLock是个读写锁：

• 一个线程获取了读锁：其他线程可并发读，但不可写
• 一个线程获取了写锁：其他线程不可读，也不可写

其次，它是ReentrantLock, 即可重入锁，这个怎样理解呢？
1、在WriteLock的内部，可获取ReadLock, 但在ReadLock内部不能获取WriteLock
2、WriteLock可降级为ReadLock，即可先在WriteLock中获取ReadLock, 然后再释放WriteLock
3、WriteLock和ReadLock都支持Interrupt
4、WriteLock支持Condition, 而ReadLock不支持Condition

什么叫可重入？
即一个线程可以重复加锁，可以对一个锁加多次，每次释放的时候释放一层，直到该线程加锁的次数为0，这个线程的锁才算释放。

详见：https://blog.csdn.net/shijinghan1126/article/details/86500663

3.3.2、

3.3、线程

3.3.1、ThreadLocal：线程局部变量

ThreadLocal是一个容器，用于存放一个线程的局部变量，并不是指“本地线程”。所以叫ThreadLocalVariable更准确。

所以，ThreadLocal可以理解为线程为单位的容器。

详见：https://blog.csdn.net/shijinghan1126/article/details/86500698

3.3.2、CountDownLatch：当前线程等待指定个数的线程执行完后再执行

CountDownLatch使一个线程等待其他线程各自执行完后再执行。CountDownLatch是通过一个计数器来实现的，计数器的初始值就是线程的数量，每执行完一个线程，计数器的值就减1，当计数器的值为0时，等待的线程就可以执行了。

基于AQS共享模式实现

应用场景：让预先执行的线程都执行完毕后，再开始执行正式的线程，如秒杀

详见：https://www.jianshu.com/p/e233bb37d2e6

附录：AQS（Abstract Queued Synchronizer）

AQS是抽象的队列同步器，AQS定义了一套多线程访问共享资源的同步器框架，许多同步类都依赖于它，如常用的ReentrantLock, Semaphore, CountDownLatch。

AQS维护了一个volatile state（共享资源）和一个FIFO的等待队列（多线程资源被阻塞时，会进入此队列）。

AQS定义了两种资源共享方式：Exclusive（独占，只有一个线程能执行，如ReentrantLock）和Share（共享，多个线程可同时执行，如Semaphore, CountDownLatch）。

以ReentrantLock为例，state初始化为0，表示未锁定状态。A线程lock()时，会调用tryAcquire()独占并将state加1，此后，其他线程再tryAcquire()时就会失败，直到A线程unlock到state为0，其他线程才有机会获得锁。当然，在释放锁之前，A线程自己可以重复获取锁（state会累加），这就是可重入的概念。

再以CountDownLatch为例，任务分为n个子线程去执行，state也初始化为n（n为子线程个数），每个子线程执行完后，countDown()一次，state会CAS减1，等到所有的子线程都执行完毕，即state=0, 会主线程会从await()函数返回，继续后续动作。

详见：https://www.cnblogs.com/waterystone/p/4920797.html

3.3.3、CyclicBarrier：所有线程执行完后执行

CyclicBarrier，字面意思是 “循环栅栏”。作用是等待所有线程都执行完，在继续执行。

打比喻的话，就好比桌上人都到齐了再开始吃饭，旅游团成员到齐后再出发一样。

CyclicBarrier是基于Condition实现：CyclicBarrier类内部也有一个计数器，每个线程到达栅栏时都会因调用await将自己阻塞，此时计算器会减1，当计数器为0时，所有因await方法而被阻塞的线程将被唤醒

应用场景：多线程计算数据，最后合并计算结果。

详见：https://www.jianshu.com/p/333fd8faa56e

CyclicBarrier和CountDownLatch的区别

1、CountDownLatch是一个计数器，线程完成一个减1，只能用1次
2、CyclicBarrier的计数器更像一个阀门，需要等所有线程都到达，才能开门放行，计数器是递增的，可以多次使用。

3.3.4、Condition接口下的await(), signal()：线程间协作

Condition是Java1.5开始出现，用来替代传统的Object的wait(), notify()。使用Condition的await(), signal() 方式，线程间协作会更加安全而高效。阻塞队列实际上就是使用condition来模拟线程间协作。

Condition是个接口，基本方法就是await()和notify()这两个方法。
Condition依赖于Lock接口，生成一个condition的基本代码是lock.newCondition()。
调用condition的await()和signal()方法，都在lock的保护之内，也就是必须在lock(), unlock()之间才能使用。

Condition的await()对应Object的wait()
Condition的signal()对应Object的notify();
Condition的signalAll()对应Object的notifyAll()

Condition可用来实现生产者、消费者：

import java.util.LinkedList;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class BoundedQueue {

    private LinkedList<Object> buffer;    //生产者容器
    private int maxSize ;           //容器最大值是多少
    private Lock lock;
    private Condition productCondition; // 生产condition
    private Condition consumeCondition; // 消费condition
    BoundedQueue(int maxSize){
        this.maxSize = maxSize;
        buffer = new LinkedList<Object>();
        lock = new ReentrantLock();
        productCondition = lock.newCondition();
        consumeCondition = lock.newCondition();
    }

    /**
     * 生产者
     * @param obj
     * @throws InterruptedException
     */
    public void put(Object obj) throws InterruptedException {
        lock.lock();    //获取锁
        try {
            while (maxSize == buffer.size()){
                productCondition.await();       //满了，生产的线程进入等待状态
            }
            buffer.add(obj);
            consumeCondition.signal(); //通知, 唤醒消费线程
        } finally {
            lock.unlock();
        }
    }

    /**
     * 消费者
     * @return
     * @throws InterruptedException
     */
    public Object get() throws InterruptedException {
        Object obj;
        lock.lock();
        try {
            while (buffer.size() == 0){ //队列中没有数据了 线程进入等待状态
                consumeCondition.await(); // 消费线程等待
            }
            obj = buffer.poll();
            productCondition.signal(); //唤醒生产线程
        } finally {
            lock.unlock();
        }
        return obj;
    }

}

Condition可通俗理解为条件队列。当一个线程调用了await() 方法后，当前线程会释放锁，并在此等待，而其他线程调用Condition对象的signal() 方法，会通知当前线程，当前线程才会从await()方法返回，并在返回前已获取到锁。

详见：https://www.cnblogs.com/gemine/p/9039012.html

3.3.5、Semaphore：信号量

Semaphore，通常叫他信号量，可以用来控制同时访问特定资源的线程个数，通过协调各个线程，以保障合理的使用资源。

Semaphore可以比喻成停车场入口的显示屏，每辆车进入显示屏都会显示剩余车位减1，每有一辆车出去，显示屏上就会把剩余车辆加1，当显示屏上剩余车位为0，停车场入口的栏杆就不会打开，车辆就无法进入了，知道有一辆车从停车场出去。

使用场景：数据库连接池中的线程数限制

常用方法：

acquire()  
获取一个令牌，在获取到令牌、或者被其他线程调用中断之前线程一直处于阻塞状态。
​
acquire(int permits)  
获取一个令牌，在获取到令牌、或者被其他线程调用中断、或超时之前线程一直处于阻塞状态。
    
acquireUninterruptibly() 
获取一个令牌，在获取到令牌之前线程一直处于阻塞状态（忽略中断）。
    
tryAcquire()
尝试获得令牌，返回获取令牌成功或失败，不阻塞线程。
​
tryAcquire(long timeout, TimeUnit unit)
尝试获得令牌，在超时时间内循环尝试获取，直到尝试获取成功或超时返回，不阻塞线程。
​
release()
释放一个令牌，唤醒一个获取令牌不成功的阻塞线程。
​
hasQueuedThreads()
等待队列里是否还存在等待线程。
​
getQueueLength()
获取等待队列里阻塞的线程数。
​
drainPermits()
清空令牌把可用令牌数置为0，返回清空令牌的数量。
​
availablePermits()
返回可用的令牌数量。

使用示例：用semaphore 实现停车场提示牌功能

public class TestCar {
​
    //停车场同时容纳的车辆10
    private  static  Semaphore semaphore=new Semaphore(10);
​
    public static void main(String[] args) {
​
        //模拟100辆车进入停车场
        for(int i=0;i<100;i++){
​
            Thread thread=new Thread(new Runnable() {
                public void run() {
                    try {
                        System.out.println("===="+Thread.currentThread().getName()+"来到停车场");
                        if(semaphore.availablePermits()==0){
                            System.out.println("车位不足，请耐心等待");
                        }
                        semaphore.acquire();//获取令牌尝试进入停车场
                        System.out.println(Thread.currentThread().getName()+"成功进入停车场");
                        Thread.sleep(new Random().nextInt(10000));//模拟车辆在停车场停留的时间
                        System.out.println(Thread.currentThread().getName()+"驶出停车场");
                        semaphore.release();//释放令牌，腾出停车场车位
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            },i+"号车");
​
            thread.start();
​
        }
​
    }
}

Semaphore的实现原理是AQS

详见：https://zhuanlan.zhihu.com/p/98593407