java-concurrent包

通常所说的concurrent包基本有3个package组成  

• java.util.concurrent：提供大部分关于并发的接口和类，如BlockingQueue,Callable,ConcurrentHashMap,ExecutorService, Semaphore等  
• java.util.concurrent.atomic：提供所有原子操作的类， 如AtomicInteger, AtomicLong等；  
• java.util.concurrent.locks:提供锁相关的类, 如Lock, ReentrantLock, ReadWriteLock, Condition等；  

1. java.util.concurrent并发的接口和类

1.1 ConcurrentHashMap

不管是HashMap还是HashTable要想线程安全都要用synchronized，而synchronized是针对整张Hash表的，即每次锁住整张表让线程独占，ConcurrentHashMap允许多个修改操作并发进行，其关键在于使用了锁分离技术。ConcurrentHashMap内部使用段(Segment)来表示这些不同的部分，每个段其实就是一个小的hash table，它们有自己的锁。只要多个修改操作发生在不同的段上，它们就可以并发进行

• 删除操作

要将删除节点e前面的结点复制一遍，尾结点指向e的下一个结点。e后面的结点不需要复制，它们可以重用。这是由entry的定义决定的，entry除了value，其他所有属性都是用final来修饰的，这意味着在第一次设置了next域之后便不能再改变它，取而代之的是将它之前的节点全都克隆一次

1.2 ConcurrentLinkedQueue

使用CAS(Compare and Swap)非阻塞算法来取得最大的吞吐量

入队注解代码如下：

public boolean offer(E e) {
    if (e == null) throw new NullPointerException();
    //入队前，创建一个入队节点
    Node<E> n = new Node<E>(e);
    retry:
    //死循环，入队不成功反复入队。
    for (;;) {
        //创建一个指向tail节点的引用
        Node<E> t = tail;
        //p用来表示队列的尾节点，默认情况下等于tail节点。
        Node<E> p = t;
        for (int hops = 0; ; hops++) {
        //获得p节点的下一个节点。
            Node<E> next = succ(p);
        //next节点不为空，说明p不是尾节点，需要更新p后在将它指向next节点
            if (next != null) {
               //循环了两次及其以上，并且当前节点还是不等于尾节点
                if (hops > HOPS && t != tail)
                    continue retry; 
                p = next;
            } 
            //如果p是尾节点，则设置p节点的next节点为入队节点。
            else if (p.casNext(null, n)) {
              //如果tail节点有大于等于1个next节点，则将入队节点设置成tair节点，更新失败了也
没关系，因为失败了表示有其他线程成功更新了tair节点。
                if (hops >= HOPS)
                    casTail(t, n); // 更新tail节点，允许失败
                return true;  
            } 
           // p有next节点,表示p的next节点是尾节点，则重新设置p节点
            else {
                p = succ(p);
            }
        }
    }
}

出队注解代码如下：

public E poll() {
    Node<E> h = head;
   // p表示头节点，需要出队的节点
    Node<E> p = h;
    for (int hops = 0;; hops++) {
        // 获取p节点的元素
        E item = p.getItem();
        // 如果p节点的元素不为空，使用CAS设置p节点引用的元素为null,如果成功则返回p节点的元素。
        if (item != null && p.casItem(item, null)) {
            if (hops >= HOPS) {
                //将p节点下一个节点设置成head节点
                Node<E> q = p.getNext();
                updateHead(h, (q != null) ? q : p);
            }
            return item;
        }
        // 如果头节点的元素为空或头节点发生了变化，这说明头节点已经被另外一个线程修改了。那么获取p节点的下一个节点 
        Node<> next = succ(p);
        // 如果p的下一个节点也为空，说明这个队列已经空了
        if (next == null) {
          // 更新头节点。
            updateHead(h, p);
            break;
        }
        // 如果下一个元素不为空，则将头节点的下一个节点设置成头节点
        p = next;
    }
    return null;
}

1.2. CyclicBarrier

一个同步辅助类，它允许一组线程互相等待，直到到达某个公共屏障点 (common barrier point),再全部同时执行。在涉及一组固定大小的线程的程序中，这些线程必须不时地互相等待，此时 CyclicBarrier 很有用。因为该 barrier 在释放等待线程后可以重用，所以称它为循环的 barrier

常用方法

public int await() throws InterruptedException,BrokenBarrierException

在所有参与者都已经在此 barrier 上调用 await方法之前，将一直等待。如果当前线程不是将到达的最后一个线程，出于调度目的，将禁用它，且在发生以下情况之一前，该线程将一直处于休眠状态：

• 最后一个线程到达；或者
• 其他某个线程中断当前线程；或者
• 其他某个线程中断另一个等待线程；或者
• 其他某个线程在等待 barrier 时超时；或者
• 其他某个线程在此 barrier 上调用 reset()。

如果当前线程：

• 在进入此方法时已经设置了该线程的中断状态；或者
• 在等待时被中断

则抛出 InterruptedException，并且清除当前线程的已中断状态。如果在线程处于等待状态时 barrier 被 reset()，或者在调用 await 时 barrier 被损坏，抑或任意一个线程正处于等待状态，则抛出 BrokenBarrierException 异常。

如果任何线程在等待时被 中断，则其他所有等待线程都将抛出 BrokenBarrierException 异常，并将 barrier 置于损坏状态。

如果当前线程是最后一个将要到达的线程，并且构造方法中提供了一个非空的屏障操作，则在允许其他线程继续运行之前，当前线程将运行该操作。如果在执行屏障操作过程中发生异常，则该异常将传播到当前线程中，并将 barrier 置于损坏状态。

返回：

到达的当前线程的索引，其中，索引 getParties() - 1 指示将到达的第一个线程，零指示最后一个到达的线程

抛出：

InterruptedException - 如果当前线程在等待时被中断

BrokenBarrierException - 如果另一个 线程在当前线程等待时被中断或超时，或者重置了 barrier，或者在调用 await 时 barrier 被损坏，抑或由于异常而导致屏障操作（如果存在）失败。

1.3. CountDownLatch  

利用它可以实现类似计数器的功能,不能重复使用，一般用于某个线程A等待若干个其他线程执行完任务之后，它才执行

示例：

public class CountDownLatchTest {

    // 模拟了100米赛跑，10名选手已经准备就绪，只等裁判一声令下。当所有人都到达终点时，比赛结束。
    public static void main(String[] args) throws InterruptedException {

        // 开始的倒数锁 
        final CountDownLatch begin = new CountDownLatch(1);  

        // 结束的倒数锁 
        final CountDownLatch end = new CountDownLatch(10);  

        // 十名选手 
        final ExecutorService exec = Executors.newFixedThreadPool(10);  

        for (int index = 0; index < 10; index++) {
            final int NO = index + 1;  
            Runnable run = new Runnable() {
                public void run() {  
                    try {  
                        // 如果当前计数为零，则此方法立即返回。
                        // 等待
                        begin.await();  
                        Thread.sleep((long) (Math.random() * 10000));  
                        System.out.println("No." + NO + " arrived");  
                    } catch (InterruptedException e) {  
                    } finally {  
                        // 每个选手到达终点时，end就减一
                        end.countDown();
                    }  
                }  
            };  
            exec.submit(run);
        }  
        System.out.println("Game Start");  
        // begin减一，开始游戏
        begin.countDown();  
        // 等待end变为0，即所有选手到达终点
        end.await();  
        System.out.println("Game Over");  
        exec.shutdown();  
    }
}

1.4. Semaphore信号量

一般用于控制对某组资源的访问权限

假若一个工厂有5台机器，但是有8个工人，一台机器同时只能被一个工人使用，只有使用完了，其他工人才能继续使用。Semaphore来实现示例：

public class Test {
    public static void main(String[] args) {
        int N = 8;            //工人数
        Semaphore semaphore = new Semaphore(5); //机器数目
        for(int i=0;i<N;i++)
            new Worker(i,semaphore).start();
    }
     
    static class Worker extends Thread{
        private int num;
        private Semaphore semaphore;
        public Worker(int num,Semaphore semaphore){
            this.num = num;
            this.semaphore = semaphore;
        }
         
        @Override
        public void run() {
            try {
                semaphore.acquire();
                System.out.println("工人"+this.num+"占用一个机器在生产...");
                Thread.sleep(2000);
                System.out.println("工人"+this.num+"释放出机器");
                semaphore.release();           
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

1.5.  Executor框架

结构如下：

Executor 接口定义了最基本的 execute 方法，用于接收用户提交任务。 ExecutorService 定义了线程池终止和创建及提交 futureTask 任务支持的方法。

AbstractExecutorService 是抽象类，主要实现了 ExecutorService 和 futureTask 相关的一些任务创建和提交的方法。

ThreadPoolExecutor 是最核心的一个类，是线程池的内部实现。线程池的功能都在这里实现了，平时用的最多的基本就是这个了。其源码很精练，远没当时想象的多。

ScheduledThreadPoolExecutor 在 ThreadPoolExecutor 的基础上提供了支持定时调度的功能。线程任务可以在一定延时时间后才被触发执行。

ThreadPoolExecutor详解：

ThreadPoolExecutor的完整构造方法的签名是：

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)

corePoolSize - 池中所保存的线程数，包括空闲线程。

maximumPoolSize-池中允许的最大线程数。

keepAliveTime - 当线程数大于核心时，此为终止前多余的空闲线程等待新任务的最长时间。

unit - keepAliveTime 参数的时间单位。

workQueue - 执行前用于保持任务的队列。此队列仅保持由 execute方法提交的 Runnable任务。排队有三种通用策略:

直接提交SynchronousQueue(默认，)

无界队列LinkedBlockingQueue(两个ReenterLock，并发性能更好)

有界队列ArrayBlockingQueue(一个ReentrantLock和两个Condition实现阻塞)，或LinkedBlockingQueue（设置容量）

threadFactory - 执行程序创建新线程时使用的工厂。

handler - 由于超出线程范围和队列容量而使执行被阻塞时所使用的处理程序，四种预定义策略：

• ThreadPoolExecutor.AbortPolicy(默认，拒绝将抛出RejectedExecutionException)，
• ThreadPoolExecutor.CallerRunsPolicy(调用运行该任务的 execute 本身)，
• ThreadPoolExecutor.DiscardPolicy(不能执行的任务将被删除)，
• ThreadPoolExecutor.DiscardOldestPolicy(如果执行程序尚未关闭，则位于工作队列头部的任务将被删除，然后重试执行程序)

1. 如果当前池大小 poolSize 小于 corePoolSize ，则创建新线程执行任务。
2. 如果当前池大小 poolSize 大于 corePoolSize ，且等待队列未满，则进入等待队列
3. 如果当前池大小 poolSize 大于 corePoolSize 且小于 maximumPoolSize ，且等待队列已满，则创建新线程执行任务。
4. 如果当前池大小 poolSize 大于 corePoolSize 且大于 maximumPoolSize ，且等待队列已满，则调用拒绝策略来处理该任务。
5. 线程池里的每个线程执行完任务后不会立刻退出，而是会去检查下等待队列里是否还有线程任务需要执行，如果在 keepAliveTime 里等不到新的任务了，那么线程就会退出。

示例代码：

private class MyTask implements Callable<Boolean>{
        private int id=-1;
        public MyTask(int i){
            this.id=i;
        }
        @Override
        public Boolean call() throws Exception {
            Thread.sleep(3000);
            System.out.println("thread:"+id+"....done");
            if(id>30)return false;
            return true;
        }
}
@Test
    public void IntergerTest() throws ExecutionException, InterruptedException {
        ExecutorService exeService= Executors.newFixedThreadPool(10);
        ArrayList<Future<Boolean>> rst=new ArrayList<Future<Boolean>>();
        for(int i=0;i<10000;i++){
            Future<Boolean> rs= exeService.submit(new MyTask(i));
            rst.add(rs);
        }

        exeService.shutdown();//阻止提交任务，执行前面的任务
        for(Future<Boolean> rs:rst){
            Boolean b=rs.get();
            if(!b)exeService.shutdownNow();//阻止等待任务启动，试图中断正在执行的任务
        }
}

注：submit有多个重载

Future<T> submit(Callable<T> task)，常用

Future<T> submit(Runnable task, T result)

Future<?> submit(Runnable task)，Future<?>.get()返回null

还有个方法：void execute(Runnable command)也可执行任务

ps: submit方法会封装Callable等为FutureTask(实现RunnableFuture<V>接口)

public interface RunnableFuture<V> extends Runnable, Future<V> {
    void run();
}

调用execute(FutureTask) ，返回FutureTask。FutureTask覆写方法run方法如下

public void run() {
　　sync.innerRun();
}

Sync调用innerSet(callable.call());执行callable的call方法，调用自身innerSet

if (compareAndSetState(s, RAN)) {
result = v;
releaseShared(0);
done();
return;

}

执行完之后保存值在Sync.result中

下面分析FutureTask.get()时都发生了什么：

调用sync.innerGet()

V innerGet() throws InterruptedException, ExecutionException {
acquireSharedInterruptibly(0);
if (getState() == CANCELLED)
throw new CancellationException();
if (exception != null)
throw new ExecutionException(exception);
return result;
}

public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0) //没完成
doAcquireSharedInterruptibly(arg);
}

关键就是这了：

private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
        final Node node = addWaiter(Node.SHARED); //双向链表队列
        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
                        return;
                    }
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    break;
            }
        } catch (RuntimeException ex) {
            cancelAcquire(node);
            throw ex;
        }
        // Arrive here only if interrupted
        cancelAcquire(node);
        throw new InterruptedException();
    }

2. 原子类

2.1 基本类型，AtomicInteger

查看incrementAndGet方法源码

public final int incrementAndGet() {
    for (;;) {
        int current = get();
        int next = current + 1;
        if (compareAndSet(current, next))
            return next;
    }
}

可知它的原理：

(1)最外层是一个死循环

(2)先获取旧值，将其复制到一个局部变量上

(3)将局部变量值+1

(4)比较旧值是否变化，如果没变化，说明没有其它线程对旧值修改，直接将新值覆盖到旧值，并返回新值，退出循环

(5)如果旧值被修改了，开始下一轮循环，重复刚才这一系列操作，直到退出循环。

2.2 引用类型，AutomicReference,AtomicStampedReference,AtomicMarkableReference

类似基本AutomicInterger，只是它操作的是复杂的引用类型

2.3 基本类型字段更新器，AtomicIntegerFieldUpdater,AtomicLongFieldUpdater

字段必须用volatile修饰，示例代码

public class AtomicIntegerFieldUpdaterTest {

    private static AtomicIntegerFieldUpdater<User> a = AtomicIntegerFieldUpdater
            .newUpdater(User.class, "old");

    public static void main(String[] args) {
        User conan = new User("conan", 10);
        System.out.println(a.getAndIncrement(conan));
        System.out.println(a.get(conan));
    }

    public static class User {
        private String name;
        public volatile int old;

        public User(String name, int old) {
            this.name = name;
            this.old = old;
        }

        public String getName() {
            return name;
        }

        public int getOld() {
            return old;
        }
    }
}

2.4 引用类型字段更新器，AtomicReferenceFieldUpdater

同样需要volatile修饰，示例代码

public class Book
{
    private String name;

    public Book()
    {
    }

    public Book( String name )
    {
        this.name = name;
    }

    public String getName()
    {
        return name;
    }

    public void setName( String name )
    {
        this.name = name;
    }
}

import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;

public class MyObject
{
    private volatile Book whatImReading;

    private static final AtomicReferenceFieldUpdater<MyObject,Book> updater =
            AtomicReferenceFieldUpdater.newUpdater( 
                       MyObject.class, Book.class, "whatImReading" );

    public Book getWhatImReading()
    {
        return whatImReading;
    }

    public void setWhatImReading( Book whatImReading )
    {
        //this.whatImReading = whatImReading;
        updater.compareAndSet( this, this.whatImReading, whatImReading );
    }
}

3. java.util.concurrent.locks接口Lock、ReadWriteLock、Condition

lock主要方法有：lock()、tryLock()、tryLock(long time, TimeUnit unit)，lockInterruptibly()和unLock()

3.1 ReentrantLock

意思是“可重入锁”，ReentrantLock是唯一实现了Lock接口的类，实现原理见参考文章7 ReentrantLock实现原理深入探究

重点是AbstractQueuedSynchronizer抽象类的实现Sync，包括NonfairSync和FairSync，默认使用NonfairSync

lock时设置state和当前独占线程，之后的线程

public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}

阻塞在acquireQueued

PS：这里涉及了一点公平锁非公平锁的概念，

　　竞争lock的线程会排队，新线程进来先判断是不是排在对头，如果是，才查看当前状态；否则，排队等待。体现了先到先服务的精神，而非公平锁，新来的线程先查看状态，满足条件直接取得锁

3.2 ReentrantReadWriteLock

实现了ReadWriteLock接口，最主要的有两个方法：readLock()和writeLock()用来获取读锁和写锁

如果有一个线程已经占用了读锁，则此时其他线程如果要申请写锁，则申请写锁的线程会一直等待释放读锁；如果有一个线程已经占用了写锁，则此时其他线程如果申请写锁或者读锁，则申请的线程会一直等待释放写锁

写线程获取写入锁后可以获取读取锁，然后释放写入锁，这样就从写入锁变成了读取锁，从而实现锁降级的特性；读取锁是不能直接升级为写入锁的。因为获取一个写入锁需要释放所有读取锁，所以如果有两个读取锁视图获取写入锁而都不释放读取锁时就会发生死锁。

3.3 Condition

无论synchronized关键字，或者是lock上的await/signal等，都只能在一个锁上做同步通知。 
假设有3个线程，要对一个资源做同步，一般只能有一个锁来做同步通知操作，那么通知的时候无法做到精确的通知3个线程中的某一个的，因为你调用了wait()/notify()的时候，具体的调度是jvm决定的，但是有的时候的确需要需要对一个锁做多种不同情况的精确通知， 比如一个缓存，满了和空了是两种不同的情况，可以分别通知取数据的线程和放数据的线程

基本使用如下:  
* Condition是个接口，基本的方法就是await()和signal()方法；  
* Condition依赖于Lock接口，生成一个Condition的基本代码是lock.newCondition()  
* 调用Condition的await()和signal()方法，都必须在lock保护之内，就是说必须在lock.lock()和lock.unlock之间才可以  
* 和Object.wait()方法一样，每次调用Condition的await()方法的时候，当前线程就自动释放了对当前锁的拥有权  

示例

class BoundedBuffer {
   final Lock lock = new ReentrantLock();
   final Condition notFull  = lock.newCondition(); 
   final Condition notEmpty = lock.newCondition(); 

   final Object[] items = new Object[100];
   int putptr, takeptr, count;

   public void put(Object x) throws InterruptedException {
     lock.lock();//第一步实现互斥
     try {
       while (count == items.length)//如果没有往数组放,线程阻塞
         notFull.await();
       items[putptr] = x;
       if (++putptr == items.length) putptr = 0;//如果putptr已经是数组的最后一个,那么putptr置为0,从第一个开始放
       ++count;//放完后,把总数加一
       notEmpty.signal();//通知其他线程可以取了
     } finally {
       lock.unlock();
     }
   }

   public Object take() throws InterruptedException {
     lock.lock();
     try {
       while (count == 0)
         notEmpty.await();
       Object x = items[takeptr];
       if (++takeptr == items.length) takeptr = 0;
       --count;
       notFull.signal();
       return x;
     } finally {
       lock.unlock();
     }
   }
}

参考文章：

1. Java集合---ConcurrentHashMap原理分析

2. 探索 ConcurrentHashMap 高并发性的实现机制

3. 聊聊并发（六）——ConcurrentLinkedQueue的实现原理分析

4. Java并发编程：CountDownLatch、CyclicBarrier和Semaphore

5. Java并发编程

6. FutureTask 源码解析

7. ReentrantLock实现原理深入探究

8. 关于Java并发编程的总结和思考

9. 深入剖析基于并发AQS的(独占锁)重入锁(ReetrantLock)及其Condition实现原理