并发编程重要吗？当然重要，因为并发在我们的项目中真实存在，如果你不能充分了解它那么很可能造成严重的生产事故。最近隔壁项目组出了一个问题，每次请求接口之后都发现线程固定增加了5个，而且线程数一直增加没有减少，他们怀疑是中间件的问题，但实际上是因为他们的代码中线程池使用不当造成。所以了解并发是很有必要的。

1.基础概念

1. 并行：两台饮水机，两个人去接水一人使用一台
2. 并发：一台饮水机，两个人去接水，需要等待一个人接完才行
3. 异步：不需要等待结果返回
4. 同步：需要等待结果返回
5. 线程：是运行在进程内：比如音乐播放器打开之后，在随机播放着音乐 的同时，还可以继续搜索我们想听的音乐。可以理解有两个线程分别处理着音乐播放和音乐搜索
6. 进程：是一个具体的应用示例：比如打开音乐播放器就是运行了一个进程
7. 线程上下文切换：cpu调度线程时切换线程同时需要保存线程相关数据，cpu和内存之间

2.Thread

2.1 线程的创建

线程的创建到底有几种？这个答案其实不是固定。要看回答的方向

There are two ways to create a new thread of execution
eg1:new Thread().start()
eg2:R implements Runnable

上面这句话是Thread类上面的注释，也就是官方说是两种，但是实际使用过程中根据根据这两种又有衍生和变化。概括有一下几种

1. new Thread().start();
2. R implements Runnablenew Thread(R).start();
3. C implements Callablenew Thread(new FutureTask<>(C)).start();
4. R implements Runnablenew ThreadPoolExecutor().execute(R);

2.2 线程的中断方法

1. Thread.interrupt() 给线程加一个中断标记
2. Thread.interrupted() 判断线程是否中断，会清除标记
3. Thread.isInterrupted() 判断线程是否中断，不会清除标记

2.3 线程的等待唤醒机制

1. synchronized 的 wait(),notify()
2. locksupport 的 park(),unpark()
3. condition wait() signal()

3.ThreadLocal（源码）

下面这段代码大概就是我们使用ThreadLocal的方式，接下来将根据这段代码中的方法对ThreadLocal原理进行剖析。

    // 1.newThreadLocal<Map<String,String>> myThreadLocal = new ThreadLocal<>();// 2.存值Map<String,String> map = new HashMap<>();map.put("org","0001");myThreadLocal.set(map);// 3.获取值map = myThreadLocal.get();String org = map.get("org");System.out.println(org);// 4.清空myThreadLocal.remove();

3.1 new ThreadLocal<>()

    public ThreadLocal() {}

3.2 ThreadLocal#set()

    public void set(T value) {// 获取当前线程Thread t = Thread.currentThread();// 获取thread对应的mapThreadLocalMap map = getMap(t);if (map != null)map.set(this, value);else// 创建mapcreateMap(t, value);}

3.2.1ThreadLocal#createMap

    void createMap(Thread t, T firstValue) {// 给thread的threadLocals属性赋值t.threadLocals = new ThreadLocalMap(this, firstValue);}

3.2.1.1 new ThreadLocalMap()

ThreadLocalMap(ThreadLocal<?> firstKey, Object firstValue) {// 创建初始长度为16的数组table = new Entry[INITIAL_CAPACITY];// firstKey传的是this 即myThreadLocal对象本身int i = firstKey.threadLocalHashCode & (INITIAL_CAPACITY - 1);// 数组中存入值：key=myThreadLocal,firstValue=maptable[i] = new Entry(firstKey, firstValue);size = 1;setThreshold(INITIAL_CAPACITY);}

3.2.1.2 ThreadLocal#getMap

    ThreadLocalMap getMap(Thread t) {// 获取thread中的ThreadLocalMap属性return t.threadLocals;}

执行完myThreadLocal.set(map)，最终结构如下图：

3.3 myThreadLocal.get() 方法

    public T get() {Thread t = Thread.currentThread();// 获取thread的属性ThreadLocalMapThreadLocalMap map = getMap(t);if (map != null) {// 获取key为myThreadLocal的Entry ThreadLocalMap.Entry e = map.getEntry(this);if (e != null) {@SuppressWarnings("unchecked")T result = (T)e.value;// 返回value即为set存储的mapreturn result;}}// 如果当前线程未set值，默认返回nullreturn setInitialValue();}

3.3.1 map.getEntry(this) 方法

       private Entry getEntry(ThreadLocal<?> key) {// 和set时一样，使用myThreadLocal对象获取数组的下标int i = key.threadLocalHashCode & (table.length - 1);Entry e = table[i];// 当前下标位置数组值不为null，且key值相等。即为get返回值if (e != null && e.get() == key)return e;elsereturn getEntryAfterMiss(key, i, e);}

3.4 myThreadLocal.remove() 方法

     public void remove() {// 获取当前程的属性ThreadLocalMapThreadLocalMap m = getMap(Thread.currentThread());if (m != null)// 移除当前myThreadLocal对象所在下标的数组的值m.remove(this);}

3.4.1 m.remove(this)

        private void remove(ThreadLocal<?> key) {Entry[] tab = table;int len = tab.length;int i = key.threadLocalHashCode & (len-1);// 获取对象下标的数组值for (Entry e = tab[i];e != null;e = tab[i = nextIndex(i, len)]) {if (e.get() == key) {e.clear();expungeStaleEntry(i);return;}}}

4.ReentrantLock(非公平锁实现源码)

下面的源码实现主要列的是非公平锁，公平锁代码和非公平锁差不多所以鼓励大家自己看看。看源码前请大家想象一个场景，现在有两个线程，第一个线程获取锁成功了在执行业务逻辑时，此时第二个线程过来加锁。下面的源码分析的就是上面这样一个场景。

常规使用如下：

 // 1.创建 ReentrantLock 对象ReentrantLock lock = new ReentrantLock(); // 2.获取锁lock.lock(); try {// 业务逻辑} finally {// 3.释放锁lock.unlock(); }

4.1 new ReentrantLock()

    public ReentrantLock() {// 给ReentrantLock属性赋值；NonfairSync实际上就是抽象的队列同步器sync = new NonfairSync();}

NonfairSync类的继承关系

4.2 lock.lock()

    public void lock() {sync.lock();}

4.2.1 sync.lock()

   final void lock() {// 使用CAS的方式尝试将同步等待队列中的state由0改为1if (compareAndSetState(0, 1))// 修改成功代表获取锁成功，将独占线程赋值为当前线程setExclusiveOwnerThread(Thread.currentThread());else// 1.获取锁acquire(1);}

4.2.1.1 acquire(1)

    public final void acquire(int arg) {// tryAcquire尝试获取一次锁，如果失败了addWaiter进入队列，然后再次获取锁acquireQueuedif (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))selfInterrupt();}

4.2.1.1.1 tryAcquire

	protected final boolean tryAcquire(int acquires) {// 非公平获取锁return nonfairTryAcquire(acquires);}

4.2.1.1.1.1 nonfairTryAcquire（）方法

	// acquires从2.1方法中传过来 固定为1final boolean nonfairTryAcquire(int acquires) {final Thread current = Thread.currentThread();// 获取同步等待队列的state属性，默认值为0，如果不为0代表有线程获取锁成功int c = getState();if (c == 0) {// 如果当前没有线程获取到锁，就直接cas加锁if (compareAndSetState(0, acquires)) {setExclusiveOwnerThread(current);return true;}}// c!=0,当前已有线程获取了锁。判断是不是同一个线程else if (current == getExclusiveOwnerThread()) {// 可重入锁实现 通过state值自增判断重入几次int nextc = c + acquires;if (nextc < 0) // overflowthrow new Error("Maximum lock count exceeded");setState(nextc);return true;}return false;}

4.2.1.1.2 addWaiter

	// node从4.2.1.1传递的是nullprivate Node addWaiter(Node mode) {// 构建node节点Node node = new Node(Thread.currentThread(), mode);// Try the fast path of enq; backup to full enq on failure// 1.将尾节点赋值给变量predNode pred = tail;// 2.判断尾节点是否为空，为空则代表链表为空未进行初始化（双向指针构造的）if (pred != null) {// 2.1 尾节点不为空，则将当前node节点添加到链表尾部node.prev = pred;if (compareAndSetTail(pred, node)) {pred.next = node;return node;}}// 2.2 尾节点为空，则构建链表，并且插入enq(node);return node;}

4.2.1.1.2.1 enq()

    private Node enq(final Node node) {for (;;) {Node t = tail;// 1.当尾节点为空，构造链表（new node()设置为头节点，并且尾节点和头节点指向同一个node）if (t == null) { // Must initializeif (compareAndSetHead(new Node()))tail = head;} else {// 2.尾节点不为空，将当前获取锁失败的线程所构建的node作为尾节点，并且修改双指针指向node.prev = t;if (compareAndSetTail(t, node)) {t.next = node;return t;}}}}

我们假设thread1，先获取到了锁，那么thread2执行完上面的enq()方法，双向链表形状大概如下图所示

4.2.1.1.3 acquireQueued（）

	// node为获取锁失败的线程构建的node，arg为1final boolean acquireQueued(final Node node, int arg) {boolean failed = true;try {boolean interrupted = false;for (;;) {//1.获取node节点的前置节点final Node p = node.predecessor();//2 如果前置节点为头节点，则链表中实际只有一个线程在等待。再次尝试获取锁,如果获取到了锁，将当前node设置为headif (p == head && tryAcquire(arg)) {setHead(node);p.next = null; // help GCfailed = false;return interrupted;}//3 判断线程是否应该阻塞，修改node节点waitStatus==-1等待被唤醒,并且阻塞线程if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())interrupted = true;}} finally {if (failed)cancelAcquire(node);}}

4.2.1.1.3.1 shouldParkAfterFailedAcquire（）

Node有一个volatile int waitStatus属性
默认=0；SIGNAL=-1；CANCELLED=1；CONDITION=-2；PROPAGATE=-3

	// pred:当前线程构建节点的前置节点 node:当前线程构建的节点private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {int ws = pred.waitStatus;// 第二次循环waitStatus==-1，因为4.2.1.1.3 有for (;;)所以会再次进入if (ws == Node.SIGNAL)return true;if (ws > 0) {do {node.prev = pred = pred.prev;} while (pred.waitStatus > 0);pred.next = node;} else {// 第一次循环waitStatus==0,默认走这里。通过cas将0修改为-1compareAndSetWaitStatus(pred, ws, Node.SIGNAL);}return false;}

执行完shouldParkAfterFailedAcquire方法后，链表状态大概如下图：

4.2.1.1.3.1 parkAndCheckInterrupt（）

	// 底层使用UNSAFE.park将线程阻塞在这里private final boolean parkAndCheckInterrupt() {LockSupport.park(this);return Thread.interrupted();}

4.3 lock.unlock()

    public void unlock() {sync.release(1);}

4.3.1AbstractQueuedSynchronizer#release

    public final boolean release(int arg) {// 1.尝试释放锁if (tryRelease(arg)) {Node h = head;if (h != null && h.waitStatus != 0)// 2.唤醒等待的后置节点unparkSuccessor(h);return true;}return false;}

4.3.1.1 ReentrantLock.Sync#tryRelease

		// releases==1protected final boolean tryRelease(int releases) {int c = getState() - releases;// 如果释放锁的线程不等于加锁线程抛异常if (Thread.currentThread() != getExclusiveOwnerThread())throw new IllegalMonitorStateException();boolean free = false;// 释放锁。即将独占线程置为null,恢复state=0if (c == 0) {free = true;setExclusiveOwnerThread(null);}setState(c);return free;}

4.3.1.2 AbstractQueuedSynchronizer#unparkSuccessor

	// 这里的node==headprivate void unparkSuccessor(Node node) {// ws==-1int ws = node.waitStatus;if (ws < 0)compareAndSetWaitStatus(node, ws, 0);Node s = node.next;if (s == null || s.waitStatus > 0) {s = null;for (Node t = tail; t != null && t != node; t = t.prev)if (t.waitStatus <= 0)s = t;}// head节点的下一个节点即为thread2所在节点。唤醒阻塞中的thread2if (s != null)LockSupport.unpark(s.thread);}

5.ArrayBlockingQueue

整个源码逻辑很多，下面梳理的代码，以最简单的情况为例。一个Producer线程向队列中添加元素，一个Consumer线程从队列中获取元素。有了这个前提再看下面的源码会更加容易一些。

常见使用方式如下：

	......省略ArrayBlockingQueue<Integer> abq = new ArrayBlockingQueue<>(16);......省略abq.put(new Random(100).nextInt());......省略abq.take();

5.1 ArrayBlockingQueue#ArrayBlockingQueue(int)

	// capacity初始容量public ArrayBlockingQueue(int capacity) {this(capacity, false);}

5.1.1 ArrayBlockingQueue#ArrayBlockingQueue(int, boolean)

	// fair是否公平？ 默认falsepublic ArrayBlockingQueue(int capacity, boolean fair) {if (capacity <= 0)throw new IllegalArgumentException();// 创建指定长度的数组this.items = new Object[capacity];// 创建非公平锁lock = new ReentrantLock(fair);// 获取conditionnotEmpty = lock.newCondition();notFull =  lock.newCondition();}

5.2 ArrayBlockingQueue#put

    public void put(E e) throws InterruptedException {checkNotNull(e);final ReentrantLock lock = this.lock;// 加锁（可中断锁）lock.lockInterruptibly();try {// count为数组中存的元素个数 相等则将producer线程await在这里while (count == items.length)notFull.await();enqueue(e);} finally {lock.unlock();}}

5.2.1 ArrayBlockingQueue#enqueue

    private void enqueue(E x) {final Object[] items = this.items;// putIndex数组下标>>存数下标 默认putIndex==0items[putIndex] = x;// 循环数组实现。数组存满之后，下标恢复为0if (++putIndex == items.length)putIndex = 0;// count记录数组中存的元素格式count++;// 队列中有元素之后，唤醒consumer线程进行消费notEmpty.signal();}

5.2.2 AbstractQueuedSynchronizer.ConditionObject#await()

     public final void await() throws InterruptedException {if (Thread.interrupted())throw new InterruptedException();// 见 5.2.2.1 Node node = addConditionWaiter();// 这里返回savedState==1int savedState = fullyRelease(node);int interruptMode = 0;// isOnSyncQueue 返回fasle,所以while条件成立while (!isOnSyncQueue(node)) {// producer 线程在这里park阻塞LockSupport.park(this);if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)break;}if (acquireQueued(node, savedState) && interruptMode != THROW_IE)interruptMode = REINTERRUPT;if (node.nextWaiter != null) // clean up if cancelledunlinkCancelledWaiters();if (interruptMode != 0)reportInterruptAfterWait(interruptMode);}

5.2.2.1 AbstractQueuedSynchronizer.ConditionObject#addConditionWaiter

    private Node addConditionWaiter() {// lastWaiter默认为nullNode t = lastWaiter;if (t != null && t.waitStatus != Node.CONDITION) {unlinkCancelledWaiters();t = lastWaiter;}Node node = new Node(Thread.currentThread(), Node.CONDITION);if (t == null)firstWaiter = node;elset.nextWaiter = node;lastWaiter = node;return node;}

第一次执行完addConditionWaiter方法，Node大概是如下结构

5.2.2.2 AbstractQueuedSynchronizer#fullyRelease

    final int fullyRelease(Node node) {boolean failed = true;try {// 在5.2中先加锁了，所以这里默认返回savedState==1int savedState = getState();// 这里if条件成立 直接return 1if (release(savedState)) {failed = false;return savedState;} else {throw new IllegalMonitorStateException();}} finally {if (failed)node.waitStatus = Node.CANCELLED;}}

5.2.2.2.1 AbstractQueuedSynchronizer#release

    public final boolean release(int arg) {if (tryRelease(arg)) {// 前面的代码没有给head赋值 所以这里null==head,这里直接返回trueNode h = head;if (h != null && h.waitStatus != 0)unparkSuccessor(h);return true;}return false;}

5.2.2.3 AbstractQueuedSynchronizer#isOnSyncQueue

  final boolean isOnSyncQueue(Node node) {// node前面执行addConditionWaiter方法时，-2==waitStatus 。所以这里直接return faleif (node.waitStatus == Node.CONDITION || node.prev == null)return false;if (node.next != null) // If has successor, it must be on queuereturn true;return findNodeFromTail(node);}

5.2 ArrayBlockingQueue#take

   public E take() throws InterruptedException {final ReentrantLock lock = this.lock;// 先加锁lock.lockInterruptibly();try {// 当数组中元素数量=0时,让consumer线程await在这里while (count == 0)notEmpty.await();// 队列中有元素时 获取数组元素return dequeue();} finally {lock.unlock();}}

5.2.1 ArrayBlockingQueue#dequeue

   private E dequeue() {final Object[] items = this.items;// takeIndex为获取数组时的下标 默认从0开始E x = (E) items[takeIndex];items[takeIndex] = null;// 当取到数组最后一个元素值重置下标if (++takeIndex == items.length)takeIndex = 0;// 没消费一个元素 数组元素个数-1count--;if (itrs != null)itrs.elementDequeued();// 当队列中元素被消费后 唤醒producer线程进行putnotFull.signal();return x;}

6 CountDownLatch

源码分析的场景如下：第一个线程启动后调用await方法被阻塞，第二个线程调用countDown方法将state置为0，再唤醒第一个阻塞的线程

常见使用方法如下：

   CountDownLatch countDownLatch = new CountDownLatch(1);countDownLatch.await();countDownLatch.countDown();

6.1 CountDownLatch#CountDownLatch

  // 这里为了简单，我们示例中假设cont==1public CountDownLatch(int count) {if (count < 0) throw new IllegalArgumentException("count < 0");// Sync extends AbstractQueuedSynchronizer (这一步相当于state=count)this.sync = new Sync(count);}

6.2 CountDownLatch#await()

    public void await() throws InterruptedException {// 判断state!=0时阻塞线程sync.acquireSharedInterruptibly(1);}

6.2.1 AbstractQueuedSynchronizer#acquireSharedInterruptibly

   public final void acquireSharedInterruptibly(int arg)throws InterruptedException {if (Thread.interrupted())throw new InterruptedException();// 判断state是否等于0 ,因为初始化时我们设置state==1,所以这 -1<0  if成立if (tryAcquireShared(arg) < 0)doAcquireSharedInterruptibly(arg);}

6.2.1.1 CountDownLatch.Sync#tryAcquireShared

   protected int tryAcquireShared(int acquires) {return (getState() == 0) ? 1 : -1;}

6.2.1.2 AbstractQueuedSynchronizer#doAcquireSharedInterruptibly

private void doAcquireSharedInterruptibly(int arg)throws InterruptedException {// addWaiter 在前面ReentrantLock中讲过。主要构建双向链表final Node node = addWaiter(Node.SHARED);boolean failed = true;try {for (;;) {final Node p = node.predecessor();if (p == head) {// 这里又判断 state是否等于0 ，所以if条件不成立int r = tryAcquireShared(arg);if (r >= 0) {setHeadAndPropagate(node, r);p.next = null; // help GCfailed = false;return;}}// 最终线程会被park方法阻塞在这里。这个方法在ReentrantLock中讲过if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())throw new InterruptedException();}} finally {if (failed)cancelAcquire(node);}
}

6.3 CountDownLatch#countDown

  public void countDown() {// 执行state-1，当sate==0，唤醒阻塞的线程sync.releaseShared(1);}

6.3.1 AbstractQueuedSynchronizer#releaseShared

 public final boolean releaseShared(int arg) {// if条件成立if (tryReleaseShared(arg)) {doReleaseShared();return true;}return false;}

6.3.1.1 CountDownLatch.Sync#tryReleaseShared

  protected boolean tryReleaseShared(int releases) {for (;;) {// 这里getState()==1int c = getState();if (c == 0)return false;int nextc = c-1;// 使用cas将state从1修改为0if (compareAndSetState(c, nextc))return nextc == 0;}}

6.3.1.2 AbstractQueuedSynchronizer#doReleaseShared

到这里再看一下前面的node的结构

private void doReleaseShared() {for (;;) {Node h = head;// 根据前面构建的node的结构，接下来的两个if条件都是满足的if (h != null && h != tail) {int ws = h.waitStatus;if (ws == Node.SIGNAL) {if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))continue; // 这里唤醒后面的阻塞线程unparkSuccessor(h);}else if (ws == 0 &&!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))continue;                // loop on failed CAS}if (h == head)                   // loop if head changedbreak;}}

6.3.1.2.1 AbstractQueuedSynchronizer#unparkSuccessor

// 这里node是前面传进来的head
private void unparkSuccessor(Node node) {int ws = node.waitStatus;if (ws < 0)compareAndSetWaitStatus(node, ws, 0);// s节点的线程就是我们自己被阻塞的线程Node s = node.next;if (s == null || s.waitStatus > 0) {s = null;for (Node t = tail; t != null && t != node; t = t.prev)if (t.waitStatus <= 0)s = t;}// 通过unpark唤醒我们前面通过park方法阻塞的线程if (s != null)LockSupport.unpark(s.thread);
}