首先何为线程(进程/线程)?
进程操作系统动态执行基本单元,系统为每个进程分配内存,包括一般情况下,包括文本区域(text region/指令)、数据区域(data region/变量)和堆栈(stack region/对象)。
我们的程序虽然可以做到多进程,但是,多进程需要切换上下文,什么是上下文?
当程序执行到多进程的指令,那么会把当前的运行环境-堆栈等,复制一份,到另一块内存区域,而又因为cpu是轮寻机制,关于CPU运行原理,百度上有篇文章写的很好,我这边引用一下。
频繁切换上下文(大概一次20微秒),属于没必要的消耗。另外就是进程间通信需要通过管道机制,比较复杂。
所以通常需要性能的场景通常合理使用线程来提高效率,线程的定义是,一个线程有且至少有一个线程,线程利用进程的资源,且本身不拥有系统资源,所以对线程的调度的开销就会小很多(相对进程)。
因为这篇文章我定义到Java的分类下面,所以还是要通过Java来描述
其实我认为要真的好好深入学习线程进程,cpu调度这块,还是要通过C来学
日后有时间,我会用C语言来模拟实现一遍
既然了解了什么是线程,看下Java怎么实现多线程:
Thread
,Runnable
,Future
至于网上有些说4种的,其实就是用ExecutorService来管理了一下。
那么从头聊一聊。
Thread
其实是Runnable的实现类,类声明如下
public class Thread implements Runnable
看下最核心的一个方法
首先判定现成的状态,0状态表示该线程是新创建的,一切不是新建状态的线程,都视为非法
第二部添加到线程组,线程组默认初始长度为4,如果满了就阔为2倍。
之后可以看到,调用了一个本地方法start0
,如果成功,则更改started
标签量
最后一个判定,启动失败,从线程组中移除当前线程。
public synchronized void start() {
/**
* This method is not invoked for the main method thread or "system"
* group threads created/set up by the VM. Any new functionality added
* to this method in the future may have to also be added to the VM.
*
* A zero status value corresponds to state "NEW".
*/
if (threadStatus != 0)
throw new IllegalThreadStateException();
/* Notify the group that this thread is about to be started
* so that it can be added to the group's list of threads
* and the group's unstarted count can be decremented. */
group.add(this);
boolean started = false;
try {
start0();
started = true;
} finally {
try {
if (!started) {
group.threadStartFailed(this);
}
} catch (Throwable ignore) {
/* do nothing. If start0 threw a Throwable then
it will be passed up the call stack */
}
}
}
Thread
中出现的Runnable,作为一个接口,只有一个方法,就是run
。
@FunctionalInterface
public interface Runnable {
/**
* When an object implementing interface <code>Runnable</code> is used
* to create a thread, starting the thread causes the object's
* <code>run</code> method to be called in that separately executing
* thread.
* <p>
* The general contract of the method <code>run</code> is that it may
* take any action whatsoever.
*
* @see java.lang.Thread#run()
*/
public abstract void run();
}
之后来看Future,
Future
其实提供了和Runnable
能力并列的接口,简单解释一下为什么这么说,Runnable
接口提供了run
,也就是可以放在线程中执行的能力,Future
其实是赋予了线程执行后可以返回的能力,run
的声明是void
,所以没有返回值。
两者结合,简单易懂的一个类RunnableFuture
的接口就出来了。
那么相当于Thread
的实现类,FutureTask
就出现了,它就是集大成者。
这么说可能有点跳跃,先看下下面的实现,一看,诶,怎么没有Future?
Future本身是一个接口,跟Runnable是相同的级别,但区别通俗来讲在于他没有run的能力,这个能力来自于Runnable。
追溯一下FutureTask,发现它继承了RunnableFuture,诶,这个单词起的就有意思了,包含了Runnable,和Future。
点进去看下
public interface RunnableFuture<V> extends Runnable, Future<V> {
/**
* Sets this Future to the result of its computation
* unless it has been cancelled.
*/
void run();
}
Future就在这,关于Future里面有什么,大家可以点进去看看,里面最关键的就是get() throws InterruptedException, ExecutionException;
这个方法,就这这个方法,让我们通过调用api,拿到线程里面的值。
如果想使用这个东西,开启线程,这个时候不能用new Thread(future)
这种方式了,因为Thread没有这种能力,只实现了一个Runnable接口,
这个时候,一个新的类出现了,源码如下
@FunctionalInterface
public interface Callable<V> {
/**
* Computes a result, or throws an exception if unable to do so.
*
* @return computed result
* @throws Exception if unable to compute a result
*/
V call() throws Exception;
}
看下Callable
的声明,是有返回值的,并且可以抛出异常的。
这个返回值就很关键了,通过这个返回值,你可以把任何你想通过线程拿到的结果拿回来。
而拿结果的方法就是FutureTask的get()方法,之前我们看源码时又看到,这个get方法来自于Future接口的V get()方法
简单看下如何使用一个有返回值的多线程操作
public class CallableTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
Callable<Integer> callable = new Task();
FutureTask task = new FutureTask(callable);
Thread oneThread = new Thread(task);
oneThread.start();
System.out.println(">>> 工作结果 " + task.get().toString());
}
}
class Task implements Callable<Integer> {
@Override
public Integer call() throws Exception {
System.out.println(">>> 线程开始工作");
Thread.sleep(1000);
System.out.println(">>> 结束工作开始返回");
return 10;
}
}
可以看到FutureTask
依然调用的是Thread
,走的是本地方法start0
。
Runbale就没什么好说的了,实现一个接口,放到Thread里面去执行,基本没什么东西,能力与Thread差不多,区别是实现Runnable接口的类必须依托于Thread类才能启动,
//使用这个构造方法
public Thread(Runnable target) {
this(null, target, "Thread-" + nextThreadNum(), 0);
}
然后用Thread的start方法,需要注意的是,千万不要调run方法, 要用start。
最后看下ExecutorService
这个类,ExecutorService
级别很高,他的爸爸直接就是Executor。
他的儿子,是AbstractExecutorService
,这里实现了submit
,doInvokeAny
等方法。
而我们调用Executors.newFixedThreadPool(poolSize);
返回的是ThreadPoolExecutor
注:一般不建议使用Executors.newFixedThreadPool(poolSize);
,什么东西全是默认,建议如下方式:
ThreadFactory factory = new ThreadFactoryBuilder().setNameFormat("test-%d").build();
ExecutorService service = new ThreadPoolExecutor(5, 200, 0L, TimeUnit.MILLISECONDS,
new LinkedBlockingDeque<>(1024), factory, new ThreadPoolExecutor.AbortPolicy());
具体参数含义就自行百度吧,很多人讲的。
那ThreadPoolExecutor
这个类,又是AbstractExecutorService
的儿子,所以这个关系就很明显了
ThreadPoolExecutor
-> AbstractExecutorService
-> ExecutorService
-> Executor
看到一堆submit
方法,然而没什么用,真正关键的方法,是execute方法,在ThreadPoolExecutor
中实现。
这个类有点意思,一上来就给我一个下马威
private final AtomicInteger ctl = new AtlmicInteger(ctlOf(RUNNING, 0));
这个ctl到底是什么?
查了很久,在cnblogs里找到一个大神的描述 -- Okevin
这个变量使用来干嘛的呢?它的作用有点类似我们在《7.ReadWriteLock接口及其实现ReentrantReadWriteLock》中提到的读写锁有读、写两个同步状态,而AQS则只提供了state一个int型变量,此时将state高16位表示为读状态,低16位表示为写状态。这里的clt同样也是,它表示了两个概念:
workerCount:当前有效的线程数
runState:当前线程池的五种状态,Running、Shutdown、Stop、Tidying、Terminate。
int型变量一共有32位,线程池的五种状态runState至少需要3位来表示,故workCount只能有29位,所以代码中规定线程池的有效线程数最多为2^29-1。
看到这先来聊一下线程提交任务的规则,--《java并发编程艺术》
- 首先会判断核心线程池里是否有线程可执行,有空闲线程则创建一个线程来执行任务。
- 当核心线程池里已经没有线程可执行的时候,此时将任务丢到任务队列中去。
- 如果任务队列(有界)也已经满了的话,但运行的线程数小于最大线程池的数量的时候,此时将会新建一个线程用于执行任务,但如果运行的线程数已经达到最大线程池的数量的时候,此时将无法创建线程执行任务。
所以实际上对于线程池不仅是单纯地将任务丢到线程池,线程池中有线程就执行任务,没线程就等待。
最后附上大神对execute的注解
/**
* corePoolSize:核心线程池的线程数量
*
* maximumPoolSize:最大的线程池线程数量
*
* keepAliveTime:线程活动保持时间,线程池的工作线程空闲后,保持存活的时间。
*
* unit:线程活动保持时间的单位。
*
* workQueue:指定任务队列所使用的阻塞队列
*/
//ThreadPoolExecutor#execute
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
//由它可以获取到当前有效的线程数和线程池的状态
/*1.获取当前正在运行线程数是否小于核心线程池,是则新创建一个线程执行任务,否则将任务放到任务队列中*/
int c = ctl.get();
if (workerCountOf(c) < corePoolSize){
if (addWorker(command, tre)) //在addWorker中创建工作线程执行任务
return ;
c = ctl.get();
}
/*2.当前核心线程池中全部线程都在运行workerCountOf(c) >= corePoolSize,所以此时将线程放到任务队列中*/
//线程池是否处于运行状态,且是否任务插入任务队列成功
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (!isRunning(recheck) && remove(command))//线程池是否处于运行状态,如果不是则使刚刚的任务出队
reject(command);//抛出RejectedExceptionException异常
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
/*3.插入队列不成功,且当前线程数数量小于最大线程池数量,此时则创建新线程执行任务,创建失败抛出异常*/
else if (!addWorker(command, false)){
reject(command); //抛出RejectedExceptionException异常
}
}
//ThreadPoolExecutor#addWorker
private boolean addWorker(Runnable firstTask, boolean core) {
/*首先会再次检查线程池是否处于运行状态,核心线程池中是否还有空闲线程,都满足条件过后则会调用compareAndIncrementWorkerCount先将正在运行的线程数+1,数量自增成功则跳出循环,自增失败则继续从头继续循环*/
...
if (compareAndIncrementWorkerCount(c))
break retry;
...
/*正在运行的线程数自增成功后则将线程封装成工作线程Worker*/
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
final ReentrantLock mainLock = this.mainLock;//全局锁
w = new Woker(firstTask);//将线程封装为Worker工作线程
final Thread t = w.thread;
if (t != null) {
//获取全局锁
mainLock.lock();
/*当持有了全局锁的时候,还需要再次检查线程池的运行状态等*/
try {
int c = clt.get();
int rs = runStateOf(c); //线程池运行状态
if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)){ //线程池处于运行状态,或者线程池关闭且任务线程为空
if (t.isAlive()) //线程处于活跃状态,即线程已经开始执行或者还未死亡,正确的应线程在这里应该是还未开始执行的
throw new IllegalThreadStateException();
//private final HashSet<Worker> wokers = new HashSet<Worker>();
//包含线程池中所有的工作线程,只有在获取了全局的时候才能访问它。将新构造的工作线程加入到工作线程集合中
workers.add(w);
int s = worker.size(); //工作线程数量
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true; //新构造的工作线程加入成功
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
//在被构造为Worker工作线程,且被加入到工作线程集合中后,执行线程任务
//注意这里的start实际上执行Worker中run方法,所以接下来分析Worker的run方法
t.start();
workerStarted = true;
}
}
} finally {
if (!workerStarted) //未能成功创建执行工作线程
//在启动工作线程失败后,将工作线程从集合中移除
addWorkerFailed(w);
}
return workerStarted;
}
//ThreadPoolExecutor$Worker,它继承了AQS,同时实现了Runnable,所以它具备了这两者的所有特性
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
final Thread thread;
Runnable firstTask;
public Worker(Runnable firstTask) {
//设置AQS的同步状态为-1,禁止中断,直到调用runWorker
setState(-1);
this.firstTask = firstTask;
//通过线程工厂来创建一个线程,将自身作为Runnable传递传递
this.thread = getThreadFactory().newThread(this);
}
public void run() {
runWorker(this); //运行工作线程
}
}
Okevin博客