《实战java高并发程序设计》笔记(三)
写在前面
前1、2章中,我们对于 并发编程中一些基本的概念和原理有了一定的了解。
在第3章中,我们将学习一些 JDK 内部提供的使用的API 和框架。主要分为3个部分:同步控制的工具、线程池的支持、支持并发的容器
第三章 JDK并发包
第三章知识框架图3.1.1 重入锁 ReentrantLock
重入锁使用 java.util.concurrent.locks.ReentrantLock 类实现。之所以叫重入是因为其允许 锁反复进入,这里的反复进入限于一个线程。
// lock 是一个 ReentrantLock 对象
lock.lock();
lock.lock();
try{
i++;
}finally{
lock.unlock();
lock.unlock();
}
重入锁跟 synchronized 比较:
(1)原始构成
- Sync 是属于JVM层面的关键字。编译后会在同步块的前后分别形成 monitorenter & monitorexit 两个字节码指令。在执行 monitorenter 指令时,首先尝试获取对象锁。如果获取锁成功,则把 锁的计数器 加1;相应的在执行 monitorexit 指令时将锁的计数器减1,计数器为0,锁就被释放了。
- ReentrantLock 是API锁
(2)使用方法 - Sync 不需要手动释放锁
- ReentrantLock 需要手动释放,不然会死锁。一般写在 finally{ unlcok() }
(3)等待是否可中断
(4)公平锁
(5)绑定多个条件Condition
ReentrantLock 通过绑定多个条件可以实现精准的分组唤醒线程。而 Sync 只能随机唤醒一个或者全部唤醒。
下面看一下重入锁的重要方法:
- lock():获得锁,锁被占用则等待。
- lockInterruptibly():获得锁,但是优先响应中断
- tryLock():尝试获得锁,返回boolean,不等待
- tryLock(long time, TimeUnit unit):给定时间内获得锁
- unlock():释放锁
下面我们看一下 ReentrantLock 可以实现的几个特殊场景
1. 等待可中断
就是说,在等待锁的同时,可以响应中断。通常情况下,这种机制被用于处理死锁。
package ConcurrencyTest;
import java.util.concurrent.locks.ReentrantLock;
/**
* 测试重入锁中的响应中断功能
*/
public class LockInterruptilyTest {
private static ReentrantLock lock = new ReentrantLock();
private static class ReentrantLockThread implements Runnable{
@Override
public void run() {
try{
lock.lockInterruptibly();
for (int i=0;i<3;i++){
System.out.println(Thread.currentThread().getName()+"得到了锁 i="+i);
}
}catch (InterruptedException e){
System.out.println(Thread.currentThread().getName()+"被打断了");
}finally {
// 查询当前线程是否保持此锁
if(lock.isHeldByCurrentThread()){
lock.unlock();
}
}
}
}
public static void main(String args[]) {
ReentrantLockThread test = new ReentrantLockThread();
Thread t1 = new Thread(test,"t1");
Thread t2 = new Thread(test,"t2");
Thread t3 = new Thread(test,"t3");
t1.start();
t2.start();
t3.start();
t2.interrupt();
}
}
运行结果
t1得到了锁 i=0
t1得到了锁 i=1
t1得到了锁 i=2
t2被打断了
t3得到了锁 i=0
t3得到了锁 i=1
t3得到了锁 i=2
2. 公平锁
只需要设置 ReentrantLock(True)即可。
ReentrantLock fairLock = new ReentrantLock(True);
下面看一个公平锁的例子
package ConcurrencyTest;
import java.util.concurrent.locks.ReentrantLock;
/**
* 用 ReentrantLock 实现公平锁
*/
public class FairLockViaReentrantLock implements Runnable{
private static ReentrantLock fairLock = new ReentrantLock(true);
@Override
public void run() {
int i = 0;
while (i<10){
try{
fairLock.lock();
System.out.println(Thread.currentThread().getName()+"获得锁");
}finally {
fairLock.unlock();
}
i++;
}
}
public static void main(String args[]){
FairLockViaReentrantLock f1 = new FairLockViaReentrantLock();
Thread t1 = new Thread(f1,"Thread_t1");
Thread t2 = new Thread(f1,"Thread_t2");
t1.start();t2.start();
}
}
由于线程是由公平锁实现的,执行结果两个线程交替获得锁。
Thread_t1获得锁
Thread_t2获得锁
Thread_t1获得锁
Thread_t2获得锁
3. 绑定多个条件实现精确唤醒
在这里例子里面,就是通过3个Condition条件,来实现锁的精确唤醒顺序。比如这个 线程执行为 A--B--C 这样的 。
/**
* 锁绑定多个条件Condition
* 题目:多线程之间按顺序执行,实现A->B->C三个线程启动,要求如下: A打印5次,B打印10次,C打印15次,
* 紧接着 A打印5次,B打印10次,C打印15次, . . . 循环执行10轮
*/
public class LockConditionDemo {
public static void main(String[] args) {
ShareResource shareResource = new ShareResource();
new Thread(() -> {
for (int i = 0; i < 10; i++) {
shareResource.print5();
}
}, "A").start();
new Thread(() -> {
for (int i = 0; i < 10; i++) {
shareResource.print10();
}
}, "B").start();
new Thread(() -> {
for (int i = 0; i < 10; i++) {
shareResource.print15();
}
}, "C").start();
}
}
/**
* 共享资源类
*/
class ShareResource {
// A:1 B:2 C:3
private int num = 1;
private Lock lock = new ReentrantLock();
private Condition conditionA = lock.newCondition();
private Condition conditionB = lock.newCondition();
private Condition conditionC = lock.newCondition();
// 循环打印5次
public void print5() {
// 1、获取锁资源
lock.lock();
try {
// 2、判断是否可以执行业务
while (num != 1) {
// 阻塞等待
conditionA.await();
}
// 模拟业务执行
for (int i = 0; i < 5; i++) {
System.out.println(Thread.currentThread().getName() + "\t" + (i + 1));
}
// 3、通知其他线程,通过signal()方法唤醒线程
num = 2;
conditionB.signal();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
// 循环打印10次
public void print10() {
// 1、获取锁资源
lock.lock();
try {
// 2、判断是否可以执行业务
while (num != 2) {
conditionB.await();
}
// 模拟业务执行
for (int i = 0; i < 10; i++) {
System.out.println(Thread.currentThread().getName() + "\t" + (i + 1));
}
// 3、通知其他线程
num = 3;
conditionC.signal();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
// 循环打印15次
public void print15() {
// 1、获取锁资源
lock.lock();
try {
// 2、判断是否可以执行业务
while (num != 3) {
conditionC.await();
}
// 模拟业务执行
for (int i = 0; i < 15; i++) {
System.out.println(Thread.currentThread().getName() + "\t" + (i + 1));
}
// 3、通知其他线程
num = 1;
conditionA.signal();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
执行结果
A 1
A 2
A 3
A 4
A 5
B 1
B 2
B 3
B 4
B 5
B 6
B 7
B 8
B 9
B 10
C 1
C 2
C 3
C 4
C 5
C 6
C 7
C 8
C 9
C 10
C 11
C 12
C 13
C 14
C 15
......// 后面循环10次
3.1.2 Condition 条件
Condition对象和 wait() & notify() 方法作用基本是类似的。但是 wait() 和 notiry() 是和 sync 关键字结合使用的,而 Condition 是跟重入锁关联的。
通过 Lock 接口(重入锁就是实现了这个接口)的 Condition newCondition() 方法,生成一个与当前重入锁绑定的 Condition 实例。利用 Condition 独享,就可以让线程在合适的时间等待,或者通知线程继续执行。
Condition 接口提供的基本方法:
- await():让当前线程等待,同时释放锁,这个等待是响应中断的
- awaitUninterruptibly():
- awaitNanos()
- await(long time, TimiUnit unit):带时间的等待
- awaitUntil(Date deadline)
- void signal():唤醒
- void signalAll()
下面看一个例子:
package ConcurrencyTest;
import java.util.concurrent.locks.*;
/**
* 测试Condition条件相关的等待和唤醒功能
*/
public class ConditionTest {
private static ReentrantLock lock = new ReentrantLock();
// 通过一个lock对象,生成 与 对象绑定的 Condition 对象
private static Condition condition = lock.newCondition();
public static class ConditionThread implements Runnable{
@Override
public void run() {
try{
lock.lock();
condition.await();
System.out.println("等待完成,线程继续执行");
}catch (InterruptedException e){
e.printStackTrace();
}
finally {
lock.unlock();
}
}
}
public static void main(String args[]) throws InterruptedException{
ConditionThread conditionThread = new ConditionThread();
Thread thread = new Thread(conditionThread,"t1");
thread.start();
Thread.sleep(2000);
// 这里还是给 signal() 方法加锁了
lock.lock();
condition.signal();
lock.unlock();
}
}
3.1.3 信号量Semaphore:允许多个线程同时访问
- 信号量用于指定同时可以有多少个线程访问某一个资源。是对锁的一种扩展
- Semaphore 有两个构造参数,第一个传入一个整数,表示某段代码最多有 n 个线程可以访问;另外一个参数表示是否公平
- acquire() 用来获取一个许可,会等待直到获得许可
- acquireUninterruptily()
- boolean tryAcquire()
- boolean tryAcquire(long timeout, TimeUnit unit)
- release() 用来释放许可。释放前必须先获得
下面我们看一个例子
package ConcurrencyTest;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
/**
* 用来测试信号量
*/
public class SemaphoreTest {
// 声明了包含3个许可的信号量
private static Semaphore semaphore = new Semaphore(3);
private static class SemaphoreThread implements Runnable{
@Override
public void run() {
try {
semaphore.acquire();
System.out.println(Thread.currentThread().getName()+"获取一个许可");
Thread.sleep(1);
System.out.println(Thread.currentThread().getName()+"完成,开始释放许可");
semaphore.release();
}catch (InterruptedException e) {
e.printStackTrace();
}
}
public static void main(String args[]){
SemaphoreThread semaphoreThread = new SemaphoreThread();
ExecutorService executorService = Executors.newFixedThreadPool(10);
for(int i=0;i<12;i++){
executorService.submit(semaphoreThread);
}
}
}
}
运行结果会以3个线程为1组,来获取许可,最后释放许可。
3.1.4 ReadWriteLock 读写锁
- 读-写互斥:读阻塞写,写也阻塞读。写-写互斥。
- 在读次数远大于写次数时使用 读写分离锁,可以有效减少锁竞争。
- 获得读锁:reentrantReadWriteLock.readLock();获得写锁 reentrantReadWriteLock.writeLock()
实际上是在使用 ReentrantReadWriteLock 这个类。
书上给了一个例子,说明用读写分离锁的时候,读线程之间并行,写线程阻塞。而用重入锁,所有读写线程之间要相互等待,程序执行时间比较长。
3.1.5 倒计数器 CountDownLatch
- CountDownLatch 让某个线程一直 等待到 倒计时结束再开始执行。作用类似于 join,join 是让当前线程等待 join线程结束。join 的实现原理是,检查 join 线程是否存活,如果存活,则一直 wait
- CountDownLatch 构造方法接受一个整数,表示当前计数器的计数个数(线程数)
- CountDownLatch.countDown() 可以让计数器减一, countDownLatch.await() 让当前线程一直阻塞到计数器为0
下面看一个例子
package ConcurrencyTest;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* 测试倒计数器 CountDownLatch
*/
public class CountDonwLatchTest implements Runnable {
static final CountDownLatch end = new CountDownLatch(10);
static final CountDonwLatchTest test = new CountDonwLatchTest();
@Override
public void run() {
try {
// 模拟火箭发射
Thread.sleep(new Random().nextInt(10)*1000);
System.out.println("检查完成");
// 一个线程已经完成任务,计数器减一
end.countDown();
}catch (InterruptedException e){
e.printStackTrace();
}
}
public static void main(String args[]) throws InterruptedException{
Executor exec = Executors.newFixedThreadPool(10);
for(int i=0;i<10;i++){
((ExecutorService) exec).submit(test);
}
// 等待检查
end.await();
// 发射火箭
System.out.println("发射");
((ExecutorService) exec).shutdown();
}
}
这里注意,计数是针对线程数。每个线程执行完,运行 countDown() 方法,减一个计数。 await() 方法要等所有计数清零。
3.1.6 循环栅栏 CyclicBarrier
- CyclicBarrier 是可以反复使用的计数器,构造函数中第一个参数int parties 指定计数器总数,第二个参数Runnable barrierAction 指定每次计数结束后要执行的动作(这个动作一般也是一个 多线程类,看下面例子中)。
- CyclicBarrier 的 await() 在所有参与线程都在此 barrier 上调用 await() 方法之前,将一直等待。
可能抛出两个异常 Interruptedexception 和 BrokenBarrierException
下面看一个例子:
package ConcurrencyTest;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
/**
* 循环栅栏 CyclicBarrier 测试类
*/
public class CyclicBarrierTest {
public static class Soldier implements Runnable {
private String soldierName;
private final CyclicBarrier cyclic;
Soldier(CyclicBarrier cyclic, String soldierName) {
this.cyclic = cyclic;
this.soldierName = soldierName;
}
@Override
public void run() {
try {
//等待其他士兵到齐
cyclic.await();
doWork();
//等待所有士兵完成工作
cyclic.await();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}
void doWork() {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(soldierName + " :任务完成");
}
}
public static class BarrierRun implements Runnable {
boolean flag;
int N;
public BarrierRun(boolean flag, int N) {
this.flag = flag;
this.N = N;
}
@Override
public void run() {
if (flag) {
System.out.println("司令:[士兵" + N + "个,任务完成!]");
} else {
System.out.println("司令:[士兵" + N + "个,集合完毕!]");
flag = true;
}
}
}
public static void main(String[] args) {
final int N = 10;
Thread[] allSoldier = new Thread[N];
boolean flag = false;
CyclicBarrier cyclic = new CyclicBarrier(N, new BarrierRun(flag, N));
//设置屏障点,主要是为了执行这个方法
System.out.println("集合队伍!");
for (int i = 0; i < N; i++) {
System.out.println("士兵" + i + " 报道!");
allSoldier[i] = new Thread(new Soldier(cyclic, "士兵" + i));
allSoldier[i].start();
}
}
}
输出如下。程序两次到达屏障点,每次到达都会调用 BarrierRun 的 run() 方法,由于 flag 的设置,两次输出不同。关于倒计时器和循环栅栏的区别
集合队伍!
士兵0 报道!
士兵1 报道!
士兵2 报道!
士兵3 报道!
士兵4 报道!
士兵5 报道!
士兵6 报道!
士兵7 报道!
士兵8 报道!
士兵9 报道!
司令:[士兵10个,集合完毕!]
士兵7 :任务完成
士兵9 :任务完成
士兵5 :任务完成
士兵6 :任务完成
士兵0 :任务完成
士兵4 :任务完成
士兵3 :任务完成
士兵8 :任务完成
士兵1 :任务完成
士兵2 :任务完成
司令:[士兵10个,任务完成!]
3.1.7 线程阻塞工具类:LockSupport
- LockSupport 可以在任意位置让线程阻塞。其静态方法 park()。和 Object.wait() 相比,park 方法不需要获取对象的锁。
- 和 suspend 比,不会造成死锁,因为 LockSupport 内部是基与信号量 Semaphore实现的,为每一个线程都准备了一个许可,如果许可可用,park() 函数会立即返回。
- LockSupport 支持中断影响,但是不抛出中断异常,需要 Thread.interrupted() 进行判断。
下面看一个例子
/**
* LockSupport 线程阻塞工具类测试
*/
public class LockSupportTest {
private static class LockSupportThread implements Runnable{
@Override
public void run() {
System.out.println(Thread.currentThread().getName()+"线程阻塞");
LockSupport.park();
for(int i=0;i<5;i++){
System.out.println(i);
}
}
}
public static void main(String args[]) throws InterruptedException{
LockSupportThread lockSupportThread = new LockSupportThread();
Thread thread = new Thread(lockSupportThread,"t1");
thread.start();
thread.sleep(1000);
// 唤醒
System.out.println("main唤醒阻塞线程");
LockSupport.unpark(thread);
}
}
t1线程阻塞
main唤醒阻塞线程
0
1
2
3
4
3.2.1 线程池
线程池的概念:
- 线程池中有几个活跃的线程,需要使用线程的时候,从池子中随便拿一个空闲线程;完成工作后,将线程放回池子中。
- 创建线程的工作变成了从池子中获取活跃线程,销毁线程变成了向线程池归还线程。
线程池产生的原因:
- 大量的线程的创建和销毁会消耗很多时间
- 线程本身也要占据内存
- 线程回收给GC带来负担
3.2.2 JDK 对线程池的支持
Executor 框架结构图对 线程池 Executor 框架更详细的介绍
Executor 框架成员
- Executors 类是一个工厂类,用于配置线程池,主要有下面的工厂方法
- public static ExecutorService newFixedThreadPool(int nThreads)
- public static ExecutorService newSingleThreadExecutor()
- public static ExecutorService newCachedThreadPool()
- public static ScheduledExecutorService newSingleThreadScheduledExecutor()
- public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)
注意上面的方法的返回类型。
接下来对上面提到的方法逐个解析:
- newFixedThreadPool() 方法。返回一个固定线程数量的线程池。线程数量不变,没有空闲线程时候,任务被暂存到任务队列。
public static void main(String args[]){
TestThread testThread = new TestThread(); // testThread 是一个多线程类
ExecutorService executorService = Executors.newFixedThreadPool(5); // 配置线程池
for(int i=0;i<5;i++){
executorService.submit(testThread); // submit() 方法,用于向线程池提交任务。
}
}
- new SingleThreadExecutor():返回只有一个线程的线程池,多余任务会等待,FIFO执行等待任务。
- new CachedThreadPool():返回可调整线程数的线程池。多余的任务会创建新的线程处理任务。
public static void main(String[] args) {
CachedThreadPoolTask cachedThreadPoolTask = new CachedThreadPoolTask();
ExecutorService executorService = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
executorService.submit(cachedThreadPoolTask);
}
}
}
运行结果会创建10个线程。
- newScheduledThreadPool() 计划任务:返回 ScheduledExecutorService对象,可以根据时间对线程进行调度。ScheduledExecutorService 接口在 ExecutorService 接口上扩展了在给定时间执行某个任务的功能,比如在固定的延迟之后执行或者周期性执行某个任务。
这里给出了3个方法, schedule() 方法在给定的时间对任务进行一次调度。scheduleAtFixedRate() 以固定的频率调度,上一次任务开始执行之后 period 时间调度下一次任务,shceduleWithFixedDelay() 在上一次任务结束后,经过 delay 时间进行任务调度。
/**
* 用来测试 计划任务
* newScheduledThreadPool 线程池
*/
public class ScheduledExecutorServiceDemo {
public static void main(String args[]){
ScheduledExecutorService ses = Executors.newScheduledThreadPool(10);
ses.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(1000);
System.out.println(System.currentTimeMillis()/1000);
}catch (InterruptedException e){
e.printStackTrace();
}
}
},0,2,TimeUnit.SECONDS);
}
}
执行结果是以3s的间隔调度任务。如果是 scheduleAtFixedRate() 方法,任务时间设置为1s,执行间隔设置为2s,那么任务调度间隔最后就是2s;5s,3s,就是5s,因为 ScheduledExecutorService 不会出现任务堆叠,就是说任务执行周期太短,就会在任务结束后立即调用下一个任务。
- newSingleThreadScheduledExecutor():返回 ScheduledExecutorService对象,线程池大小为1。
跟上面唯一的区别就在于线程池size的区别,深入的还需要进一步学习。
3.2.3 核心线程池的内部实现
上一个小节中的各种创建线程池的工厂方法,实际上,其内部都使用了 ThreadPoolExecutor 类。因此,我们需要了解一下 ThreaPoolExecutor 类最重要的构造函数
public ThreadPoolExecutor(
int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler
)
参数含义如下:
- int corePoolSize 指定了线程池中的线程数量
- int maximumPoolSize 指定了线程池中最大线程数量
- long keepAliveTime 指定了线程池线程超过了 corePoolSize 时,多余空闲线程的存活时间(理解成多余线程的存活时间)
- BlockingQueue<Runnable> workQueue 指定了 任务队列,提交但还未完成的任务。
- ThreadFactory threadFactory 指定了线程工厂,用于创建线程,一般使用默认的即可
- RejectedExecutionHandler handler 指定了 拒绝策略,当任务很多来不及处理的时候,如何拒绝任务。
上面的参数都很直观,但是对于 workQueue 和 handler 我们需要进一步了解。
workQueue 是一个 BlockingQueue 对象,用于存放 Runnable 对象。在 ThreadPoolExecutor 的构造函数中,可以使用下面几种 BlockingQueue 接口:
- 直接提交的队列:直接提交队列功能由 SynchronousQueue 对象提供,该对象没有容量,插入和删除需要互相等待,提交的任务不会保存,没有空闲线程则创建线程来执行,线程数量超过最大线程数执行拒绝策略。
下面的 newCachedThreadPool() 方法就是用直接提交的队列实现的。
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
- 有界的任务队列:使用 ArrayBlockingQueue 实现,必须带容量,表示队列最大容量。
- 无界的任务队列:通过 LinkedBlockingQueue 实现。由于是无界的,所以除非系统资源耗尽,则不存在入队失败的情况
newFixedThreadPool() 产生固定大小线程池就是用 无界任务队列来实现的。
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
- 优先任务队列:通过 PriorityBlockingQueue 实现,是一个特殊的无界队列,可以根据任务自身的优先级执行任务。相比之下,无论是 有界队列ArrayBlockingQueue 类 还是 指定大小的 LinkedBlockingQueue 类,都只是按照FIFO规则来执行任务的。
-
线程池核心调度逻辑:任务提交数量 < 线程池线程数量corePoolSize ,直接分配线程执行任务; 大于,提交到等待队列,提交成功则等待执行;提交失败(比如说有界队列达到上限,或者说是用了 SynchronousQueue类),则要重新提交向线程池,判断是否 达到 最大线程数(maximumPoolSize),大于就执行拒绝策略。
ThreadPoolExecutor 类的任务调度逻辑
3.2.4 拒绝策略
刚刚我们提到过,当二次提交超过 maxPoolSize 时候,需要执行拒绝策略。这里的拒绝策略有4种:
- AbortPolicy(中止策略):直接拒绝,抛出异常,阻止系统工作
- CallerRunsPolicy(调用者运行策略):直接在调用者的县城中运行当前被丢弃的任务
- DiscardOldestPolicy(抛弃最老策略):尝试丢弃最老的一个请求也就是即将被执行的一个任务,然后尝试再次提交(妥协重试的思想)
- DiscardPolicy(抛弃策略):直接丢弃,不处理。(注意这个和 终止策略 的区别在于,抛弃策略不会停止程序运行。)
3.2.5 ThreadFactory 接口:自定义线程创建
线程池的产生时为了避免频繁的创建销毁线程,但是肯定还是会创建线程的,最开始的那些线程从哪里来呢,就是 ThreadFactory 接口,这个接口只有一个创建线程的方法
Thread newThread(Runnable r);
3.2.3 小节中提到过,ThreadPoolExecutor 的构造方法中,倒数第二个参数就是 ThreadFactory 对象,通常情况下,使用默认的,不用指定。
书上给出了一个重写 ThreadFactory 对象的例子,说明通过自定义 ThreadFactory,我们可以灵活地创建线程池中的线程。
public static void main(String args[]) throws InterruptedException{
ThreadFactoryTask threadFactoryTask = new ThreadFactoryTask();
ExecutorService es = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS, new SynchronousQueue<Runnable>(), new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r);
thread.setDaemon(true);
System.out.println("creat"+ thread);
return thread;
}
});
for(int i=0;i<5;i++){
es.submit(threadFactoryTask);
}
Thread.sleep(2000);
}
由于这里将线程池中的线程都设置为 后台线程,所以在 main 线程结束后,整个程序结束。
3.2.6 扩展线程池
ThreadPoolExecutor 其实也是一个可以扩展的线程池。其提供了 beforeExecute()、afterExecute() 、terminated() 三个方法来对线程池进行监控。
/**
* 测试 ThreadPoolExecutor 的可扩展性
*/
public class ThreadPoolExecutorTest {
public static class ThreadPoolExecutorTask implements Runnable{
@Override
public void run() {
System.out.println(Thread.currentThread().getName()+":"+System.currentTimeMillis());
}
}
public static void main(String args[]) throws InterruptedException{
ExecutorService es = new ThreadPoolExecutor(5,5,
0L,TimeUnit.MILLISECONDS,
new SynchronousQueue<Runnable>()){
@Override
protected void beforeExecute(Thread t, Runnable r) {
System.out.println("准备执行" + Thread.currentThread().getName());
}
@Override
protected void afterExecute(Runnable r, Throwable t) {
System.out.println("执行结束" + Thread.currentThread().getName());
}
@Override
protected void terminated() {
System.out.println("线程退出");
}
};
ThreadPoolExecutorTask threadPoolExecutorTask = new ThreadPoolExecutorTask();
for(int i=0;i<5;i++){
es.submit(threadPoolExecutorTask);
Thread.sleep(1000);
}
es.shutdown();
}
}
执行结果
准备执行pool-1-thread-1
pool-1-thread-1:1551530510144
执行结束pool-1-thread-1
准备执行pool-1-thread-2
pool-1-thread-2:1551530510160
执行结束pool-1-thread-2
准备执行pool-1-thread-4
pool-1-thread-4:1551530510160
执行结束pool-1-thread-4
准备执行pool-1-thread-5
准备执行pool-1-thread-3
pool-1-thread-3:1551530510160
执行结束pool-1-thread-3
pool-1-thread-5:1551530510160
执行结束pool-1-thread-5
线程退出
3.2.9 Fork/Join 框架
-
使用 fork() 方法后系统多一个执行分支,所以要 join 到这个分支执行完成之后,才能得到结果。如下图,将主任务分解成为若干个子任务,等待(join)所有子任务都有结果后,才能得到主任务的结果。
Fork&Join 框架 -
JDK 提供了 ForkJoinPool 线程池来节省资源,可以向线程池来提交任务 ForkJoinTask。而 ForkJoinTask 模板类有两个子类,有返回值的 RecursiveTask 和无返回值的 RecursiveAction
最后书上给出了一个求和的例子,通过 ForkJoinPool 和 RecuresiveTask,将很大数的求和转换成为 若干个小数求和的子任务,最终获得总共的结果。
3.3.1 并发集合简介
- ConcurrentHashMap :线程安全的 HashMap
- CopyOnWriteArrayList:在 读多写少的场合,性能非常好
- ConcurrentLinkedQueue:高效的并发队列,使用链表实现。看成是一个线程安全的 LinkedList
- BlockingQueue:阻塞队列。JDK 内部通过 链表、数组实现。适合作为数据共享的通道。
- ConcurrentSkipListMap:跳表的实现。是一个 Map,使用跳表的数据结构进行快速查找
除了上述之外, Vector 也是线程安全的。Collections 工具类也可以将集合包装成线程安全的。
3.3.2 ConcurrentHashMap
- Collections.synchronizedMap() 方法可以用来包装任意 Map,从而生成一个并发安全的 Map。但是这种在高并发情境下,等待问题较为严重。
- 实际上,更为专业的方法是使用 ConcurrentHashMap
- ConcurrentHashMap 在 jdk1.7中,通过减小锁粒度来提高性能。
对于 HashMap 和 ConcurrentHashMap ,这是一个较为重点的内容。
ConcurrentHashMap较为详细的解释
3.3.3 有关List的线程安全
跟上面提到过的一样,我们可以用 Collections.SynchronizedList() 方法来包装 list。获得一个线程安全的 List。
List<String> l = Collections.synchronizedList(new LinkedList<String>());
3.3.4 ConcurrentLinkedQueue
高效的并发队列(只需要记住是性能最好的并发环境下的队列即可),基于链表实现,可以看成是线程安全的 LinkedList。
3.3.5 CopyOnWriteArrayList
- 读多写少的场合,这个 List 性能非常好,因为其读取不加锁,写入不阻塞,只有 写入和写入之间同步.
- 写入操作使用锁来用于 写-写 的情况,然后通过内部复制生成一个新的数组,再用新数组替换老数组,修改完成后,读线程会察觉到 volatile 修饰的 array 被修改了,但是整个修改过程是不影响读的。所以从这个角度来说,是不用读取加锁的。
public boolean add(E e) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
Object[] elements = getArray();
int len = elements.length;
Object[] newElements = Arrays.copyOf(elements, len + 1);
newElements[len] = e;
setArray(newElements);
return true;
} finally {
lock.unlock();
}
}
3.3.6 BlockingQueue
- 是一个接口。适合用作数据共享的通道。
- 队列为空,take 操作时,当前线程会等待(在 notEmpty上),直到新的元素插入;队列满,put 操作会等待,直到有空位。
3.3.7 随机数据结构:跳表 SkipList
- 跳表可以进行快速查找。数据结构与平衡树类似,但是插入和删除操作无需和平衡树一样进行全局调整。因此在并发情况下并不需要对全局加锁,只需要部分加锁。
- 跳表维持了多个链表,最底层链表维持了所有元素,上一层是下一层的子集,采用空间换时间的算法来提高查找速度。
比如在上图这个结构中,查找元素 7。 顶层元素最少,在顶层找,没有;下一层;找到8,;下一层,确定6--8之间。即可以快速查找。