1. 线程简介
1.1 什么是线程
线程是现代操作系统能够进行调度和运算的基本单位
在一个进程中可以创建多个线程, 这些线程都有各自的计数器, 堆栈和局部变量等属性, 并且能够访问共享的内存变量, CPU在这些线程上高速切换, 让使用者感觉到这些线程在同时执行
1.2 .为什么使用多线程
- 处理器核心数量越来越多, 超线程技术的广泛应用
- 更快的影响时间, 多线程处理复杂业务
- 提高CPU利用率
1.3 线程优先级
现代操作系统基本采用时分的形式调度运行的线程, 操作系统会分出一个个时间片, 线程会分配到若干时间片, 当线程的时间片用完了就会发生线程调度, 并等待下次分配
线程分配到的时间片多少也就决定了线程使用处理器资源的多少, 线程优先级作用是控制分配给线程的时间片多一些或少一些
在java中, 通过priority属性来控制线程优先级, 优先级范围1~10, 默认优先级为5, 优先级高的线程分配时间片的数量要多余优先级低的线程
在不同的JVM以及操作系统上, 线程调度会存在差异, 有些操作系统甚至会忽略对线程优先级的设定, 所以优先级不能作为程序正确性的依赖
1.4 多线程缺点
- 线程的管理以及线程上下文切换占用额外资源
- 线程导致的死锁,共享资源的竞争等操作导致等待时间过长
- 相应的线程不安全的问题
1.5 线程的生命周期
thread_baisc_life.png
| 状态名称 | 说明 |
|---|---|
| 新建状态(new) | 线程被构建, 还没有调用start()方法 |
| 就绪状态(runnable) | 调用start()方法之后, 操作系统调度, 准备就绪 |
| 运行状态(running) | 操作系统执行线程 |
| 阻塞状态(blocked) | 线程阻塞于锁 |
| 等待状态(waiting) | 线程等待其他线程通知或者中断等操作 |
| 超时等待状态(time_waiting) | 等待其他线程通知或者中断等操作, 可以超时自动返回 |
| 终止状态(terminated) | 线程执行完毕 |
各种状态对应的java代码
- 阻塞状态
- 等待进入synchronized代码块和方法
- 等待状态和唤醒
- Object.wait()和Object.notify()
- Thread.join()
- LockSupport.park()和LockSupport.unpark(Thread)
- 等待超时
- Thread.sleep(long)
- Object.wait(long)
- Thread.join(long)
- LockSupport.parkNanos()
- LockSupport.parkUntil()
1.6 Daemon线程
Daemon线程是一种支持型线程, 主要被用作程序中后台调度以及支持型工作
当一个虚拟机只有Daemon线程的时候, java虚拟机将会被退出, 所以Daemon线程的finally语句不一定会执行
Thread.setDaemon(true)将线程设置为Daemon线程
public class DaemonExample {
public static void main(String[] args) {
Thread thread = new Thread(new Runnable() {
@Override
public void run() {
try {
//线程sleep 1秒
//这时候main函数已经执行完成, JVM发现没有非Daemon线程
//JVM直接退出, 而不会执行finally语句
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
System.out.println("DeamonThread finally finish.");
}
}
});
thread.setDaemon(true);
thread.start();
}
}
2. 实现多线程的三种方法
- 继承Thread类
- 实现Runnable接口
- 实现Callable接口,配合FutureTask类
2.1 继承Thread类
//继承Thread接口
public class ThreadTest extends Thread {
@Override
public void run() {
System.out.println("hello,thread!");
}
public static void main(String[] args) {
Thread t1 = new ThreadTest();
t1.start();
}
}
运行结果
hello,thread!
2.2 实现Runnable接口
//实现Runable接口
public class RunableTest implements Runnable {
@Override
public void run() {
System.out.println("hello,runnable!");
}
public static void main(String[] args) {
RunableTest r = new RunableTest();
Thread t2 = new Thread(r);
t2.start();
}
}
运行结果
hello,runnable!
2.3 实现Callable接口
//实现Callable接口
public class CallableTest implements Callable<String> {
@Override
public String call() throws Exception {
return "hello,callable!";
}
public static void main(String[] args) {
CallableTest callable = new CallableTest();
//使用FutureTask,可以获取多线程返回值
RunnableFuture<String> rf = new FutureTask<>(callable);
//线程start()
Thread t3 = new Thread(rf);
t3.start();
try {
System.out.println(rf.get());
} catch (Exception e) {
e.printStackTrace();
}
}
}
运行结果
hello,callable!
FutureTask的继承关系
jdk_callable_future.png
//FutureTask实现了Runnable接口
//所以使用Thread来处理
public class FutureTask<V> implements RunnableFuture<V> {
/**
* 构造一个FutureTask对象
* @param callable 传入的callable实例
*/
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW;
}
}
三种实现方式的联系与区别
- Thread类继承了Runnable接口
- Thread类使用了装饰者模式,内部的run方法实际是调用的是Runnable的run方法
- RunnableFuture继承了Runnable和Future接口
- 实例化一个Thread对象,然后调用其start方法,线程进入就绪状态
2.4 线程中断
中断可以理解为线程的一个标识位属性, 表示一个运行中的线程是否被其他线程进行了中断操作 可以使用那么interrupt()方法进行复位
interrupt()方法实际作用是, 给受阻塞的线程ThreadA设置一个中断信号, 在线程ThreadB中, 调用了ThreadA.interrupt()方法, 其执行顺序如下
- 如果线程ThreadA没有被阻塞, 那么interrupt()方法将不会起作用, 当线程ThreadA执行到wait(), sleep(), join()方法时, 才会抛出InterruptedException, 然后退出阻塞状态
- 如果线程ThreadA被阻塞 那么被执行interrupt()方法的线程会抛出InterruptedException, 然后退出阻塞状态
- 线程的sleep(), join()和Object#wait方法会不断检查线程中断状态的值
2.4.1 sleep和interrupt
public class InterruptDemo {
// 两个全局线程对象, threadB用来interrupt ThreadA
static Thread threadA = new Thread(new ThreadA());
static Thread threadB = new Thread(new ThreadB());
static Logger logger = Logger.getLogger("InterruptDemo");
public static void main(String[] args) {
threadA.start();
threadB.start();
}
// 线程A, 先for循环一秒钟, 在sleep一秒钟
// 观察interrupt()之后的执行顺序
static class ThreadA implements Runnable {
@Override
public void run() {
logger.info("threadA start, isInterrupted = " + threadA.isInterrupted());
// threadA for循环一秒钟
long startTime = System.currentTimeMillis();
// threadA for循环一秒钟
long startTime = System.currentTimeMillis();
for (;;)
if (System.currentTimeMillis() - startTime > 1000) break;
logger.info("after for loop, threadA isInterrupted = " + threadA.isInterrupted());
// threadA sleep 一秒钟
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
logger.severe("threadA InterruptedException " + e.getMessage() + ", isInterrupted = "
+ threadA.isInterrupted());
}
logger.info("after threadA sleep isInterrupted = " + threadA.isInterrupted());
}
}
// 线程B, 用来中断线程A
static class ThreadB implements Runnable {
@Override
public void run() {
logger.info("before threadB interrupt threadA");
// 中断threadA
threadA.interrupt();
logger.info("after threadB interrupt threadA, threadA isInterrupted = " + threadA.isInterrupted());
}
}
}
执行结果
10:57:07 thread.InterruptDemo$ThreadA run
信息: threadA start, isInterrupted = false
10:57:07 thread.InterruptDemo$ThreadB run
信息: before threadB interrupt threadA
10:57:07 thread.InterruptDemo$ThreadB run
信息: after threadB interrupt threadA, threadA isInterrupted = true
10:57:08 thread.InterruptDemo$ThreadA run
信息: after for loop, threadA isInterrupted = true
10:57:08 thread.InterruptDemo$ThreadA run
严重: threadA InterruptedException sleep interrupted, isInterrupted = false
10:57:08 thread.InterruptDemo$ThreadA run
信息: after threadA sleep isInterrupted = false
结果分析
- threadA开始执行的时候, isInterrupted默认为false
- threadB执行了threadA.interrupte()之后, threadA.isInterrupted()被设置为true, threadA处于中断状态
- 此时线程A继续执行for循环, 并没有抛出InterruptedException
- 然后在threadA.sleep()时候, 抛出InterruptedException, 并将isInterrupted复位为false
- threadA并没有进入阻塞状态, 而是继续执行后续代码
2.4.2 join和interrupt
public class InterruptDemo {
// 两个全局线程对象, threadB用来interrupt ThreadA
static Thread threadA = new Thread(new ThreadA());
static Thread threadB = new Thread(new ThreadB());
static Logger logger = Logger.getLogger("InterruptDemo");
public static void main(String[] args) {
threadA.start();
threadB.start();
}
// threadA启动后执行threadB.join()
// 等待threadB执行完成, threadA再继续执行
static class ThreadA implements Runnable {
@Override
public void run() {
logger.info("threadA start, isInterrupted = " + threadA.isInterrupted());
try {
//threadA会等待threadB执行完成, 再继续执行
threadB.join();
} catch (InterruptedException e) {
logger.severe("threadA InterruptedException " + e.getMessage() + ", isInterrupted = "
+ threadA.isInterrupted());
}
// threadA for循环一秒钟
long startTime = System.currentTimeMillis();
for (;;)
if (System.currentTimeMillis() - startTime > 1000) break;
logger.info("after for loop, threadA isInterrupted = " + threadA.isInterrupted());
}
}
// 线程B, 用来中断线程A
static class ThreadB implements Runnable {
@Override
public void run() {
logger.info("before threadB interrupt threadA");
// 中断threadA
threadA.interrupt();
logger.info("after threadB interrupt threadA, threadA isInterrupted = " + threadA.isInterrupted());
}
}
}
执行结果
10:57:07 thread.InterruptDemo$ThreadA run
信息: threadA start, isInterrupted = false
10:57:07 thread.InterruptDemo$ThreadB run
信息: before threadB interrupt threadA
10:57:07 thread.InterruptDemo$ThreadB run
信息: after threadB interrupt threadA, threadA isInterrupted = true
10:57:08 thread.InterruptDemo$ThreadA run
严重: threadA InterruptedException null, isInterrupted = false
10:57:08 thread.InterruptDemo$ThreadA run
信息: after for loop, threadA isInterrupted = false
结果分析
- threadA开始执行的时候, isInterrupted默认为false
- threadA执行threadB.join()之后, 会等待threadB执行结束,
- threadB执行了threadA.interrupte()之后, threadA.isInterrupted()被设置为true, threadA于中断状态
- 此时theadA处于等待状态, 抛出InterruptedException, 并将isInterrupted复位为false
- threadA没有等待threadB执行完成, 而是立即恢复, 继续执行后续代码
2.4.3 wait和interrupt
public class InterruptDemo {
// 两个全局线程对象, threadB用来interrupt ThreadA
static Thread threadA = new Thread(new ThreadA());
static Thread threadB = new Thread(new ThreadB());
static Logger logger = Logger.getLogger("InterruptDemo");
//lock锁
static Object lock = new Object();
public static void main(String[] args) {
threadA.start();
threadB.start();
}
// 线程A,
static class ThreadA implements Runnable {
@Override
public void run() {
logger.info("threadA start, isInterrupted = " + threadA.isInterrupted());
synchronized (lock) {
try {
//等待其他线程唤醒
lock.wait();
} catch (InterruptedException e) {
logger.severe("threadA InterruptedException " + e.getMessage() + ", isInterrupted = "
+ threadA.isInterrupted());
}
logger.info("after threadA wait, isInterrupted = " + threadA.isInterrupted());
}
}
}
// 线程B, 用来中断线程A
static class ThreadB implements Runnable {
@Override
public void run() {
logger.info("before threadB interrupt threadA");
// 中断threadA
threadA.interrupt();
logger.info("after threadB interrupt threadA, threadA isInterrupted = " + threadA.isInterrupted());
}
}
}
执行结果
10:57:07 thread.InterruptDemo$ThreadA run
信息: threadA start, isInterrupted = false
10:57:07 thread.InterruptDemo$ThreadB run
信息: before threadB interrupt threadA
10:57:07 thread.InterruptDemo$ThreadB run
信息: after threadB interrupt threadA
10:57:08 thread.InterruptDemo$ThreadA run
严重: after threadB interrupt threadA, threadA isInterrupted = true
10:57:08 thread.InterruptDemo$ThreadA run
信息: after threadA wait, isInterrupted = false
结果分析
- threadA开始执行的时候, isInterrupted默认为false
- threadA锁住lock对象, 然后执行lock.wait(), threadA进入等待状态, 等待其他线程唤醒
- threadB执行了threadA.interrupte()之后, threadA.isInterrupted()被设置为true, threadA处于中断状态
- 此时theadA处于等待状态, 抛出InterruptedException, 并将isInterrupted复位为false
- threadA恢复执行
2.4.4 线程终结和interrupt
如果一个线程threadA已经处于终结状态, 即使该线程正处于中断状态, 在调用threadA.isInterrupted()会返回false
public class InterruptDemo {
public static void main(String[] args) throws InterruptedException {
Thread threadA = new Thread(() -> {
System.out.println("threadA run");
});
threadA.start();
//threadA中断
threadA.interrupt();
//主线程等待threadA执行结束
threadA.join();
System.out.println("after threadB interrupt threadA, threadA isInterrupted = " + threadA.isInterrupted());
}
}
运行结果
threadA run
after threadB interrupt threadA, threadA isInterrupted = false
2.4.5 Thread#interrupt源码简介
public class Thread implements Runnable {
/**
* 判断线程是否被中断过
* 不复位线程中断标记位
*/
public boolean isInterrupted() {
return isInterrupted(false);
}
/**
* 中断线程, 并复位线程中断标记位
* @return 线程是否被打断
*/
public static boolean interrupted() {
return currentThread().isInterrupted(true);
}
/**
* 中断this线程
*/
public void interrupt() {
//判断this线程是否有权限访问Thread.currentThread
if (this != Thread.currentThread())
checkAccess();
synchronized (blockerLock) {
Interruptible b = blocker;
if (b != null) {
interrupt0();
b.interrupt(this);
return;
}
}
interrupt0();
}
private native boolean isInterrupted(boolean ClearInterrupted);
public static native void sleep(long millis) throws InterruptedException;
}
2.4.5 Thread#interrupt使用场景
安全地终止线程, interrupt操作对进行中的线程没有影响, 但是会打断线程的等待, 最合适用来取消或者停止任务, 例如ThreadPoolExecutor#shutdown()就通过调用interrupt方法, 让线程快速执行完成, 最后再终止线程池中的任务
3. 线程间通信
每个线程拥有自己的栈空间, 相互配合的话就需要实现线程间通信
3.1 等待/通知机制
wait, notify, notifyAll都是Object类的final native方法
public class Object {
/**
* 在同步代码块中,持有当前对象锁的线程threadWait,
* 放弃对象锁,进入waiting状态, 线程被放到等待队列WaitQueue中
*
* 直到其他持有对象锁的线程threadNotify,执行notify或notifyAll方法, 唤醒等待队列中的一个或所有线程,
* 线程从等待队列WaitQueue中移到同步队列SynchronizedQueue,
* 线程从waiting状态进入blocked状态, 等到threadNotify执行结束, 同步队列中的线程开始争夺对象锁,
* 获取到对象锁之后, 获取到锁的线程会继续执行
*/
public final void wait() throws InterruptedException {
wait(0);
}
/**
* 在同步代码块中,持有当前对象锁的线程threadWait,
* 放弃对象锁,进入waiting状态, 线程被放到等待队列WaitQueue中
*
* 直到其他持有对象锁的线程threadNotify,执行notify或notifyAll方法, 唤醒等待队列中的一个或所有线程,
* 或者线程等待超时
* 线程从等待队列WaitQueue中移到同步队列SynchronizedQueue,
* 线程从waiting状态进入blocked状态, 等到threadNotify执行结束, 同步队列中的线程开始争夺对象锁,
* 获取到对象锁之后, 获取到锁的线程会继续执行
*/
public final native void wait(long timeout) throws InterruptedException;
/**
* 在同步代码块中,持有当前对象锁的线程threadNotify,
* 唤醒一个处于waiting状态的线程threadWait
*
* 线程threadWait从等待队列WaitQueue移到同步队列SynchronizedQueue,
* threadWait线程从waiting状态进入blocked状态
*
* 等到线程threadNotify同步代码块执行结束, 放弃对象锁
* threadWait线程获取到对象锁, 开始执行同步代码
*/
public final native void notify();
/**
* 在同步代码块中,持有当前对象锁的线程threadNotify,
* 唤醒所有处于waiting状态的线程
* 线程从等待队列WaitQueue移到同步队列SynchronizedQueue,
* 线程从waiting状态进入blocked状态
*
* 等到线程threadNotify同步代码块执行结束, 放弃对象锁
* 被唤醒的多个线程争夺对象锁, 获取到对象锁的线程, 开始执行同步代码
*/
public final native void notifyAll();
}
- 使用wait, notify和notifyAll方法时需要先对对象加锁
- 调用wait之后, 线程由runnabe状态转为waiting状态, 并将线程放入等待队列
- notify和notifyAll方法调用后, 等待线程依旧不会从wait()返回, 需要等待调用notify或notifyAll方法的线程释放锁之后, 等待线程才有机会从wait返回
- notify方法将等待队列中的一个等待线程从等待队列WaitQueue中移到同步队列SynchronizedQueue中, notifyAll方法则是将等待队列WaitQueue中的所有线程移到同步队列SynchronizedQueue中, 被移动的线程状态有waiting状态变为blocked状态
- 从wait方法返回的前期是获得了调用对象的锁
等待/通知经典范式
- 等待方:
- 获取对象锁
- 如果条件不满足, 那么调用对象的wait()方法, 被通知后仍要检查条件
- 条件满足则执行对应逻辑
synchronized(object){ while(条件不满足){ object.wait(); } doSomething(); } - 通知方
- 获取对象锁
- 改变条件
- 通知等待线程
synchronized(object){ doSomething2(); 改变条件 object.notifyAll(); }
3.2 管道输入流/输出流
管道输入/输出流主要用于线程之间的数据传输
PipedOutputStream/PipedInputStream管道字节流, PipedReader/PipedWriter管道字符流
如下: 我们创建了一个输入线程, 用来获取控制台的输入的字符串, 保存到内存中, 然后创建了一个打印线程, 用于获取内存中保存的控制台字符串, 然后打印字符串
public class PipedExample {
public static void main(String[] args) {
//实例化输入和打印线程
Inputer inputer = new Inputer();
Thread inputThread = new Thread(inputer);
Thread printThread = new Thread(new Printer(inputer));
//启动线程
inputThread.start();
printThread.start();
}
//打印线程
private static class Printer implements Runnable {
//PipedReader用于读取流, 并打印出来
private PipedReader in = new PipedReader();
public Printer(Inputer inputer) {
try {
//PipedReader和PipedWriter必须先connect
inputer.out.connect(in);
} catch (IOException e) {}
}
@Override
public void run() {
int receive = 0;
try {
while ((receive = in.read()) != -1) {
//读取内存中的数据, 并打印
System.out.print((char) receive);
}
} catch (IOException e) {}
}
}
private static class Inputer implements Runnable {
//输出流
private PipedWriter out = new PipedWriter();
@Override
public void run() {
int receive = 0;
try {
while ((receive = System.in.read()) != -1) {
//读取console的输入,输出到内存中
out.write(receive);
}
} catch (IOException e) {}
}
}
}
在控制台输入 hello, world! 然后回车, 运行结果如下:
hello, world!
hello, world!
3.3 Thread.join
如果一个线程threadA执行了threadB.join()语句, 那么threadA会等到线程threadB执行中止之后, 才从threadB.join()返回, 继续执行threadA的代码
下面的代码通过join()方法, 让10个线程按顺序执行
public class ThreadJoinExample {
public static void main(String[] args) {
Thread pre = new Thread(new OrderThread(null), "thread0");
pre.start();
for (int i = 1; i < 10; i++) {
Thread thread = new Thread(new OrderThread(pre), "thread" + i);
thread.start();
pre = thread;
}
}
// 按顺序执行的线程
static class OrderThread implements Runnable {
// 上一个线程
private Thread pre;
public OrderThread(Thread pre) {
this.pre = pre;
}
@Override
public void run() {
if (pre == null) return;
try {
pre.join();
} catch (InterruptedException e) {}
System.out.println("after " + pre.getName() + " join, "
+ Thread.currentThread().getName() + " terminate.");
}
}
}
运行结果:
after thread0 join, thread1 terminate.
after thread1 join, thread2 terminate.
after thread2 join, thread3 terminate.
after thread3 join, thread4 terminate.
after thread4 join, thread5 terminate.
after thread5 join, thread6 terminate.
after thread6 join, thread7 terminate.
after thread7 join, thread8 terminate.
after thread8 join, thread9 terminate.
结果分析:
每个线程终止的前提就是前驱线程的终止, 每个线程等待前驱线程终止后, 才从join()方法返回
实现原理:
Thread类的join方法就是通过wait方法实现的
public final synchronized void join(long millis){
if (millis == 0) {
while (isAlive()) {
wait(0);
}
} else {}
}
线程threadA调用join()方法, 相当于threadA调用了wait()方法, 那么主线程threadMain就会进入waiting状态, 等待threadA执行结束, 调用自身的notifyAll()方法, 通知所有等待在threadA的线程
3.4 ThreadLocal
线程本地变量, 是一个以ThreadLocal为键, 任意对象为值的存储结构, 线程可以通过ThreadLocal对象, 查询或者设置绑定在这个线程上的值
ThreadLocal有一个内部类ThreadLocalMap, ThreadLocalMap内部有一个内部类Entry, Entry以ThreadLocal为key, Object为value
ThreadLocal#get()方法执行步骤
- 先获取当前线程
- 获取当前线程绑定的值(ThreadLocalMap)
- 如果Map不为空, 那么获取当前ThreadLocal对应的entry
- 如果Map为空, 那么初始化值
- 设置线程绑定的值, 方法类似
//ThreadLocal源码
public class ThreadLocal<T> {
//获取ThreadLocal中保存的元素
public T get() {
//获取当前线程
Thread t = Thread.currentThread();
//获取线程对应的Map
ThreadLocalMap map = getMap(t);
if (map != null) {
//获取当前ThreadLocal对应的entry
ThreadLocalMap.Entry e = map.getEntry(this);
if (e != null) {
//获取并返回值
T result = (T)e.value;
return result;
}
}
//初始化并返回值
return setInitialValue();
}
//初始化并返回值
private T setInitialValue() {
//初始化值, 默认为null
T value = initialValue();
//获取当前线程
Thread t = Thread.currentThread();
//获取当前线程t的threadLocals
ThreadLocalMap map = getMap(t);
if (map != null)
map.set(this, value);
else
//如果Map为空,那么创建map
createMap(t, value);
return value;
}
void createMap(Thread t, T firstValue) {
//实例化当前线程的threadLocals
t.threadLocals = new ThreadLocalMap(this, firstValue);
}
//ThreadLocalMap静态内部类
static class ThreadLocalMap {
//Entry静态内部类
static class Entry extends WeakReference<ThreadLocal<?>> {
//和ThreadLocal绑定的元素
Object value;
//ThreadLocal为key
//任意Object为value
Entry(ThreadLocal<?> k, Object v) {
super(k);
value = v;
}
}
}
}
4. 线程池
服务端程序, 经常面对的是客户端传入执行时间短, 工作内容较为单一的任务, 需要服务端快速处理并返回结果
如果服务端每次接收到一个任务, 就创建一个线程, 那么可能会创建很多线程, 耗尽CPU资源, 这会使操作系统频繁的进行上下文切换, 增加系统负载, 而且线程的创建和消亡都要耗费系统资源
线程池很好的解决了这个问题, 预先创建若干数量的线程, 并重复使用这些固定数目的线程来完成任务的执行
