Java中高级核心知识全面解析(2)，netty实战百度网盘_java读源码之netty深入剖析百度网盘

}
else if (current == getExclusiveOwnerThread()) { 
	int nextc = c + acquires; 
	if (nextc < 0) 
		throw new Error("Maximum lock count exceeded"); 
	setState(nextc); 
	return true; 
}
return false; 
} 
1
2
3
4
5
6
7
8
9
10

}

非公平锁的 lock 方法：
```java
static final class NonfairSync extends Sync { 
	final void lock() { 
		// 2. 和公平锁相比，这里会直接先进行一次CAS，成功就返回了 
		if (compareAndSetState(0, 1)) 
			setExclusiveOwnerThread(Thread.currentThread()); 
			else
				acquire(1); 
}
// AbstractQueuedSynchronizer.acquire(int arg) 
public final void acquire(int arg) { 
	if (!tryAcquire(arg) && 
		acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) 
		selfInterrupt(); 
}
protected final boolean tryAcquire(int acquires) { 
	return nonfairTryAcquire(acquires); 
	} 
}
/**
 1. Performs non-fair tryLock. tryAcquire is implemented in 
 2. subclasses, but both need nonfair try for trylock method. 
*/ 
final boolean nonfairTryAcquire(int acquires) { 
	final Thread current = Thread.currentThread(); 
	int c = getState(); 
	if (c == 0) { 
		// 这里没有对阻塞队列进行判断 
		if (compareAndSetState(0, acquires)) { 
			setExclusiveOwnerThread(current); 
			return true; 
		} 
	}
	else if (current == getExclusiveOwnerThread()) { 
		int nextc = c + acquires; 
		if (nextc < 0) // overflow 
			throw new Error("Maximum lock count exceeded");
		setState(nextc);
		return true; 
	}
		return false; 
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

总结：公平锁和非公平锁只有两处不同：

1. 非公平锁在调用lock后，首先就会调用CAS进行一次抢锁，如果这个时候恰巧锁没有被占用，那么直接就获取到锁返回了。
2. 非公平锁在CAS失败后，和公平锁一样都会进入到tryAcquire方法，在 tryAcquire方法中，如果发现锁这个时候被释放了（state == 0），非公平锁会直接CAS抢锁，但是公平锁会判断等待队列是否有线程处于等待状态，如果有则不去抢锁，乖乖排到后面。

公平锁和非公平锁就这两点区别，如果这两次CAS都不成功，那么后面非公平锁和公平锁是一样的，都要进入到阻塞队列等待唤醒。

相对来说，非公平锁会有更好的性能，因为它的吞吐量比较大。当然，非公平锁让获取锁的时间变得更加不确定，可能会导致在阻塞队列中的线程长期处于饥饿状态。

2)Share（共享）

多个线程可同时执行，如Semaphore/CountDownLatch。Semaphore、CountDownLatCh、CyclicBarrier、ReadWriteLock我们都会在后面讲到。

ReentrantReadWriteLock可以看成是组合式，因为ReentrantReadWriteLock也就是读写锁允许多个线程同时对某一资源进行读。

不同的自定义同步器争用共享资源的方式也不同。自定义同步器在实现时只需要实现共享资源state的获取与释放方式即可，至于具体线程等待队列的维护（如获取资源失败入队/唤醒出队等），AQS已经在上层已经帮我们实现好了。

3. AQS底层使用了模板方法模式

同步器的设计是基于模板方法模式的，如果需要自定义同步器一般的方式是这样（模板方法模式很经典的一个应用）：

1. 使用者继承AbstractQueuedSynchronizer并重写指定的方法。（这些重写方法很简单，无非是对于共享资源state的获取和释放）
2. 将AQS组合在自定义同步组件的实现中，并调用其模板方法，而这些模板方法会调用使用者重写的方法。

这和我们以往通过实现接口的方式有很大区别，这是模板方法模式很经典的一个运用，下面简单的给大家介绍一下模板方法模式，模板方法模式是一个很容易理解的设计模式之一。

模板方法模式是基于”继承“的，主要是为了在不改变模板结构的前提下在子类中重新定义模板中的内容以实现复用代码。举个很简单的例子假如我们要去一个地方的步骤是：购票
buyTicket()->安检securityCheck()->乘坐某某工具回家ride()->到达目的地 arrive() 。我们可能乘坐不同的交通工具回家比如飞机或者火车，所以除了 ride()方法，其他方法的实现几乎相同。我们可以定义一个包含了这些方法的抽象类，然后用户根据自己的需要继承该抽象类然后修改ride()方法。

AQS使用了模板方法模式，自定义同步器时需要重写下面几个AQS提供的模板方法：

isHeldExclusively()//该线程是否正在独占资源。只有用到condition才需要去实现它。 
tryAcquire(int)//独占方式。尝试获取资源，成功则返回true，失败则返回false。 
tryRelease(int)//独占方式。尝试释放资源，成功则返回true，失败则返回false。 
tryAcquireShared(int)//共享方式。尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源； 正数表示成功，且有剩余资源。 
tryReleaseShared(int)//共享方式。尝试释放资源，成功则返回true，失败则返回false。
1
2
3
4
5

默认情况下，每个方法都抛出UnsupportedOperationException。 这些方法的实现必须是内部线程安全的，并且通常应该简短而不是阻塞。AQS类中的其他方法都是final，所以无法被其他类使用，只有这几个方法可以被其他类使用。

以ReentrantLock为例，state 初始化为 0，表示未锁定状态。A 线程 lock()时，会调用tryAcquire()独占该锁并将state+1。此后，其他线程再 tryAcquire()时就会失败，直到 A 线程 unlock()到 state=0（即释放锁）为止，其它线程才有机会获取该锁。当然，释放锁之前，A 线程自己是可以重复获取此锁的（state 会累加），这就是可重入的概念。但要注意，获取多少次就要释放多么次，这样才能保证 state是能回到零态的。

再以 CountDownLatch 以例，任务分为 N 个子线程去执行，state 也初始化为 N（注意 N 要与线程个数一致）。这 N 个子线程是并行执行的，每个子线程执行完后 countDown()一次，state会CAS(Compare and Swap)减 1。等到所有子线程都执行完后(即state=0)，会unpark()主调用线程，然后主调用线程就会从await()函数返回，继续后余动作。

一般来说，自定义同步器要么是独占方法，要么是共享方式，他们也只需实现tryAcquire- tryRelease 、tryAcquireShared-tryReleaseShared中的一种即可。但AQS也支持自定义同步器同时实现独占和共享两种方式，如ReentrantReadWriteLock。

三、Semaphore(信号量)-允许多个线程同时访问

synchronized和ReentrantLock 都是一次只允许一个线程访问某个资源，Semaphore(信号量)可以指定多个线程同时访问某个资源。

示例代码如下：

/**
* @author Snailclimb 
* @date 2018年9月30日 
* @Description: 需要一次性拿一个许可的情况 
*/
public class SemaphoreExample1 { 
	// 请求的数量 
	private static final int threadCount = 550; 
	public static void main(String[] args) throws InterruptedException { 
		// 创建一个具有固定线程数量的线程池对象（如果这里线程池的线程数量给太少的话你会发现执行的很 慢） 
		ExecutorService threadPool = Executors.newFixedThreadPool(300); 
		// 一次只能允许执行的线程数量。 
		final Semaphore semaphore = new Semaphore(20); 
		
		for (int i = 0; i < threadCount; i++) { 
			final int threadnum = i; 
			threadPool.execute(() -> {// Lambda 表达式的运用 
				try { 
					semaphore.acquire();// 获取一个许可，所以可运行线程数量为20/1=20
				test(threadnum); 
				semaphore.release();// 释放一个许可 
			} catch (InterruptedException e) { 
				// TODO Auto-generated catch block 
				e.printStackTrace(); 
			} 
			
		}); 
	}
	threadPool.shutdown(); 
	System.out.println("finish"); 
}
	public static void test(int threadnum) throws InterruptedException { 
		Thread.sleep(1000);// 模拟请求的耗时操作 
		System.out.println("threadnum:" + threadnum); 
		Thread.sleep(1000);// 模拟请求的耗时操作 
	} 
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

执行acquire方法阻塞，直到有一个许可证可以获得然后拿走一个许可证；每个release方法增加一个许可证，这可能会释放一个阻塞的acquire方法。然而，其实并没有实际的许可证这个对象，Semaphore只是维持了一个可获得许可证的数量。Semaphore经常用于限制获取某种资源的线程数量。

当然一次也可以一次拿取和释放多个许可，不过一般没有必要这样做：

semaphore.acquire(5);// 获取5个许可，所以可运行线程数量为20/5=4 
test(threadnum); 
semaphore.release(5);// 获取5个许可，所以可运行线程数量为20/5=4
1
2
3

除了acquire方法之外，另一个比较常用的与之对应的方法是tryAcquire方法，该方法如果获取不到许可就立即返回 false。

Semaphore有两种模式，公平模式和非公平模式。

• 公平模式：调用acquire的顺序就是获取许可证的顺序，遵循FIFO；
• 非公平模式：抢占式的。

Semaphore 对应的两个构造方法如下：

public Semaphore(int permits) { 
	sync = new NonfairSync(permits); 
}
public Semaphore(int permits, boolean fair) { 
	sync = fair ? new FairSync(permits) : new NonfairSync(permits); 
}
1
2
3
4
5
6

这两个构造方法，都必须提供许可的数量，第二个构造方法可以指定是公平模式还是非公平模式，默认非公平模式。

补充：Semaphore与CountDownLatch一样，也是共享锁的一种实现。它默认构造AQS的state为permits。当执行任务的线程数量超出permits,那么多余的线程将会被放入阻塞队列Park,并自旋判断state是否大于0。只有当state大于0的时候，阻塞的线程才能继续执行,此时先前执行任务的线程继续执行release方法，release方法使得state的变量会加1，那么自旋的线程便会判断成功。

如此，每次只有最多不超过permits数量的线程能自旋成功，便限制了执行任务线程的数量。

四、CountDownLatch（倒计时器）

CountDownLatch允许count个线程阻塞在一个地方，直至所有线程的任务都执行完毕。在 Java 并发中，countdownlatch的概念是一个常见的面试题，所以一定要确保你很好的理解了它。

CountDownLatch是共享锁的一种实现,它默认构造 AQS 的 state 值为 count。当线程使用countDown方法时,其实使用了 tryReleaseShared 方法以CAS的操作来减少state,直至state为0就代表所有的线程都调用了countDown方法。当调用await方法的时候，如果state不为0，就代表仍然有线程没有调用countDown方法，那么就把已经调用过countDown的线程都放入阻塞队列Park,并自旋CAS判断state== 0，直至最后一个线程调用了countDown，使得state == 0，于是阻塞的线程便判断成功，全部往下执行。

1.CountDownLatch的两种典型用法

1. 某一线程在开始运行前等待 n 个线程执行完毕。将CountDownLatch的计数器初始化为 n ： new CountDownLatch(n) ，每当一个任务线程执行完毕，就将计数器减 1
   countdownlatch.countDown() ，当计数器的值变为 0 时，在CountDownLatch上await()的线程就会被唤醒。一个典型应用场景就是启动一个服务时，主线程需要等待多个组件加载完毕，之后再继续执行。
2. 实现多个线程开始执行任务的最大并行性。注意是并行性，不是并发，强调的是多个线程在某一时刻同时开始执行。类似于赛跑，将多个线程放到起点，等待发令枪响，然后同时开跑。做法是初始化一个共享的 CountDownLatch 对象，将其计数器初始化为 1 ： new CountDownLatch(1) ，多个线程在开始执行任务前首先 coundownlatch.await() ，当主线程调用 countDown() 时，计数器变为 0，多个线程同时被唤醒。

2. CountDownLatch的使用示例

/**
* @author SnailClimb 
* @date 2020年12月10日 
* @Description: CountDownLatch 使用方法示例 
*/

public class CountDownLatchExample1 { 
	// 请求的数量 
	private static final int threadCount = 550; 
	
	public static void main(String[] args) throws InterruptedException { 
		// 创建一个具有固定线程数量的线程池对象（如果这里线程池的线程数量给太少的话你会发现执行的很 慢） 
		ExecutorService threadPool = Executors.newFixedThreadPool(300); 
		final CountDownLatch countDownLatch = new CountDownLatch(threadCount); 
		for (int i = 0; i < threadCount; i++) { 
			final int threadnum = i; 
			threadPool.execute(() -> {// Lambda 表达式的运用 
				try { 
					test(threadnum); 
				} catch (InterruptedException e) { 
					// TODO Auto-generated catch block 
					e.printStackTrace(); 
				} finally { 
					countDownLatch.countDown();// 表示一个请求已经被完成 	
				}
			}); 
		}
		countDownLatch.await(); 
		threadPool.shutdown(); 
		System.out.println("finish"); 
	}
	
	public static void test(int threadnum) throws InterruptedException { 
		Thread.sleep(1000);// 模拟请求的耗时操作 
		System.out.println("threadnum:" + threadnum); 
		Thread.sleep(1000);// 模拟请求的耗时操作 
	} 
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

上面的代码中，我们定义了请求的数量为 550，当这 550 个请求被处理完成之后，才会执行System.out.println("finish"); 。

与 CountDownLatch 的第一次交互是主线程等待其他线程。主线程必须在启动其他线程后立即调用CountDownLatch.await() 方法。这样主线程的操作就会在这个方法上阻塞，直到其他线程完成各自的任务。

其他 N 个线程必须引用闭锁对象，因为他们需要通知CountDownLatch对象，他们已经完成了各自的任务。这种通知机制是通过CountDownLatch.countDown()方法来完成的；每调用一次这个方法，在构造函数中初始化的count值就减1。所以当N个线程都调 用了这个方法，count的值等于 0，然后主线程就能通过await()方法，恢复执行自己的任务。

再插一嘴：CountDownLatch的await()方法使用不当很容易产生死锁，比如我们上面代码中的for循环改为：

for (int i = 0; i < threadCount-1; i++) { 
	....... 
}
1
2
3

这样就导致count的值没办法等于0，然后就会导致一直等待。

3.CountDownLatch的不足

CountDownLatch是一次性的，计数器的值只能在构造方法中初始化一次，之后没有任何机制再次对其设置值，当CountDownLatch使用完毕后，它不能再次被使用。

4. CountDownLatch 相常见面试题

1. 解释一下 CountDownLatch 概念？
2. CountDownLatch 和 CyclicBarrier 的不同之处？
3. 给出一些 CountDownLatch 使用的例子？
4. CountDownLatch 类中主要的方法？

五.CyclicBarrier(循环栅栏)

CyclicBarrier 和 CountDownLatch 非常类似，它也可以实现线程间的技术等待，但是它的功能比CountDownLatch 更加复杂和强大。主要应用场景和 CountDownLatch 类似。

CountDownLatch的实现是基于AQS的，而CycliBarrier是基于 ReentrantLock(ReentrantLock也属于AQS同步器)和 Condition 的.

CyclicBarrier 的字面意思是可循环使用（Cyclic）的屏障（Barrier）。它要做的事情是，让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续干活。CyclicBarrier 默认的构造方法是 CyclicBarrier(int parties) ，其参数表示屏障拦截的线程数量，每个线程调用 await 方法告诉 CyclicBarrier 我已经到达了屏障，然后当前线程被阻塞。

再来看一下它的构造函数：

public CyclicBarrier(int parties) { 
	this(parties, null); 
}
public CyclicBarrier(int parties, Runnable barrierAction) { 
	if (parties <= 0) throw new IllegalArgumentException(); 
	this.parties = parties; 
	this.count = parties; 
	this.barrierCommand = barrierAction; 
}
1
2
3
4
5
6
7
8
9

其中，parties就代表了有拦截的线程的数量，当拦截的线程数量达到这个值的时候就打开栅栏，让所有线程通过。

1. CyclicBarrier的应用场景

CyclicBarrier可以用于多线程计算数据，最后合并计算结果的应用场景。比如我们用一个Excel保存了用户所有银行流水，每个Sheet保存一个帐户近一年的每笔银行流水，现在需要统计用户的日均银行流水，先用多线程处理每个 sheet里的银行流水，都执行完之后，得到每个sheet的日均银行流水，最后，再用barrierAction用这些线程的计算结果，计算出整个 Excel 的日均银行流水。

2.CyclicBarrier的使用示例

示例 1：

/**
 *
 * @author Snailclimb 
 * @date 2018年10月1日 
 * @Description: 测试 CyclicBarrier 类中带参数的 await() 方法 
*/
public class CyclicBarrierExample2 { 
	// 请求的数量 
	private static final int threadCount = 550; 
	// 需要同步的线程数量 
	private static final CyclicBarrier cyclicBarrier = new CyclicBarrier(5); 
	public static void main(String[] args) throws InterruptedException { 
		// 创建线程池
		 ExecutorService threadPool = Executors.newFixedThreadPool(10); 
		 
		 for (int i = 0; i < threadCount; i++) { 
		 	final int threadNum = i; Thread.sleep(1000); 
		 	threadPool.execute(() -> { 
		 		try { 
		 			test(threadNum); 
		 		} catch (InterruptedException e) { 
		 		// TODO Auto-generated catch block
		 		e.printStackTrace(); 
		 	} catch (BrokenBarrierException e) { 
		 		// TODO Auto-generated catch block 
		 		e.printStackTrace(); 
		 	} 
		 }); 
		}
		threadPool.shutdown(); 
	}
	public static void test(int threadnum) throws InterruptedException, 
BrokenBarrierException { 
		System.out.println("threadnum:" + threadnum + "is ready"); 
		try { 
		/**等待60秒，保证子线程完全执行结束*/ 
		cyclicBarrier.await(60, TimeUnit.SECONDS); 
	} catch (Exception e) { 
		System.out.println("-----CyclicBarrierException------"); 
	}
	System.out.println("threadnum:" + threadnum + "is finish"); 
	} 
	
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

运行结果，如下：

threadnum:0is ready 
threadnum:1is ready 
threadnum:2is ready 
threadnum:3is ready 
threadnum:4is ready 
threadnum:4is finish 
threadnum:0is finish 
threadnum:1is finish 
threadnum:2is finish 
threadnum:3is finish 
threadnum:5is ready 
threadnum:6is ready 
threadnum:7is ready 
threadnum:8is ready 
threadnum:9is ready 
threadnum:9is finish 
threadnum:5is finish 
threadnum:8is finish 
threadnum:7is finish 
threadnum:6is finish 
......
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

可以看到当线程数量也就是请求数量达到我们定义的 5 个的时候， await 方法之后的方法才被执行。

另外，CyclicBarrier 还提供一个更高级的构造函数 CyclicBarrier(int parties, Runnable barrierAction) ，用于在线程到达屏障时，优先执行 barrierAction ，方便处理更复杂的业务场景。示例代码如下：

/**
 *
 * @author SnailClimb 
 * @date 2018年10月1日 
 * @Description: 新建 CyclicBarrier 的时候指定一个 Runnable 
*/
public class CyclicBarrierExample3 { 
	// 请求的数量 
	private static final int threadCount = 550; 
	// 需要同步的线程数量 
	private static final CyclicBarrier cyclicBarrier = new CyclicBarrier(5, () -> 
{ 
	System.out.println("------当线程数达到之后，优先执行------"); 
}); 

public static void main(String[] args) throws InterruptedException { 
	// 创建线程池 
	ExecutorService threadPool = Executors.newFixedThreadPool(10); 
	
	for (int i = 0; i < threadCount; i++) { 
		final int threadNum = i; 
		Thread.sleep(1000); 
		threadPool.execute(() -> { 
			try { test(threadNum); 
		} catch (InterruptedException e) { 
			// TODO Auto-generated catch block 
			e.printStackTrace(); 
		} catch (BrokenBarrierException e) { 
			// TODO Auto-generated catch block 
			e.printStackTrace(); 
		} 
	}); 
}
	threadPool.shutdown(); 
}
public static void test(int threadnum) throws InterruptedException, 
BrokenBarrierException { 
	System.out.println("threadnum:" + threadnum + "is ready"); 
	cyclicBarrier.await(); 
	System.out.println("threadnum:" + threadnum + "is finish"); 
	} 
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

运行结果，如下：

threadnum:0is ready 
threadnum:1is ready 
threadnum:2is ready 
threadnum:3is ready 
threadnum:4is ready 
------当线程数达到之后，优先执行------ 
threadnum:4is finish 
threadnum:0is finish 
threadnum:2is finish 
threadnum:1is finish 
threadnum:3is finish
threadnum:5is ready 
threadnum:6is ready 
threadnum:7is ready 
threadnum:8is ready 
threadnum:9is ready 
------当线程数达到之后，优先执行------ 
threadnum:9is finish 
threadnum:5is finish 
threadnum:6is finish 
threadnum:8is finish 
threadnum:7is finish 
......
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

3. CyclicBarrier 源码分析

当调用CyclicBarrier对象调用await()方法时，实际上调用的是dowait(false, 0L)方法。await() 方法就像树立起一个栅栏的行为一样，将线程挡住了，当拦住的线程数量达到parties的值时，栅栏才会打开，线程才得以通过执行。

public int await() throws InterruptedException, BrokenBarrierException { 
	try {
		return dowait(false, 0L); 


1
2
3
4
5