Java并发编程-多线程

Thread

简介

现代操作系统调度的最小单元是线程,也叫轻量级进程,在一个进程里可创建多个线程,这些线程都拥有各自的计数器、堆栈和局部变量等属性,并且能够访问共享的内存变量。处理器在这些线程上高速切换,让使用者感觉到这些线程在同时执行。

为什么要使用多线程

  • 更多的处理器核心
  • 更快的相应时间
  • 更好到编程模型

线程的状态

Java线程在运行的生命周期中可能处于6种不同的状态,在给定的一个时刻,线程只能处于其他的一个状态

状态名称 说明
NEW 初始状态,线程被构建,但是还没有调用start()方法
RUNNABLE 运行状态,Java线程将操作系统中的就绪和运行两种状态笼统地称为“运行中”
BLOCKED 阻塞状态,表示线程阻塞于锁
WAITING 等待状态,表示线程进入等待状态,进入该状态表示当前线程需要等待其他线程做出一些特定的动作(通知或者中断)
TIME_WAITING 超时等待状态,该状态不用与WAITING,它是可以在指定的时间自行返回的
TREMINATED 终止状态,表示当前线程已经执行完毕

常用方法介绍

方法 描述
start方法 start()用来启动一个线程,当调用start方法后,系统才会开启一个新的线程来执行用户定义的子任务,在这个过程中,会为相应的线程分配需要的资源。
run方法 run()方法是不需要用户来调用的,当通过start方法启动一个线程之后,当线程获得了CPU执行时间,便进入run方法体去执行具体的任务。注意,继承Thread类必须重写run方法,在run方法中定义具体要执行的任务
sleep方法 sleep相当于让线程睡眠,交出CPU,让CPU去执行其他的任务,sleep方法不会释放锁,也就是说如果当前线程持有对某个对象的锁,则即使调用sleep方法,其他线程也无法访问这个对象
yield方法 用yield方法会让当前线程交出CPU权限,让CPU去执行其他的线程。它跟sleep方法类似,同样不会释放锁。但是yield不能控制具体的交出CPU的时间,另外,yield方法只能让拥有相同优先级的线程有获取CPU执行时间的机会。调用yield方法并不会让线程进入阻塞状态,而是让线程重回就绪状态,它只需要等待重新获取CPU执行时间,这一点是和sleep方法不一样的
join方法 join方法顾名思义 就是往线程中添加东西的;join方法可以用于临时加入线程,一个线程在运算过程中,如果满足于条件,我们可以临时加入一个线程,让这个线程运算完,另外一个线程再继续运行
interrupt方法 nterrupt,顾名思义,即中断的意思。单独调用interrupt方法可以使得处于阻塞状态的线程抛出一个异常,也就说,它可以用来中断一个正处于阻塞状态的线程;另外,通过interrupt方法和isInterrupted()方法来停止正在运行的线程。注意通过interrupt方法可以中断处于阻塞状态的线程,直接调用interrupt方法不能中断正在运行中的线程。一般通过增加一个属性 isStop来标志是否结束while循环
过期的suspend()、resume()和stop() suspend()、resume()和stop()方法完成了线程的暂停、恢复和终止工作,以suspend()方法为例,在调用后,线程不会释放已经占有的资源(比如锁),而是占有着资源进入睡眠状态,这样容易引发死锁问题。同样,stop()方法在终结一个线程时不会保证线程的资源正常释放,通常是没有给予线程完成资源释放工作的机会,因此会导致程序可能工作在不确定状态下
getId 用来得到线程ID
getName和setName 用来得到或者设置线程名称
getPriority和setPriority 用来获取和设置线程优先级
setDaemon和isDaemon 用来设置线程是否成为守护线程和判断线程是否是守护线程,守护线程和用户线程的区别在于:守护线程依赖于创建它的线程,而用户线程则不依赖

等待与通知

一个线程修改了一个对象的值,而另一个线程感知到了变化,然后进行相应的操作,整个过程开始于一个线程,而最终执行又是另一个线程。前者是生产者,后者是消费者,这种模式隔离了”怎么做”,”做什么”,在功能层面上实现了解耦,体系结构上具备了良好的伸缩性。

等待方的原则:

  1. 获取对象的锁
  2. 如果条件不满足,那么调用锁的wait()方法,使该线程进入waiting,被通知后依然要检查条件
  3. 条件满足则执行对应的逻辑
    1
    2
    3
    4
    5
    6
    7
    8
    伪代码:
    synchronized(对象){
    while(条件不满足){
    对象.wait();
    }
    对应的逻辑处理
    }

通知方的原则:

  1. 获取对象的锁
  2. 改变条件
  3. 通知所有等待在该对象上的线程
1
2
3
4
5
6
伪代码:
synchronized(对象){
改变条件
对象.notifyAll();
}

实现代码:

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
public class WaitNotify{
static boolean flag = true;
static Object lock = new Object();
public static void main(String[] args) throws Exception{
Thread waitThread = new Thread(new Wait(),"WaitThread");
waitThread.start();
TimeUtil.SECONDS.sleep(1);
Thread notifyThread = new Thread(new Notify(),"NotifyThread");
notifyThread.start();
}
static class Wait implements Runnable{
public void run(){
//加锁
synchronized(lock){
//当条件不满足的时候,进入WAITTING状态,同时释放lock锁
while(flag){
System.out.println("flag is true ");
lock.wait();
}
//条件满足
System.out.println("doSomething");
}
}
}
static class Notify implements Runnable{
public void run(){
//加锁
synchronized(lock){
//获取lock的锁,然后进行通知,通知不会释放lock锁
//直到发出通知的线程执行完毕释放了lock锁,WaitThread线程才能从wait方法返回
lock.notifyAll();
System.out.println("flag is false now");
flag = false;
}
}
}

输出内容如下:

flag is true

flag is false now

doSomething

线程池

线程池的应用场景

Java中的线程池是运用场景最多的并发框架,几乎所有需要异步或并发执行任务的程序都可以使用线程池。在开发过程中,合理地使用线程池能够带来3个好处

  1. 降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。

  2. 提高响应速度。当任务到达时,任务可以不需要等到线程创建就能立即执行。

  3. 提高线程的可管理性。线程是稀缺资源,如果无限制地创建,不仅会消耗系统资源,还会降低系统的稳定性,使用线程池可以进行统一分配、调优和监控。但是,要做到合理利用线程池,必须对其实现原理了如指掌

线程池的实现原理-ThreadPoolExecutor

ThreadPoolExecutor执行execute方法分下面4种情况。

  1. 如果当前运行的线程少于corePoolSize,则创建新线程来执行任务(注意,执行这一步骤需要获取全局锁)。
  2. 如果运行的线程等于或多于corePoolSize,则将任务加入BlockingQueue。
  3. 如果无法将任务加入BlockingQueue(队列已满),则创建新的线程来处理任务(注意,执
    行这一步骤需要获取全局锁)。
  4. 如果创建新线程将使当前运行的线程超出maximumPoolSize,任务将被拒绝,并调用RejectedExecutionHandler.rejectedExecution()方法。

ThreadPoolExecutor采取上述步骤的总体设计思路,是为了在执行execute()方法时,尽可能地避免获取全局锁(那将会是一个严重的可伸缩瓶颈)。在ThreadPoolExecutor完成预热之后(当前运行的线程数大于等于corePoolSize),几乎所有的execute()方法调用都是执行步骤2,而步骤2不需要获取全局锁。

executor执行流程(基于JDK7.0)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();//AtomicInteger
//1.
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
//2.
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))//(21)
reject(command);
else if (workerCountOf(recheck) == 0)//(22)
addWorker(null, false);//为什么是false
}//3.
else if (!addWorker(command, false))
reject(command);
}
  1. 首先查看了当前线程池中的线程数量是否小于我们指定的核心线程池的数目,如果是就尝试新建一个线程,把command作为他的第一个任务,并把他们加入到线程池中。但是我们在判断了线程池的数量合法后,调用addWorker(command, true)把线程加入到线程池中时,是多线程并发的,可能会导致加入失败。如果加入成功,则直接返回,若假如失败,则重新获取clt,因为此时clt必发生了变化,否则不会失败,继续往下执行(2)。
  2. 通过isRunning(c) 判断如果线程池还在运行,那我们就尝试把当前的command加入到阻塞队列中。加入的过程也是并发的,也可能会出现失败。如果失败在继续执行(3)。加入阻塞队列成功后我们要重新在检查一遍,防止在加入的过程中线程时关闭了或者线程池中没有线程了,全部因为空闲时间超过了我们指定的alivetime被回收了。如果是线程池已经不再是RUNNING状态,则用我们的拒绝策略去丢弃它(21)。如果是线程池没有了线程,那我们新建一个空线程,让他去阻塞队列中去获取任务执行(22)。
  3. 如果上面的两步都没有执行成功,那我们此时就需要使用我们指定的最大线程池,来处理它,但是此时也是可能失败的,可能有多个线程执行么,如果失败,就用拒绝策略丢弃该线程。
    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
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    private boolean addWorker(Runnable firstTask, boolean core) {
    //(1)循环CAS操作,将线程池中的线程数+1.
    retry:
    for (;;) {
    int c = ctl.get();
    int rs = runStateOf(c);
    // Check if queue empty only if necessary.
    if (rs >= SHUTDOWN &&
    ! (rs == SHUTDOWN &&
    firstTask == null &&
    ! workQueue.isEmpty()))
    return false;
    for (;;) {
    int wc = workerCountOf(c);
    //core true代表是往核心线程池中增加线程 false代表往最大线程池中增加线程
    //线程数超标,不能再添加了,直接返回
    if (wc >= CAPACITY ||
    wc >= (core ? corePoolSize : maximumPoolSize))
    return false;
    //CAS修改clt的值+1,在线程池中为将要添加的线程流出空间,成功退出cas循环,失败继续
    if (compareAndIncrementWorkerCount(c))
    break retry;
    c = ctl.get(); // Re-read ctl
    //如果线程池的状态发生了变化回到retry外层循环
    if (runStateOf(c) != rs)
    continue retry;
    // else CAS failed due to workerCount change; retry inner loop
    }
    }
    //(2)新建线程,并加入到线程池workers中。
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
    //对workers操作要通过加锁来实现
    final ReentrantLock mainLock = this.mainLock;
    w = new Worker(firstTask);
    final Thread t = w.thread;
    if (t != null) {
    //细化锁的力度,防止临界区过大,浪费时间
    mainLock.lock();
    try {
    // Recheck while holding lock.
    // Back out on ThreadFactory failure or if
    // shut down before lock acquired.
    int c = ctl.get();
    int rs = runStateOf(c);
    //判断线程池的状态
    if (rs < SHUTDOWN ||
    (rs == SHUTDOWN && firstTask == null)) {
    //判断添加的任务状态,如果已经开始丢出异常
    if (t.isAlive()) // precheck that t is startable
    throw new IllegalThreadStateException();
    //将新建的线程加入到线程池中
    workers.add(w);
    int s = workers.size();
    //修正largestPoolSize的值
    if (s > largestPoolSize)
    largestPoolSize = s;
    workerAdded = true;
    }
    } finally {
    mainLock.unlock();
    }
    //线程添加线程池成功,则开启新创建的线程
    if (workerAdded) {
    t.start();//(3)
    workerStarted = true;
    }
    }
    } finally {
    //线程添加线程池失败或者线程start失败,则需要调用addWorkerFailed函数,如果添加成功则需要移除,并回复clt的值
    if (! workerStarted)
    addWorkerFailed(w);
    }
    return workerStarted;
    }

如何实现线程复用

线程池中,每一个线程都是一个Worker,Worker是一个内部类,继承了AbstractQueuedSynchronizer。

Worker的run方法会调用runWorker方法,runWorker方法会循环调用getTask获取阻塞队列的任务,达到线程复现的目的

Worker的主要字段就下面三个,代码也比较简单。

1
2
3
4
5
6
7
//线程池中正真运行的线程。通过我们指定的线程工厂创建而来
final Thread thread;
//线程包装的任务。thread 在run时主要调用了该任务的run方法
Runnable firstTask;
//记录当前线程完成的任务数
volatile long completedTasks;

Worker的构造函数

1
2
3
4
5
6
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker(写的很清楚)
this.firstTask = firstTask;
//利用我们指定的线程工厂创建一个线程,注意,参数是this,也就是在执行thread.run时,正真执行的是我们Woker类的run方法
this.thread = getThreadFactory().newThread(this);
}
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
45
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
while (task != null || (task = getTask()) != null) {
w.lock();
//线程池处于stop状态或者当前线程被中断时,线程池状态是stop状态。但是当前线程没有中断,则发出中断请求
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
//开始执行任务前的Hook,类似回调函数
beforeExecute(wt, task);
Throwable thrown = null;
try {
//执行任务
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
//任务执行后的Hook,类似回调函数
afterExecute(task, thrown);
}
} finally {
//执行完毕后task重置,completedTasks计数器++,解锁
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
//线程空闲达到我们设定的值时,Worker退出销毁。
processWorkerExit(w, completedAbruptly);
}
}
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
45
46
47
48
49
50
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
//如果线程池处于shutdown状态,并且队列为空,或者线程池处于stop或者terminate状态,在线程池数量-1,返回null,回收线程
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
//标识当前线程在空闲时,是否应该超时回收
boolean timed;
for (;;) {
int wc = workerCountOf(c);
//如果allowCoreThreadTimeOut 为ture或者当前线程数量大于核心线程池数目,则需要超时回收
timed = allowCoreThreadTimeOut || wc > corePoolSize;
//(1)
//如果线程数目小于最大线程数目,且不允许超时回收或者未超时,则跳出循环,继续去阻塞队列中取任务(2)
if (wc <= maximumPoolSize && ! (timedOut && timed))
break;
//如果上面if没有成立,则当前线程数-1,返回null,回收该线程
if (compareAndDecrementWorkerCount(c))
return null;
//如果上面if没有成立,则CAS修改ctl失败,重读,cas循环重新尝试修改
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
(2)
try {
//如果允许空闲回收,则调用阻塞队列的poll,否则take,一直等到队列中有可取任务
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
//取到任务,返回任务,否则超时timedOut = true;进入下一个循环,并且在(1)处会不成立,进而进入到cas修改ctl的程序中
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}

如何处理异常

当前运行的线程超出maximumPoolSize,任务将被拒绝,并调用RejectedExecutionHandler.rejectedExecution()方法,默认的异常处理为AbortPolicy

  • ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    /**
    * A handler for rejected tasks that throws a
    * {@code RejectedExecutionException}.
    */
    public static class AbortPolicy implements RejectedExecutionHandler {
    /**
    * Creates an {@code AbortPolicy}.
    */
    public AbortPolicy() { }
    /**
    * Always throws RejectedExecutionException.
    *
    * @param r the runnable task requested to be executed
    * @param e the executor attempting to execute this task
    * @throws RejectedExecutionException always.
    */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    throw new RejectedExecutionException("Task " + r.toString() +
    " rejected from " +
    e.toString());
    }
    }
  • ThreadPoolExecutor.DiscardPolicy:也是丢弃任务,但是不抛出异常。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
/**
* A handler for rejected tasks that silently discards the
* rejected task.
*/
public static class DiscardPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code DiscardPolicy}.
*/
public DiscardPolicy() { }
/**
* Does nothing, which has the effect of discarding task r.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}
  • ThreadPoolExecutor.DiscardOldestPolicy:丢弃队列最前面的任务,然后重新尝试执行任务(重复此过程)
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
/**
* A handler for rejected tasks that discards the oldest unhandled
* request and then retries {@code execute}, unless the executor
* is shut down, in which case the task is discarded.
*/
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code DiscardOldestPolicy} for the given executor.
*/
public DiscardOldestPolicy() { }
/**
* Obtains and ignores the next task that the executor
* would otherwise execute, if one is immediately available,
* and then retries execution of task r, unless the executor
* is shut down, in which case task r is instead discarded.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}
  • ThreadPoolExecutor.CallerRunsPolicy:由调用线程处理该任务
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    /**
    * A handler for rejected tasks that throws a
    * {@code RejectedExecutionException}.
    */
    public static class AbortPolicy implements RejectedExecutionHandler {
    /**
    * Creates an {@code AbortPolicy}.
    */
    public AbortPolicy() { }
    /**
    * Always throws RejectedExecutionException.
    *
    * @param r the runnable task requested to be executed
    * @param e the executor attempting to execute this task
    * @throws RejectedExecutionException always.
    */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    throw new RejectedExecutionException("Task " + r.toString() +
    " rejected from " +
    e.toString());
    }
    }

Executor框架

FixedThreadPool

FixedThreadPool被称为可重用固定线程数的线程池

1
2
3
4
5
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}

FixedThreadPool适用于为了满足资源管理的需求,而需要限制当前线程数量的应用场景,它适用于负载比较重的服务器

SingleThreadExecutor

SingleThreadExecutor。下面是Executors提供的,创建使用单个线程的SingleThread-
Executor的API

1
2
3
4
5
6
7
public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(),
threadFactory));
}

SingleThreadExecutor适用于需要保证顺序地执行各个任务;并且在任意时间点,不会有多
个线程是活动的应用场景

CachedThreadPool

CachedThreadPool。下面是Executors提供的,创建一个会根据需要创建新线程的CachedThreadPool的API

1
2
3
4
5
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}

CachedThreadPool是大小无界的线程池,适用于执行很多的短期异步任务的小程序,或者是负载较轻的服务器。

ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor通常使用工厂类Executors来创建。Executors可以创建2种类型的ScheduledThreadPoolExecutor,如下。

  • ScheduledThreadPoolExecutor。包含若干个线程的ScheduledThreadPoolExecutor。
  • SingleThreadScheduledExecutor。只包含一个线程的ScheduledThreadPoolExecutor。
1
2
3
4
public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
return new DelegatedScheduledExecutorService
(new ScheduledThreadPoolExecutor(1));
}
1
2
3
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}

ScheduledThreadPoolExecutor适用于需要多个后台线程执行周期任务,同时为了满足资源管理的需求而需要限制后台线程的数量的应用场景

参考资料

《Java并发编程的艺术》

JDK1.7中的ThreadPoolExecutor源码剖析

Java Thread 的使用