(10);
Producer2 producer = new Producer2(integers);
Thread thread = new Thread(producer);
thread.start();
Thread.sleep(1000);
// 假设消费者想通知生产者停止向队列添加元素
// 通过interrupt的方式终止线程
thread.interrupt();
}
}
class Producer2 implements Runnable {
private ArrayBlockingQueue arrayBlockingQueue;
public Producer2(ArrayBlockingQueue arrayBlockingQueue) {
this.arrayBlockingQueue = arrayBlockingQueue;
}
@Override
public void run() {
try {
while (true) {
arrayBlockingQueue.put(new Random().nextInt(10));
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
System.out.println("producer quit");
}
}
}
```
### 3.3 重要函数源码解析
- interrupt方法
- 判断是否已被中断相关的方法
- static boolean interrupted()
- boolean isInterrupted()
- Thread.interrupted()的目标对象
```java
/**
* Interrupts this thread.
*
* Unless the current thread is interrupting itself, which is
* always permitted, the {@link #checkAccess() checkAccess} method
* of this thread is invoked, which may cause a {@link
* SecurityException} to be thrown.
*
*
If this thread is blocked in an invocation of the {@link
* Object#wait() wait()}, {@link Object#wait(long) wait(long)}, or {@link
* Object#wait(long, int) wait(long, int)} methods of the {@link Object}
* class, or of the {@link #join()}, {@link #join(long)}, {@link
* #join(long, int)}, {@link #sleep(long)}, or {@link #sleep(long, int)},
* methods of this class, then its interrupt status will be cleared and it
* will receive an {@link InterruptedException}.
*
*
If this thread is blocked in an I/O operation upon an {@link
* java.nio.channels.InterruptibleChannel InterruptibleChannel}
* then the channel will be closed, the thread's interrupt
* status will be set, and the thread will receive a {@link
* java.nio.channels.ClosedByInterruptException}.
*
*
If this thread is blocked in a {@link java.nio.channels.Selector}
* then the thread's interrupt status will be set and it will return
* immediately from the selection operation, possibly with a non-zero
* value, just as if the selector's {@link
* java.nio.channels.Selector#wakeup wakeup} method were invoked.
*
*
If none of the previous conditions hold then this thread's interrupt
* status will be set.
*
* Interrupting a thread that is not alive need not have any effect.
*
* @throws SecurityException
* if the current thread cannot modify this thread
*
* @revised 6.0
* @spec JSR-51
*/
public void interrupt() {
if (this != Thread.currentThread())
checkAccess();
synchronized (blockerLock) {
Interruptible b = blocker;
if (b != null) {
interrupt0(); // Just to set the interrupt flag
b.interrupt(this);
return;
}
}
interrupt0();
}
```
#### 3.3.1 如何分析native方法
- 进github/gitee 查找openjdk（https://gitee.com/ckl111/openjdk-jdk8u）
- 进行文件搜索，Thread.c
https://gitee.com/ckl111/openjdk-jdk8u/blob/master/jdk/src/share/native/java/lang/Thread.c
```
static JNINativeMethod methods[] = {
{"start0", "()V", (void *)&JVM_StartThread},
{"stop0", "(" OBJ ")V", (void *)&JVM_StopThread},
{"isAlive", "()Z", (void *)&JVM_IsThreadAlive},
{"suspend0", "()V", (void *)&JVM_SuspendThread},
{"resume0", "()V", (void *)&JVM_ResumeThread},
{"setPriority0", "(I)V", (void *)&JVM_SetThreadPriority},
{"yield", "()V", (void *)&JVM_Yield},
{"sleep", "(J)V", (void *)&JVM_Sleep},
{"currentThread", "()" THD, (void *)&JVM_CurrentThread},
{"countStackFrames", "()I", (void *)&JVM_CountStackFrames},
{"interrupt0", "()V", (void *)&JVM_Interrupt},
{"isInterrupted", "(Z)Z", (void *)&JVM_IsInterrupted},
{"holdsLock", "(" OBJ ")Z", (void *)&JVM_HoldsLock},
{"getThreads", "()[" THD, (void *)&JVM_GetAllThreads},
{"dumpThreads", "([" THD ")[[" STE, (void *)&JVM_DumpThreads},
{"setNativeName", "(" STR ")V", (void *)&JVM_SetNativeThreadName},
};
```
上述代码能看到native代码中方法对应c++代码的方法。
https://github.com/openjdk-mirror/jdk7u-hotspot
搜索JVM_Interrupt
定位到 https://github.com/openjdk-mirror/jdk7u-hotspot/blob/50bdefc3afe944ca74c3093e7448d6b889cd20d1/src/share/vm/prims/jvm.cpp
```c++
JVM_ENTRY(void, JVM_Interrupt(JNIEnv* env, jobject jthread))
JVMWrapper("JVM_Interrupt");
// Ensure that the C++ Thread and OSThread structures aren't freed before we operate
oop java_thread = JNIHandles::resolve_non_null(jthread);
MutexLockerEx ml(thread->threadObj() == java_thread ? NULL : Threads_lock);
// We need to re-resolve the java_thread, since a GC might have happened during the
// acquire of the lock
JavaThread* thr = java_lang_Thread::thread(JNIHandles::resolve_non_null(jthread));
if (thr != NULL) {
Thread::interrupt(thr);
}
JVM_END
```
定位到Thread.cpp（https://gitee.com/czr27/openjdk8-hotspot/blob/master/runtime/thread.cpp）
```c++
void Thread::interrupt(Thread* thread) {
trace("interrupt", thread);
debug_only(check_for_dangling_thread_pointer(thread);)
os::interrupt(thread);
}
```
定位到os_linux.cpp （https://github.com/openjdk-mirror/jdk7u-hotspot/blob/50bdefc3afe944ca74c3093e7448d6b889cd20d1/src/os/linux/vm/os_linux.cpp）
```c++
void os::interrupt(Thread* thread) {
assert(Thread::current() == thread || Threads_lock->owned_by_self(),
"possibility of dangling Thread pointer");
OSThread* osthread = thread->osthread();
if (!osthread->interrupted()) {
// 设置interrupted状态位true
osthread->set_interrupted(true);
// More than one thread can get here with the same value of osthread,
// resulting in multiple notifications. We do, however, want the store
// to interrupted() to be visible to other threads before we execute unpark().
OrderAccess::fence();
// _SleepEvent就是对应Thread.sleep方法
ParkEvent * const slp = thread->_SleepEvent ;
if (slp != NULL) slp->unpark() ;
}
// For JSR166. Unpark even if interrupt status already was set
if (thread->is_Java_thread())
// 对应LockSupport.park
((JavaThread*)thread)->parker()->unpark();
// _ParkEvent对应synchronized同步块以及Object.wait方法
ParkEvent * ev = thread->_ParkEvent ;
if (ev != NULL) ev->unpark() ;
}
```
#### 3.3.2 interrupted方法
```java
/**
* Tests whether the current thread has been interrupted. The
* interrupted status of the thread is cleared by this method. In
* other words, if this method were to be called twice in succession, the
* second call would return false (unless the current thread were
* interrupted again, after the first call had cleared its interrupted
* status and before the second call had examined it).
*
*
A thread interruption ignored because a thread was not alive
* at the time of the interrupt will be reflected by this method
* returning false.
*
* @return true if the current thread has been interrupted;
* false otherwise.
* @see #isInterrupted()
* @revised 6.0
*/
public static boolean interrupted() {
// 这里拿到的是当前执行的线程的中断标志位，并且会把标志位重置
return currentThread().isInterrupted(true);
}
```
#### 3.3.3 isInterrupted方法
```java
/**
* Tests whether this thread has been interrupted. The interrupted
* status of the thread is unaffected by this method.
*
*
A thread interruption ignored because a thread was not alive
* at the time of the interrupt will be reflected by this method
* returning false.
*
* @return true if this thread has been interrupted;
* false otherwise.
* @see #interrupted()
* @revised 6.0
*/
public boolean isInterrupted() {
// 不清除中断标志位
return isInterrupted(false);
}
/**
* Tests if some Thread has been interrupted. The interrupted state
* is reset or not based on the value of ClearInterrupted that is
* passed.
*/
private native boolean isInterrupted(boolean ClearInterrupted);
```
#### 3.3.4 interrupted方法小测试
```java
package threadcorekownledage.stopthread;
/**
* 注意Thread.interrupted()方法的目标对象是"当前线程"，而不管本方法来自哪个对象
*/
public class RightWayInterrupted {
public static void main(String[] args) throws InterruptedException {
Thread thread = new Thread(() -> {
for (; ; ) {
}
});
// 启动线程
thread.start();
// 设置中断标志
thread.interrupt();
// 获取中断标志
System.out.println("isInterrupted:" + thread.isInterrupted()); // true
// 获取中断标志并重置
System.out.println("isInterrupted:" + thread.interrupted()); // false 实际执行的是 currentThread().isInterrupted(true) 这里也就是获取main线程的中断标志位
// 获取中断标志并重置
System.out.println("isInterrupted:" + Thread.interrupted()); // false 同上也是获取main线程的中断标志位
// 获取中断标志
System.out.println("isInterrupted:" + thread.isInterrupted()); // true
thread.join();
System.out.println("main thread is over");
}
}
```
## 4. 线程的声明周期
### 4.1 线程的6个状态
- New
- Runnable
- Blocked
- Waiting
- Timed Waiting
- Terminated
### 4.2 线程状态转换
其中New->Runnable->Terminated之间是不可逆的。Block、Waiting、Timed_Waiting之间不能直接转换，必须同Runnable这个中间状态进行转换。
#### 4.2.1 NEW RUNNABLE TERMINATED 状态演示
```java
package threadcorekownledage.threadstatus;
/**
* 演示NEW RUNNABLE TERMINATED这三个线程状态
*/
public class NewRunnableTerminated {
public static void main(String[] args) throws InterruptedException {
Thread thread = new Thread(() -> {
System.out.println(Thread.currentThread().getState()); // RUNNABLE
});
System.out.println(thread.getState()); // NEW
thread.start();
thread.join();
System.out.println(thread.getState()); // TERMINATED
}
}
```
#### 4.2.2 BLOCKED WAITING TIMED_WAITING 状态演示
```java
package threadcorekownledage.threadstatus;
/**
* 演示BLOCKED WAITING TIMED_WAITING这三个线程状态
*/
public class BlockedWaitingTimedwaiting implements Runnable{
public static void main(String[] args) throws InterruptedException {
BlockedWaitingTimedwaiting runnable = new BlockedWaitingTimedwaiting();
Thread thread1 = new Thread(runnable);
thread1.start();
Thread thread2 = new Thread(runnable);
thread2.start();
System.out.println("thread1 state: " + thread1.getState()); // TIMED_WAITING
System.out.println("thread2 state: " + thread2.getState()); // BLOCKED
Thread.sleep(1300);
System.out.println("thread1 state: " + thread1.getState()); // WAITING
}
public synchronized void blockedMethod() {
try {
Thread.sleep(1000);
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Override
public void run() {
blockedMethod();
}
}
```
## 5. Thread和Object类中线程相关的方法
方法概览
| 类 | 方法名 | 简介 |
| ------ | --------------------------- | ------------------------------------------------------------ |
| Thread | sleep相关： | 本表的“相关”，指的是重载方法，也就是方法名相同，但是参数不同，例如sleep有多个方法，只是参数不同，实际作用大同小异。 |
| | join | 等待其他线程执行完毕 |
| | yield相关 | 放弃已经获取到的CPU资源 |
| | currentThread | 获取当前执行线程的引用 |
| | start，run相关 | 启动线程相关 |
| | interrupt相关 | 中断线程 |
| | stop(),suspend,resume()相关 | 已废弃 |
| Object | wait/notify/notifyAll相关 | 让线程暂时休眠和唤醒 |
### 5.1 wait、notify、notifyAll方法详解
#### 5.1.1 wait的作用和用法
作用：阻塞当前线程并释放monitor锁
简单用法：
```java
package threadcorekownledage.blockcoremethod;
/**
* wait和notify基本用法
*/
public class Wait {
public static void main(String[] args) throws InterruptedException {
new Thread1().start();
Thread.sleep(100);
new Thread2().start();
}
public static Object object = new Object();
static class Thread1 extends Thread {
@Override
public void run() {
synchronized (object) {
System.out.println(Thread.currentThread().getName() + " start");
try {
object.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + " 获取到了锁");
}
}
}
static class Thread2 extends Thread {
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + "准备获取锁进入同步代码块");
synchronized (object) {
System.out.println(Thread.currentThread().getName() + "获取到了锁，准备执行notify。这里能获取到锁也说明了wait方法调用后会释放锁");
object.notify();
System.out.println(Thread.currentThread().getName() + "执行了notify。注意：synchronized代码块执行完毕线程才会被唤醒");
}
}
}
}
```
线程被唤醒的四种情况：
- 另一个线程调用了这个对象的notify方法且刚好唤醒的是本线程
- 另一个线程调用了这个对象的notifyAll方法
- 过了wait(long timeout)设置的超时时间，如果传入0就是永久等待
- 线程自身调用了interrupt方法
#### 5.1.2 notify和notifyAll的区别
notify只会唤醒一个等待该monitor锁线程，notifyAll会唤醒所有等待该monitor锁的所有线程。
```java
package threadcorekownledage.blockcoremethod;
/**
* 3个线程，线程1和线程2首先被阻塞，线程3唤醒它们。notify notifyAll
*/
public class WaitNotifyAll implements Runnable{
private static final Object resouceA = new Object();
public static void main(String[] args) throws InterruptedException {
WaitNotifyAll r = new WaitNotifyAll();
new Thread(r).start();
new Thread(r).start();
Thread.sleep(100);
new Thread(()->{
synchronized (resouceA) {
System.out.println(Thread.currentThread().getName() + " 获取到resouceA，并唤醒其他线程");
// resouceA.notify();
resouceA.notifyAll();
}
}).start();
}
@Override
public void run() {
synchronized (resouceA) {
try {
System.out.println(Thread.currentThread().getName() + " 获取到resouceA，阻塞当前线程并释放resouceA");
resouceA.wait();
System.out.println(Thread.currentThread().getName() + " 被唤醒");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
```
notifyAll把线程1和线程2都唤醒了，但是notify只唤醒了一个线程，另一个线程仍然处于等待
```
"Thread-1" #14 prio=5 os_prio=31 tid=0x00007ff9f4865000 nid=0xa603 in Object.wait() [0x00007000054c1000]
java.lang.Thread.State: WAITING (on object monitor)
at java.lang.Object.wait(Native Method)
- waiting on <0x000000076adf2dc0> (a java.lang.Object)
at java.lang.Object.wait(Object.java:502)
at threadcorekownledage.blockcoremethod.WaitNotifyAll.run(WaitNotifyAll.java:31)
- locked <0x000000076adf2dc0> (a java.lang.Object)
at java.lang.Thread.run(Thread.java:748)
```
wait/notify的原理、特点
用wait/notify实现生产者消费者模式
为什么要使用生产者和消费者模式？
消息（数据）的生产和消费的速率的会有所不同，会出现消息的消费方不知道何时有数据可以进行消费，而消息的生产方不知道消费放是否能够消费掉已经生产的消息。而生产者和消费者模式将解决这种信息不对称的问题。该模式将采用一个阻塞队列用来存储消息。
#### 5.1.3 wait高频面试题
- 用程序实现**两个线程交替**打印0-100的奇偶数
```java
package threadcorekownledage.blockcoremethod;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 两个线程交替打印奇偶
*/
public class PrintOddEvenByturn {
public static void main(String[] args) {
// printOddEventByturnUseSynchorized();
printOddEventByturnUseWaitAndNotify();
}
private static void printOddEventByturnUseSynchorized() {
AtomicInteger i = new AtomicInteger();
new Thread(()->{
// 奇数
while (i.get() < 100) {
synchronized (i) {
if ((i.get() & 1) == 1 ) {
System.out.println(Thread.currentThread().getName() + " print " + i.getAndIncrement());
}
}
}
}).start();
new Thread(() -> {
// 偶数
while (i.get() <= 100) {
synchronized (i) {
if ((i.get() & 1) == 0) {
System.out.println(Thread.currentThread().getName() + " print " + i.getAndIncrement());
}
}
}
}).start();
}
public static void printOddEventByturnUseWaitAndNotify() {
AtomicInteger i = new AtomicInteger();
Runnable runnable = () -> {
while (i.get() < 101) {
synchronized (i) {
System.out.println(Thread.currentThread().getName() + " print " + i.getAndIncrement());
// 唤醒另一个线程进行打印
i.notify();
if (i.get() < 101) {
try {
// 释放锁，进入阻塞
i.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
};
new Thread(runnable).start();
new Thread(runnable).start();
}
}
```
- 手写**生产者消费者设计模式**
- 使用wait/notify的方法实现
```java
package threadcorekownledage.producerconsumermodel;
import java.util.Date;
import java.util.LinkedList;
public class ProducerConsumerModel {
public static void main(String[] args) {
MyBlockingQueue myBlockingQueue = new MyBlockingQueue();
new Thread(new Producer(myBlockingQueue)).start();
new Thread(new Consumer(myBlockingQueue)).start();
}
}
class Producer implements Runnable {
private MyBlockingQueue storage;
public Producer(MyBlockingQueue myBlockingQueue) {
this.storage = myBlockingQueue;
}
@Override
public void run() {
for (int i = 0; i < 100; i++) {
storage.put(new Date());
}
}
}
class Consumer implements Runnable {
private MyBlockingQueue storage;
public Consumer(MyBlockingQueue myBlockingQueue) {
this.storage = myBlockingQueue;
}
@Override
public void run() {
for (int i = 0; i < 100 ; i++) {
storage.take();
}
}
}
class MyBlockingQueue {
private int maxSize;
private LinkedList storage;
public MyBlockingQueue() {
this.maxSize = 10;
this.storage = new LinkedList<>();
}
public synchronized void put(Date integer) {
// 如果队列满了，所有put线程进入阻塞
while (storage.size() == maxSize) {
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
storage.add(integer);
System.out.println("队列中有了" + storage.size() + "个元素");
// 通知唤醒其他线程
notify();
}
public synchronized void take() {
// 当队列为空，则阻塞所有take线程
while (storage.size() == 0) {
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println("获取到了" + storage.poll() + "元素，还剩下" + storage.size());
notify();
}
}
```
- 使用阻塞队列实现
```java
package threadcorekownledage.producerconsumermodel;
import java.util.concurrent.ArrayBlockingQueue;
/**
* 使用阻塞队列实现生产者消费者模式
*/
public class ProducerConsumerModel2 {
public static void main(String[] args) {
ArrayBlockingQueue queue = new ArrayBlockingQueue(10);
new Thread(new Producer2(queue)).start();
new Thread(new Consumer2(queue)).start();
}
}
class Producer2 implements Runnable {
private ArrayBlockingQueue queue;
public Producer2(ArrayBlockingQueue queue) {
this.queue = queue;
}
@Override
public void run() {
try {
for (int i = 0; i < 1000; i++) {
System.out.println(Thread.currentThread().getName() + " 生产 " + i);;
queue.put(i);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
class Consumer2 implements Runnable {
private ArrayBlockingQueue queue;
public Consumer2(ArrayBlockingQueue queue) {
this.queue = queue;
}
@Override
public void run() {
try {
for (int i = 0; i < 1000; i++) {
System.out.println(Thread.currentThread().getName() + " 消费掉 " + queue.take());;
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
```
- 为什么wait()需要在**同步代码块内**使用，而sleep()不需要
- **还没想好怎么回答**
- 为什么线程通信的方法wait(),notify()和notifyAll()被定义在Object类里？而Sleep定义在Thread类里？
- wait(),notify()和notifyAll()是琐级别的操作。
### 5.2 sleep方法详解
作用：让线程进入waiting状态，释放CPU资源但不释放锁，直到规定时间后再执行，休眠期间如果被中断，会抛出异常并清除中断状态。
特点：不释放锁，包括synchronized和lock。这与wait不同。
#### 5.2.1 wait和sleep方法的异同
- 相同
- wait和sleep都能使线程阻塞，对应线程状态是waiting或者time_waiting
- wait和sleep都可以响应中断Thread.interrupt()
- 不同
- wait方法的执行必须在同步方法中进行，而sleep则不需要
- 在同步方法里执行sleep方法时，不会释放monitor锁，但是wait方法会释放monitor锁
- sleep方法短暂休眠后会主动退出阻塞，而没有指定时间和wait方法则需要被其他线程中断后才能退出阻塞
- wait、notify、notifyAll都是Object类的方法，sleep和yield时Thread类的方法
### 5.3 join方法
作用：加入新的线程，也就意味着需要等待新的线程执行完，再继续main线程后续的操作。（main线程等待其他线程执行完毕）
注意：JUC中的CountDownLatch或CyclicBarrier类可以完成join。优先考虑使用JUC中现成的工具类。
#### 5.3.1 join的简单用法
```java
package threadcorekownledage.blockcoremethod;
public class Join {
public static void main(String[] args) throws InterruptedException {
Thread thread1 = new Thread(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + "执行完毕");
});
Thread thread2 = new Thread(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + "执行完毕");
});
thread1.start();
thread2.start();
System.out.println(Thread.currentThread().getName() + "开始等待子线程执行完毕");
thread1.join();
thread2.join();
System.out.println("所有子线程执行完毕");
}
}
```
#### 5.3.2 join遇到中断的处理
```java
package threadcorekownledage.blockcoremethod;
public class JoinInterrupt {
public static void main(String[] args) {
Thread thread = Thread.currentThread();
Thread thread1 = new Thread(() -> {
try {
thread.interrupt();
Thread.sleep(5000);
System.out.println(Thread.currentThread().getName() + "执行完毕");
} catch (InterruptedException e) {
e.printStackTrace();
}
});
thread1.start();
try {
thread1.join();
} catch (InterruptedException e) {
System.out.println(Thread.currentThread().getName() + "线程中断了");
// 将中断标志传递给子线程，否则子线程会正常执行完
thread1.interrupt();
e.printStackTrace();
}
System.out.println("子线程运行完毕");
}
}
```
#### 5.3.3 探索join过程中main线程的状态
```java
package threadcorekownledage.blockcoremethod;
/**
* 先join再 mainThread.getState
*/
public class JoinThreadState {
public static void main(String[] args) throws InterruptedException {
Thread mainThread = Thread.currentThread();
Thread thread1 = new Thread(() -> {
try {
Thread.sleep(1000);
// 这里获取到的main线程的状态为WAIT
System.out.println(mainThread.getName() + "线程状态为：" + mainThread.getState());
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + "执行完毕");
});
thread1.start();
System.out.println(Thread.currentThread().getName() + "开始等待子线程执行完毕");
thread1.join();
System.out.println("所有子线程执行完毕");
}
}
```
#### 5.3.4 join源码分析
##### 5.3.4.1 join源码分析
```java
public final synchronized void join(long millis)
throws InterruptedException {
long base = System.currentTimeMillis();
long now = 0;
if (millis < 0) {
throw new IllegalArgumentException("timeout value is negative");
}
if (millis == 0) {
while (isAlive()) {
// 实际调用的是wait方法
wait(0);
}
} else {
while (isAlive()) {
long delay = millis - now;
if (delay <= 0) {
break;
}
wait(delay);
now = System.currentTimeMillis() - base;
}
}
}
```
join方法中并没有notify这样的唤醒方法，那么线程是怎么被唤醒的呢？实际上每个Thread在执行run方法后会自动的调用notifyAll方法。
```c++
void JavaThread::exit(booldestory_vm,ExitTypeexit_type);
// JavaThread::exit中调用了
static void ensure_join(JavaThread*thread){
Handle threadObj(thread, thread->threadObj());
ObjeckLocker lock(threadObj,thread);
thread->clear_pending_exception();
java_lang_Thread::set_thread_status(threadObj(),java_lang_Thread::TERMINATED);
java_lang_Thread::set_thread(threadObj(),NULL);
lock.notify_all(thread);// 这里执行了notify_all,进行了wait的唤醒
thread->clear_pending_exception();
}
```
##### 5.3.4.2 join的等价写法
```java
package threadcorekownledage.blockcoremethod;
public class JoinPrinciple {
public static void main(String[] args) throws InterruptedException {
Thread thread1 = new Thread(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + "执行完毕");
});
thread1.start();
System.out.println(Thread.currentThread().getName() + "开始等待子线程执行完毕");
// thread1.join(); 等价写法
// 主线程拿到锁执行wait方法。子线程执行完后调用notifyAll方法将唤醒主线程
synchronized (thread1) {
thread1.wait();
}
System.out.println("所有子线程执行完毕");
}
}
```
### 5.4 yield方法
作用：让线程释放所占用CPU时间片，但是JVM不保证遵循，CPU资源比较充裕的话可能会忽略yield请求。
## 6. 线程各属性
线程各属性纵览
| 属性名称 | 用途 | 注意事项 |
| -------------------------- | ------------------------------------------------------------ | --------------------------------------------------------- |
| 编号（ID） | 每个线程有自己的ID，用于标识不同的线程 | 被后续创建的线程使用；唯一性；不允许被修改 |
| 名称（Name） | 让用户或程序员在开发、调试和运维过程中，更容易区分每个不同的线程以便于问题定位 | 尽量取清晰有意义的名字 |
| 是否是守护线程（isDaemon） | true代表线程是【守护线程】，false表示线程为非守护线程，也就是【用户线程】 | 该属性默认继承父线程；也可以通过setDaemon设置 |
| 优先级（Priority） | 优先级这个属性的目的是告诉线程调度器，用户希望哪些线程相对多运行、哪些少运行。通常使用默认的优先级。这个属性生效取决于操作系统的支持。优先级也可能被操作系统改变。 | 默认和父线程的优先级相等，共10个等级，默认值为5；不应依赖 |
## 7. 线程异常处理
思考题
- Java异常体系图
- 实际工作中，如何全局处理异常？为什么要做全局处理？不处理行不行？
线程的未捕获异常UncaughtException应该如何处理？
- 为什么需要UncaughtExceptionHandler？
- 主线程可以轻松发现异常，子线程却不行
- 子线程异常无法用传统方法捕获
- 不能直接捕获会有不好的后果
- 两种解决办法
- 方案一（不推荐）：手动在每个run方法里进行try catch。
- 方案二（推荐）：利用Uncaught ExceptionHandler
以下就是子线程异常处理默认的实现
```java
public class ThreadGroup implements Thread.UncaughtExceptionHandler {
public void uncaughtException(Thread t, Throwable e) {
if (parent != null) {
parent.uncaughtException(t, e);
} else {
Thread.UncaughtExceptionHandler ueh =
Thread.getDefaultUncaughtExceptionHandler();
if (ueh != null) {
ueh.uncaughtException(t, e);
} else if (!(e instanceof ThreadDeath)) {
System.err.print("Exception in thread \""
+ t.getName() + "\" ");
e.printStackTrace(System.err);
}
}
}
}
```
几种实现方式
- 给程序**统一**设置
- 给每个线程**单独**设置
- 给**线程池**设置
```java
package threadcorekownledage.uncaughtexception;
import java.util.logging.Level;
import java.util.logging.Logger;
/**
* 自定义UncaughtExceptionHandler
*/
public class MyUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {
private String name;
public MyUncaughtExceptionHandler(String name) {
this.name = name;
}
@Override
public void uncaughtException(Thread t, Throwable e) {
// 统一异常逻辑处理
Logger logger = Logger.getAnonymousLogger();
logger.log(Level.WARNING, "线程异常，终止了 " + t.getName(), e);
System.out.println(name + "捕获了异常");
}
}
```
```java
package threadcorekownledage.uncaughtexception;
public class UseMyUncaughtExceptionHandler implements Runnable {
public static void main(String[] args) throws InterruptedException {
// try catch捕获不到子线程抛出的异常
// 设置异常处理方法
Thread.setDefaultUncaughtExceptionHandler(new MyUncaughtExceptionHandler("MyUncaughtExceptionHandler"));
try {
new Thread(new UseMyUncaughtExceptionHandler(), "mythread-1").start();
Thread.sleep(200);
new Thread(new UseMyUncaughtExceptionHandler(), "mythread-2").start();
Thread.sleep(200);
new Thread(new UseMyUncaughtExceptionHandler(), "mythread-3").start();
Thread.sleep(200);
new Thread(new UseMyUncaughtExceptionHandler(), "mythread-4").start();
} catch (RuntimeException e) {
System.out.println("catch RuntimeException");
}
}
@Override
public void run() {
throw new RuntimeException();
}
}
```
## 8. 多线程带来的问题
### 8.1 线程安全问题
什么是线程安全
当多个线程访问一个对象时，如果**不考虑**这些线程在运行时环境的调度和交替执行，也**不需要进行额外的同步**，或者在调用进行任何其他的协调操作，调用这个对象的行为都可以获得正确的结果，那这个对象是线程安全的。
什么时候会出现线程安全问题
#### 8.1.1 数据竞争
如果两个或者多个任务在临界段之外对一个共享变量进行写入操作，也就是说没有使用任何同步机制，那么应用程序可能存在**数据竞争**（也叫做**竞争条件**）。
在这些情况下，应用程序的最终结果可能取决于任务的执行顺序。
```java
package part1.concurrent_matter;
public class ResourceCompete {
private int ii;
public void add(int add) {
ii += add;
}
public static void main(String[] args) throws InterruptedException {
ResourceCompete resourceCompete = new ResourceCompete();
Thread thread1 = new Thread(() -> {
for (int i = 0; i < 1000; i++) {
resourceCompete.add(10);
}
});
Thread thread2 = new Thread(() -> {
for (int i = 0; i < 1000; i++) {
resourceCompete.add(10);
}
});
thread1.start();
thread2.start();
thread1.join();
thread2.join();
// 这里得到的值可能会小于2000
System.out.println(resourceCompete.ii);
}
}
```
假设有两个不同的任务执行了同一个add方法。add方法不是原子的，ResourceCompete也不是线程安全的。
#### 8.1.2 死锁
当两个（或多个）任务正在等待必须有另一线程释放的某个共享资源，而该线程又正在等待必须由前述任务之一释放的另一共享资源时，并发应用程序就出现了死锁。当系统中同时出现如下四种条件时，就会导致这种情形。我们将其称为Coffman条件。
- 互斥：死锁中涉及的资源必须是不可共享的。一次只有一个任务可以使用该资源
- 占用并等待条件：一个任务在占有某一互斥的资源时由请求另一互斥的资源。当它在等待时，不会释放任何资源
- 不可剥夺：资源只能被那些持有它们的任务释放
- 循环等待：任务1正在等待任务2所占用的资源，而任务2又正在等待任务3所占有的资源，以此类推，最终任务n又在等待又任务1所占的资源，这样就出现了循环等待
有些机制可以用来避免死锁：
- 忽略它们：这是最常用的机制。你可以假设自己的系统绝对不会出现死锁。如果发生死锁，结果就是你可以停止应用程序并且重新执行它
- 检测：系统中有一项专门分析系统状态的任务，可以检测是否发生死锁，如果检测到了死锁，可以采取一些措施来修复该问题，例如，结束某个任务或者强制释放某一资源。
- 预防：如果你想防止系统出现死锁，就必须预防Coffman条件中一条或者多条出现
- 规避：如果你可以在某一任务执行之前得倒该任务所使用资源的相关信息，那么死锁是可以规避的。当一个任务要开始执行时，你可以对系统中空闲的资源和任务所需的资源进行分析，这样就可以判断任务是否能够开始执行。
##### 8.1.2.1 最简单的死锁
```java
package deadlock;
/**
* 最简单的死锁案例
*/
public class MustDeadLock implements Runnable {
int flag = 1;
static Object o1 = new Object();
static Object o2 = new Object();
public static void main(String[] args) {
MustDeadLock r1 = new MustDeadLock();
MustDeadLock r2 = new MustDeadLock();
r1.flag = 1;
r2.flag = 2;
Thread thread1 = new Thread(r1);
Thread thread2 = new Thread(r2);
thread1.start();
thread2.start();
// ThreadMXBean检查死锁
Thread.sleep(1000);
ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean();
long[] deadlockedThreads = threadMXBean.findDeadlockedThreads();
if (deadlockedThreads != null && deadlockedThreads.length > 0) {
for (int i = 0; i < deadlockedThreads.length; i++) {
ThreadInfo threadInfo = threadMXBean.getThreadInfo(deadlockedThreads[i]);
System.out.println("发现死锁" + threadInfo.getThreadName());
}
}
}
@Override
public void run() {
System.out.println("flag=" + flag);
if (flag == 1) {
synchronized (o1) {
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
synchronized (o2) {
System.out.println(Thread.currentThread().getName() + "成功拿到两把锁");
}
}
}
if (flag == 2) {
synchronized (o2) {
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
synchronized (o1) {
System.out.println(Thread.currentThread().getName() + "成功拿到两把锁");
}
}
}
}
}
```
```c++
Found one Java-level deadlock:
=============================
"Thread-1":
waiting to lock monitor 0x00007fcf3c01af68 (object 0x000000076adf1350, a java.lang.Object),
which is held by "Thread-0"
"Thread-0":
waiting to lock monitor 0x00007fcf3c01d8a8 (object 0x000000076adf1360, a java.lang.Object),
which is held by "Thread-1"
Java stack information for the threads listed above:
===================================================
"Thread-1":
at deadlock.MustDeadLock.run(MustDeadLock.java:48)
- waiting to lock <0x000000076adf1350> (a java.lang.Object)
- locked <0x000000076adf1360> (a java.lang.Object)
at java.lang.Thread.run(Thread.java:748)
"Thread-0":
at deadlock.MustDeadLock.run(MustDeadLock.java:35)
- waiting to lock <0x000000076adf1360> (a java.lang.Object)
- locked <0x000000076adf1350> (a java.lang.Object)
at java.lang.Thread.run(Thread.java:748)
Found 1 deadlock.
```
##### 8.1.2.2 多人转账案例
```java
package deadlock;
import java.util.Random;
/**
* 多人转账的时候遇到死锁
*/
public class MultiTransferMoney {
private static final int NUM_ACCOUNTS = 500;
private static final int NUM_MONEY = 1000;
private static final int NUM_ITERATIONS = 1000000;
private static final int NUM_THREAD = 20;
int flag = 1;
public static void main(String[] args) throws InterruptedException {
Random random = new Random();
TransferMoney.Account[] accounts = new TransferMoney.Account[NUM_ACCOUNTS];
for (int i = 0; i < accounts.length; i++) {
accounts[i] = new TransferMoney.Account(NUM_MONEY);
}
class TransferThread extends Thread {
@Override
public void run() {
for (int i = 0; i < NUM_ITERATIONS; i++) {
int fromAcct = random.nextInt(NUM_ACCOUNTS);
int toAcct = random.nextInt(NUM_ACCOUNTS);
int amount = random.nextInt(200);
TransferMoney.Account from = accounts[fromAcct];
TransferMoney.Account to = accounts[toAcct];
synchronized (from) {
synchronized (to) {
if (from.balance - amount < 0) {
System.out.println("余额不足，转账失败！");
return;
}
from.balance -= amount;
to.balance += amount;
System.out.println("转账成功" + amount + "元");
}
}
}
}
}
for (int i = 0; i < NUM_THREAD; i++) {
new TransferThread().start();
}
}
}
```
思路：避免相反的获取锁的顺序
```java
package deadlock;
import java.util.Random;
/**
* 多人转账的时候遇到死锁
*/
public class MultiTransferMoneyFix {
private static final int NUM_ACCOUNTS = 500;
private static final int NUM_MONEY = 1000;
private static final int NUM_ITERATIONS = 1000000;
private static final int NUM_THREAD = 20;
private static Object lock = new Object();
public static void main(String[] args) {
Random random = new Random();
TransferMoney.Account[] accounts = new TransferMoney.Account[NUM_ACCOUNTS];
for (int i = 0; i < accounts.length; i++) {
accounts[i] = new TransferMoney.Account(NUM_MONEY);
}
class TransferThread extends Thread {
@Override
public void run() {
for (int i = 0; i < NUM_ITERATIONS; i++) {
int fromAcct = random.nextInt(NUM_ACCOUNTS);
int toAcct = random.nextInt(NUM_ACCOUNTS);
int amount = random.nextInt(200);
TransferMoney.Account from = accounts[fromAcct];
TransferMoney.Account to = accounts[toAcct];
transferMoney(from, to, amount);
}
}
}
for (int i = 0; i < NUM_THREAD; i++) {
new TransferThread().start();
}
}
public static void transferMoney(TransferMoney.Account from, TransferMoney.Account to, int amount) {
class Helper {
public void transfer() {
if (from.balance - amount < 0) {
System.out.println("余额不足，转账失败！");
return;
}
from.balance -= amount;
to.balance += amount;
System.out.println("转账成功" + amount + "元");
}
}
// 根据对象锁的id大小给定获取锁的顺序
int fromHash = System.identityHashCode(from);
int toHash = System.identityHashCode(to);
if (fromHash < toHash) {
synchronized (from) {
synchronized (to) {
new Helper().transfer();
}
}
} else if (fromHash > toHash) {
synchronized (to) {
synchronized (from) {
new Helper().transfer();
}
}
}
// 如果hash值一样的话，加同步锁
else {
synchronized (lock) {
synchronized (from) {
synchronized (to) {
new Helper().transfer();
}
}
}
}
}
}
```
##### 8.1.2.3 哲学家就餐
要吃饭得先拿起左边的筷子再拿起。要思考的时候先放下右边的筷子再放下左边的筷子
```java
// 伪代码
while(true){
// thinking about life
think();
// prepare to eat
pick_up_left_fork();
pick_up_right_fork();
eat();
put_down_right_fork();
put_down_left_fork();
// not hungry,go to thinking
}
```
解决方案
- 服务员检查（避免策略）
- 改变一个哲学家拿筷子的顺序（避免策略）
- 餐票（避免策略）
- 领导调节（检测与恢复策略）
```java
package deadlock;
/**
* 哲学家就餐问题导致的死锁
*/
public class DiningPhilosophers {
public static class Philosopher implements Runnable {
private Object leftChopstick;
private Object rightChopstick;
public Philosopher(Object leftChopstick, Object rightChopstick) {
this.leftChopstick = leftChopstick;
this.rightChopstick = rightChopstick;
}
@Override
public void run() {
while (true) {
// thinking
try {
doAction("thinking");
// get leftChopstick
synchronized (leftChopstick) {
doAction("get leftChopstick");
//Picked up right chopstick;
synchronized (rightChopstick) {
doAction("get rightChopstick");
doAction("put down rightChopstick");
}
doAction("put down leftChopstick");
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
private void doAction(String action) throws InterruptedException {
System.out.println(Thread.currentThread().getName() + " " + action);
Thread.sleep((long)(Math.random()*10));
}
}
public static void main(String[] args) {
// 定义5个哲学家
Philosopher[] philosophers = new Philosopher[5];
Object[] chopsticks = new Object[philosophers.length];
for (int i = 0; i < chopsticks.length; i++) {
chopsticks[i] = new Object();
}
// 5根筷子
for (int i = 0; i < philosophers.length; i++) {
Object leftChopstick = chopsticks[i];
Object rightChopstick = chopsticks[(i + 1) % chopsticks.length];
// if (i == philosophers.length - 1) {
// philosophers[i] = new Philosopher(rightChopstick, leftChopstick);
// } else {
// philosophers[i] = new Philosopher(leftChopstick, rightChopstick);
// }
philosophers[i] = new Philosopher(leftChopstick, rightChopstick);
new Thread(philosophers[i], "哲学家" + (i + 1) + "号").start();
}
}
}
```
改变一个哲学家拿筷子的顺序
```java
public static void main(String[] args) {
// 定义5个哲学家
Philosopher[] philosophers = new Philosopher[5];
Object[] chopsticks = new Object[philosophers.length];
for (int i = 0; i < chopsticks.length; i++) {
chopsticks[i] = new Object();
}
// 5根筷子
for (int i = 0; i < philosophers.length; i++) {
Object leftChopstick = chopsticks[i];
Object rightChopstick = chopsticks[(i + 1) % chopsticks.length];
// 改变一个哲学家拿筷子的顺序
if (i == philosophers.length - 1) {
philosophers[i] = new Philosopher(rightChopstick, leftChopstick);
} else {
philosophers[i] = new Philosopher(leftChopstick, rightChopstick);
}
new Thread(philosophers[i], "哲学家" + (i + 1) + "号").start();
}
}
```
检测与恢复策略
- 允许发生死锁
- 每次调用锁都记录
- 定时检查“锁的调用链路图”中是否存在环路
- 一旦发现死锁，就用死锁恢复机制进行恢复
- 进程终止：逐个终止线程，直到死锁消除
- 终止顺序：
- 优先级（是前台交互还是后台处理）
- 已占用资源，还需要的资源
- 已经运行时间
- 资源抢占
- 把已经分发出去的锁收回来
- 让线程回退几步，这样就不用结束整个线程，成本比较低
##### 8.1.2.4 实际工程中如何避免死锁
tips1:设置超时时间
- Lock的tryLock(long timeout, TimeUnit unit)
- synchronized不具备尝试锁的能力
- 造成超时的可能性多：发生了死锁、线程陷入死循环、线程执行很慢
- 获取锁失败：打日志、发警报邮件、重启等
tips2:多使用并发类而不是自己设计锁
tips3:尽量降低锁的粒度，用不同的锁而不是一个锁
tips4:如果能使用同步代码块，就不使用同步方法，可以自己指定锁对象
tips5:给线程起有意义的名字，排查问题事半功倍
tips6:避免锁的嵌套
tips7:分配资源钱看能不能回收回来
Tips8:尽量不要几个功能使用同一把锁：专锁专用
#### 8.1.3 活锁
虽然线程并没有阻塞，也始终在运行，但是程序却得不到进展，因为线程始终重复做同样的事。
如果系统中有两个任务，它们总是因为对方的行为而改变自己的状态，那么就出现了活锁。最终结果是它们陷入了状态变更的循环而无法继续向下执行。
例如，有两个任务：任务1和任务2，它们都需要用到两个资源：资源1和资源2。假设任务1对资源1加了一个锁，而任务2对资源2加了一个锁。当它们无法访问所需的资源时，就会释放自己的资源并且重新开始循环。这种情况可以无限地持续下去，所以这两个任务都不会结束自己的执行过程。
```java
package deadlock;
import java.util.Random;
/**
* 演示活锁问题
*/
public class LiveLock {
static class Spoon {
private Diner owner;
public Spoon(Diner owner) {
this.owner = owner;
}
public Diner getOwner() {
return owner;
}
public void setOwner(Diner owner) {
this.owner = owner;
}
public synchronized void use() {
System.out.printf("%s吃完了!", owner.name);
}
}
static class Diner {
private String name;
private boolean isHungry;
public Diner(String name) {
this.name = name;
isHungry = true;
}
public void eatWith(Spoon spoon, Diner spouse) {
while (isHungry) {
if (spoon.owner != this) {
try {
Thread.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
continue;
}
Random random = new Random();
// if (spouse.isHungry && random.nextInt(10) < 9) { 添加随机因素避免活锁
if (spouse.isHungry) {
System.out.println(name + ": 亲爱的" + spouse.name + "你先吃吧");
spoon.setOwner(spouse);
continue;
}
spoon.use();
isHungry = false;
System.out.println(name + ": 我吃完了");
spoon.setOwner(spouse);
}
}
}
public static void main(String[] args) {
Diner husband = new Diner("牛郎");
Diner wife = new Diner("织女");
Spoon spoon = new Spoon(husband);
new Thread(() -> husband.eatWith(spoon, wife)).start();
new Thread(() -> wife.eatWith(spoon, husband)).start();
}
}
```
#### 8.1.4 资源不足(饥饿)
当某个任务在系统中无法获取维持其继续执行所需的资源时，就会出现资源不足。当有多个任务在等待某一资源且该资源被释放时，系统需要选择下一个可以使用该资源的任务。如果你的系统中没有设计良好的算法，那么系统中有些线程很可能要为获取该资源而等待很长时间。
要解决这一问题就要确保公平原则。所有等待某一资源的任务必须在某一给定时间之内占有该资源。可选方案之一就是实现一个算法，在选择下一个将占有某一资源的任务时，对任务已等待该资源的时间因素加以考虑。然而，实现锁的公平需要增加额外的开销，这可能会降低程序的吞吐量。
#### 8.1.5 优先权反转
当一个低优先权的任务持有了一个高优先级任务所需的资源时，就会发生优先权反转。这样的话，低优先权的任务就会在高优先权的任务之前执行。
### 8.2 性能问题
上下问切换问题
# JUC部分
### 1 线程池
#### 1.1 什么是线程池
线程池就是创建一个缓冲池存放线程，执行结束以后，该线程并不会死亡，而是再次返回线程池中成为空闲状态，等候下次任务来临，这使得线程池比手动创建线程有着更多的优势：
- 降低系统资源消耗：通过重用已存在的线程，降低线程创建和销毁造成的消耗;
- 提高系统响应速度：当有任务到达时，通过复用已存在的线程，无需等待新线程的创建便能立即执行;
- 方便线程并发数的管控：因为线程若是无限制的创建，可能会导致内存占用过多而产生OOM；
- 节省cpu切换线程的时间成本(需要保持当前执行线程的现场，并恢复要执行线程的现场)；
- 提供更强大的功能：比如延时定时线程池。 Timer vs ScheduledThreadPoolExecutor
#### 1.2 线程池构造函数分析
线程池构造函数的参数
| 参数名 | 类型 | 含义 |
| ------------- | ------------------------ | ------------------------------------------------------- |
| corePoolSize | int | 核心线程数 |
| maxPoolSize | int | 最大线程数 |
| keepAliveTime | long | 存活时间 |
| workQueue | BlockingQueue | 任务存储队列 |
| threadFactory | ThreadFactory | 当线程需要新的线程时，会使用threadFactory来生成新的线程 |
| Handler | RejectedExecutionHandler | 由于线程池无法接受你所提交的任务的拒绝策略 |
- corePoolSize指的是核心线程数：线程池在完成初始化后，默认情况下，线程池中并没有任何线程，线程池会等待有任务到来时，再创建新的线程去执行任务；
- 线程池有可能会在核心线程数的基础上，额外增加一些线程，但是这些新的线程数有一个上限，这个上限就是maxPoolSize；
- 如果线程池当前的线程数多于corePoolSize，那么如果多余的线程空闲时间超过keepAliveTime，它们就会被销毁
- 新的线程时有ThreadFactory创建的，默认使用Executors.defaultThreadFactory()，创建出来的线程都在同一个线程组，拥有同样的NORM_PRIORITY优先级且都不是守护线程。如果自己指定ThreadFactory，那么就可以改变线程名、线程组、优先级、是否时守护线程等。
- 有三种常见类型的队列
- 直接交接：SynchronousQueue
- 无界队列：LinkedBlockingQueue
- 有界队列：ArrayBlockingQueue
#### 1.3 添加线程的规则
1. 如果线程数小于corePoolSize，即使其他工作线程处于空闲状态，也会创建一个新线程来运行新任务。
2. 如果线程数等于（或大于）corePoolSize但少于maxPoolSize，则将任务放入队列。
3. 如果队列满了，且线程数小于maxPoolSize，则创建一个新线程来运行任务。
4. 如果队列满了，且线程数大于或等于maxPoolSize则拒绝该任务。
#### 1.4 线程池应该手动创建还是自动创建
结论：手动创建更好，因为这样可以让我们更加明确线程池的运行规则，避免资源耗尽的风险。
阿里巴巴规范提示：
```xml
线程池不允许使用Executors去创建，而是通过ThreadPoolExecutor的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。 说明：Executors返回的线程池对象的弊端如下：
1）FixedThreadPool和SingleThreadPool:
允许的请求队列长度为Integer.MAX_VALUE，可能会堆积大量的请求，从而导致OOM。
2）CachedThreadPool:
允许的创建线程数量为Integer.MAX_VALUE，可能会创建大量的线程，从而导致OOM。
Positive example 1：
//org.apache.commons.lang3.concurrent.BasicThreadFactory
ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1,
new BasicThreadFactory.Builder().namingPattern("example-schedule-pool-%d").daemon(true).build());
Positive example 2：
ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
.setNameFormat("demo-pool-%d").build();
//Common Thread Pool
ExecutorService pool = new ThreadPoolExecutor(5, 200,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());
pool.execute(()-> System.out.println(Thread.currentThread().getName()));
pool.shutdown();//gracefully shutdown
Positive example 3：
This method is typically used for debugging and testing purposes.
*
* @return the current count
*/
public long getCount() {
return sync.getCount();
}
```
```java
/**
* Synchronization control For CountDownLatch.
* Uses AQS state to represent count.
*/
private static final class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 4982264981922014374L;
Sync(int count) {
setState(count);
}
int getCount() {
return getState();
}
// 当geState==0时返回正数，反之则返回负数
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
// state--操作
protected boolean tryReleaseShared(int releases) {
// Decrement count; signal when transition to zero
// 自旋+cas，state减至0返回true
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
}
```
await 方法解析
```java
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
// 注意：该方法在AQS中实现
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
// tryAcquireShared方法是CountDownLatch.Sync内部类重写AQS的方法
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
// 注意：该方法在AQS中实现
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
// 构建一个节点放入队列中
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
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
failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
```
countDown 方法解析
```java
public void countDown() {
sync.releaseShared(1);
}
// 注意：该方法在AQS中实现
public final boolean releaseShared(int arg) {
// state--操作，如果state为0，则执行doReleaseShared
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
// 注意：该方法在AQS中实现
private void doReleaseShared() {
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h);
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}
```
### 9 Future获取子线程执行结果
#### 9.1 Future和Callable
Runnable的缺陷
- 不能返回一个返回值
- 不能抛出checked Exception
Callable接口
类似于Runnable，被其它线程执行的任务。实现call方法。有返回值。
Future类的作用：Future相当于一个存储器，它存储了call()这个任务的结果而这个任务的执行时间是无法提前确定的，因为这完全取决于call()方法执行的情况。
Callable和Future的关系：我们可以通过Future.get来获取Callable接口返回的执行结果，还可以通过Future.isDone()来判断任务是否已经执行完了，以及取消这个任务，限时获取任务的结果等。在call()未执行完毕之前，调用get()的线程（假定此时是主线程）会被阻塞，直到call()方法返回了结果后，此时future.get()才会得到该结果，然后主线程才会切换到runnable状态。
Future类的重要方法：
- get()方法的行为取决于Callable任务的状态，只有以下5种情况：
1. 任务正常完成：get方法会立即返回结果
2. 任务尚未完成（任务还没开始货进行中）：get将阻塞并直到任务完成。
3. 任务执行过程中抛出Exception：get方法会抛出ExecutionException：这里的抛出异常，是call()执行时产生的那个异常，看到这个异常类型是`java.util.concurrent.ExecutionException`。不论call()执行时抛出的异常类型是什么，最后get方法抛出的异常类型都是ExecutionException。
4. 任务被取消：get方法会抛出CancellationException
5. 任务超时：get方法有个重载方法，是传入一个延迟时间的，如果时间到了还没有获取到结果，get(long timeout, TimeUnit unit)方法就会抛出TimeoutException。（但是任务其实还是在执行的，需要手动取消。cancel方法用于取消任务的执行）
- cancel()方法：用于取消任务的执行
- isDone()方法：判断线程是否执行完毕（不代表任务成功与否）
- isCancelled()方法：判断任务是否被取消
#### 9.2 线程池获取Future
```java
public class OneFuture {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(10);
Future future = executorService.submit(() -> {
TimeUnit.SECONDS.sleep(3);
return new Random().nextInt();
});
Integer integer = future.get();
System.out.println(integer);
executorService.shutdown();
}
}
```
```java
public class MultiFutures {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(2);
Callable callable = () -> {
TimeUnit.SECONDS.sleep(3);
return new Random().nextInt();
};
ArrayList futures = new ArrayList<>();
for (int i = 0; i < 20; i++) {
futures.add(executorService.submit(callable));
}
for (int i = 0; i < 20; i++) {
System.out.println(futures.get(i).get());
}
executorService.shutdown();
}
}
```
##### 9.2.1 Future.get方法超时处理
```java
package future;
import java.util.Random;
import java.util.concurrent.*;
/**
* get的超时方法
*/
public class Timeout {
public static void main(String[] args) {
ExecutorService executorService = Executors.newFixedThreadPool(2);
Future future = executorService.submit(() -> {
try {
TimeUnit.SECONDS.sleep(10);
} catch (InterruptedException e) {
// 执行cancel(true) 时，这里会接受到中断信号，下边的语句将被打印
System.out.println("执行被中断");
// 注意：被中断的值不会被用作最终的结果
return 1;
}
return new Random().nextInt();
});
Integer integer = null;
try {
integer = future.get(2, TimeUnit.SECONDS);
} catch (InterruptedException e) {
System.out.println("----------InterruptedException---------");
} catch (ExecutionException e) {
System.out.println("----------ExecutionException---------");
} catch (TimeoutException e) {
System.out.println("----------TimeoutException---------");
integer = 0;
System.out.println("init new integer for result");
// 取消任务的执行。避免不必要的等待时间
boolean cancel = future.cancel(true);
System.out.println("cancel结果：" + cancel);
}
executorService.shutdown();
System.out.println("result:" + integer);
}
}
```
##### 9.2.2 Future.cancel方法使用原则
cancel方法执行可能遇到的情况
1. 如果这个任务还没有开始执行，那么任务会被正常的取消，未来也不会被执行，方法返回true。
2. 如果任务已完成，或者已取消。那么cancel方法会执行失败，方法返回false。
3. 如果这个任务已经开始执行了，那么这个取消方法将不会直接取消该任务，而是会根据我们填的参数mayInterruptIfRunning做判断。
Future.cancel(true)适用于：
1. 任务能处理interrupt
Future.cancel(false)仅用于避免启动尚未启动的任务，适用于：
1. 未能处理interrupt的任务
2. 不清楚任务是否支持取消
3. 需要等待已经开始的任务执行完成
#### 9.3 FutureTask来创建Future
用FutureTask来获取Future和任务的结果。FutureTask是一种包装器，可以把Callable转化成Future和Runnable，它同时实现两者的接口。把Callable实例当作参数，生成FutureTask的对象，然后把这个对象当作一个Runnable对象，用线程池或另起线程去执行这个Runnable对象，最后通过FutureTask获取刚才执行的结果。
```java
package future;
import java.util.concurrent.*;
public class FutureTaskDemo {
public static void main(String[] args) {
Task task = new Task();
FutureTask futureTask = new FutureTask<>(task);
// new Thread(futureTask).start();
ExecutorService executorService = Executors.newFixedThreadPool(2);
executorService.submit(futureTask);
try {
System.out.println("futureTask:" + futureTask.get().intValue());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
}
class Task implements Callable {
@Override
public Integer call() throws Exception {
System.out.println("子线程正在计算");
Thread.sleep(3000);
int sum = 0;
for (int i = 0; i < 100; i++) {
sum += i;
}
return sum;
}
}
```