Java核心技术卷一 8. java并发

Posted 2020-12-03 lovezyu

什么是线程

每个进程拥有自己的一整套变量，而线程则共享数据。

没有使用多线程的程序，调用 Thread.sleep 不会创建一个新线程，用于暂停当前线程的活动。程序未结束前无法与程序进行交互。

使用线程给其他任务提供机会

将代码放置在一个独立的线程中，事件调度线程会关注事件，并处理用户的动作。

在一个单独的线程中执行一个任务的简单过程：

1. 将任务代码移到实现了 Runnable 接口的类的 run 方法中。

   public interface Runnable{
       void run();
   }
   
   Runnable r = () -> { task code; };

2. 由 Runnable 创建一个 Thread 对象：

   Thread t = new Thread(r);

3. 启动线程：

   t.start();//启动这个线程，将引发调用 run() 方法。新程序将并发运行。

中断线程

线程的 run 方法执行方法体重最后一条语句后，并由执行 return 语句返回时，或者在方法中没有捕获异常，程序终止。

java 早期有一个 stop 方法，可以终止线程，目前已被弃用。

没有可以强制线程终止的方法。然而，interrupt 方法可以用来请求终止线程。

线程调用 interrupt 方法时，线程的中断状态将被置位。这是每一个线程都具有 boolean 标志。每个线程不时的检查这个标志，判断线程是否被中断。

判断线程是否被置位，currentThread 方法获得当前线程，isInterrupted 方法判断此线程是否被置位：

Thread.currentThread().isInterrupted()

在一个被阻塞的线程（调用 sleep 或 wait ）上调用 interrupt 方法时，阻塞调用将会被 Interrupted Exception 异常中断。

中断一个线程不会让程序终止，不过是引起它的注意。被中断的线程可以决定如何响应中断。一般线程将简单地将中断作为一个终止的请求。

如果在中断状态被置位时调用 sleep 方法，它不会休眠。它会清楚状态抛出 InterruptedException。

异常有两种合理的选择：

• 在 catch 子句中调用 Thread.currentThread().interrupt()来设置中断状态。调用者可对其检测。
• 用 throws InterruptedException标记你的方法，调用者可以捕获这个异常。

//java.lang.Thread
void interrupt() //发送中断线程请求。中断状态为 true。如果线程 sleep 抛出异常。
static boolean interrupted() //测试线程是否中断。静态方法副作用，中断状态重置为 false
boolean isInterrupted() //测试线程是否被终止。
static Thread currentThread() //当前执行线程的 Thread 对象

线程状态

有 6 种状态：

• New（新创建）
• Runnable（可运行）
• Blocked（被阻塞）
• Waiting（等待）
• Timed waiting（计时等待）
• Terminated（被终止）

使用 getState 方法确定线程的状态。

新创建线程 New

当用 new 操作符创建一个新线程时，如 new Thread(r)，该线程还没有开始运行。

可运行线程 Runnable

一旦调用 start 方法，线程处于 runnable 状态。

一个可运行的选择可能正在运行也可能没有运行。

抢占式调度系统给每一个可运行线程一个时间片来执行任务。当时间片用完，操作系统剥夺该线程的运行权，并给另一个线程运行机会。

被阻塞线程和等待线程 Blocked Waiting

这个状态不允许任何代码且消耗最小的资源。知道线程调度器重新激活它。细节取决于它是怎样达到非活动状态的。

• 当一个线程视图获取一个内部的对象锁，而该锁被其他线程持有，进入阻塞状态。其他线程释放该锁，并且线程调度器运行本线程持有它的时候，该线程将变为非阻塞状态。
• 当线程等待另一个线程通知调度器一个条件时，他自己进入等待状态。在调用Object.wait或Thread.join方法，或者等待java.util.concurrent中的Lock或Condition时，就会出现这种情况。被阻塞状态与等待状态是有很大不同的。
• 调用一些方法将导致线程进入计时等待状态。这一状态一直保持到超时期满或者接受到适当的通知。带有超时参数的方法有Thread.sleep和Object.wait、Thread.join、Lock.tryLock以及Condition.await的计时版。

被终止的线程 Terminated

两种原因被终止：

• 因为 run 方法正常退出而自然死亡
• 没有捕获的异常终止了 run 方法意外死亡

线程属性

线程优先级

java 程序中，每个线程有一个优先级。默认，一个线程继承它的父线程的优先级。

可以用 setPriority 方法提高或降低任何一个线程的优先级。(1~10)

使用 static void yield 方法会导致当前执行线程处于让步状态，有其他的可运行线程具有至少与此线程同样高的优先级，那么这些线程接下来会被调度。

守护线程

通过调用 t.setDaemon(true);将线程装换为守护线程。

唯一用途为其他线程提供服务。只剩下守护线程时，虚拟机就退出了。如计时线程，定时给其他线程发送信号。

守护线程因该永远不要去访问固有资源，如文件、数据库。因为它会随时中断。

未捕获异常处理器

线程的 run 方法不能抛出任何受查异常，非受查异常会导致线程终止。此时线程会死亡，死亡之前异常被传递到一个用于未捕获异常的处理器。

处理器必须处于一个实现 Thread.UncaughtExceptionHandler接口的类。类有一个方法void uncaughtException(Thread t, Throwable e)

可以用setUncaughtExceptionHandler方法为任何线程安装一个处理器。

用静态方法 Thread.setDefaultUncaughtExceptionHandler为所有线程安装一个默认的处理器。

不为独立的线程安装处理器，此时的处理器就是该线程的 ThreadGroup 对象。

线程组

线程组是一个统一管理的线程集合。默认下，创建的所有线程属于相同的线程组。

ThreadGroup 类实现Thread.UncaughtExceptionHandler接口。它的方法uncaughtException操作如下：

1. 如果该线程有父线程组，那么父线程组的 uncaughtException 方法被调用。
2. 否则，如果Thread.getDefaultExceptionHandler方法返回一个非空的处理器，则调用该处理器。
3. 否则，如果Throwable是ThreadDeath的一个实例，什么都不做。
4. 否则，线程的名字以及 Throwable的栈轨迹被输出到System.err上。

同步

两个或两个以上的线程需要共享同一数据的存取。如果不使用同步，线程都调用了一个修改对象状态的方法，它们会产生讹误的对象。这一情况为竞争条件。

竞争条件详解

当两个线程视图同时更新同一个账户的时候，会出现一些问题，假设他们同时执行指令：

accounts[to] += amout;

处理过程：

1. 将 accouts[to] 加载到寄存器。
2. 增加 amount。
3. 将结果写回 accounts[to]。

如果第 1 个线程执行完步骤1和2后，被剥夺了运行权。第 2 个线程被唤醒并修改了 accounts 数组中的同一项。然后，第 1 个线程被唤醒并完成其第 3 步。

这一动作让线程 1 擦除了线程 2 所做的操作。总金额不在正确。

线程1	线程2	寄存器 accouts[to]	accouts[to]
加载 accouts[to]		线程1 10	10
增加 amout		线程1 12	10
	加载 accouts[to]	线程2 10	10
	增加 amout	线程2 12	10
	写入 accouts[to]	线程2写入 12	12
写入 accouts[to]		线程1写入 12	12

如果能够确保线程在失去控制之前方法运行完成，那么银行账户对象的状态永远不会出现讹误。

锁对象 ReentrantLock 重入锁

有两种机制防止代码块受并发访问的干扰：

• synchronized 关键字
• java SE 5.0 引入 ReentrantLock 类

synchronized 关键字自动提供一个锁以及相关的“条件”，对于大多数小显示锁的情况。

java.util.concurrent 框架为这些基础机制提供独立的类。

用 ReentrantLock 保护代码的基本结构如下：

private Lock bankLock = new ReentrantLock();

bankLock.lock();
try{
    critical section
} finally {
    bankLock.unlock();
}

这个结构能够确保任何时候只有一个线程进入临界区。一旦一个程序使用 lock 方法封锁了对象，其他任何程序都无法通过 lock 语句。当其他线程调用 lock 时，它们被阻塞，直到第一个线程释放锁对象。

锁是可重入的，线程可以重复地获取已经持有的锁。锁保持一个持有计数来跟踪对 lock 方法的嵌套调用，每进入一次计数器加1。线程在每一次调用 lock 都要调用 unlock 来释放锁。由于这一特性，被一个锁保护的代码可以调用另一个使用相同的锁的方法。

ReentrantLock 重入锁，还可以构建一个带有公平策略的锁。

条件对象 Condition

使用条件对象来管理那些已经获得了一个锁但是却不能做有用工作的线程。条件对象经常被称为条件变量。

一个锁对象可以有一个或多个相关的条件对象。用 newCondition 方法获得一个条件对象。

class Bank{
    private Condition sufficientFunds;
    public Bank(){
        ...
        sufficientFunds = bankLock.newCondition();
    }
}

//发现 transfer 方法余额不足，调用
sufficientFunds.await();

调用 await 方法的线程和等待获得锁的线程存在本质上的不同。

使用 await 方法的线程进入该条件的等待集，当锁可用时，不会马上解除阻塞，知道另一个线程调用同一条件上的 signalAll 方法时为止。

当另一个线程做完操作，应调用：

sufficientFunds.singnalAll();

此时重新激活因为这一条件而等待的所有线程。从阻塞的地方继续执行。

一个线程调用了 await 方法，而别的程序没有调用 signalAll 方法，那么就是死锁，永远阻塞。

signal 方法可以随机解除一个线程的阻塞状态。但有如果再次阻塞，并没有其他线程再次调用 singnal ，线程就会死锁。

synchronized 关键字

锁和条件的关键之处：

• 锁用来保护代码片段，任何时刻只能有一个线程执行被保护的代码。
• 锁可以管理视图进入被保护代码段的线程。
• 锁可以拥有一个或多个相关的条件对象。
• 每个条件对象管理那些已经进入保护的代码段但不能运行的程序。

java 每一个对象都有一个内部锁，如果一个方法用 synchronized 关键字声明，那么对象的锁将保护整个方法。当调用该方法时，线程必须获得内部的对象锁。

public synchronized void method(){
    method body
}
//等价
public void method(){
    this.intrinsicLock.lock();
    try {
        method body
    } finally {
        this.intrinsicLock.unlock();
    }
}

synchronized 内部对象锁只有一个相关条件。

• wait 方法添加一个线程到等待集中。
• notifyAll / notify 方法解除等待线程的阻塞状态。
• 他们是 Object 类的 final 方法。

与下面的方法等价：

intrinsicLock.await();
intrinsicLock.signallAll();
intrinsicLock.signall();

内部锁存在一些局限：

• 不能中断一个正在试图获得锁的线程。
• 试图获得锁时不能设定超时。
• 每个锁仅有单一的条件，可能是不够的。

使用哪种锁的建议：

• 最好既不是用 Lock / Condition 也不使用 synchronized 关键字。许多情况下你可以使用 java.util.concurrent 包中的一种机制，它会为处理所有的加锁。
• 如果 synchronized 适合程序，尽量使用它，它可以减少编写的代码数量。
• 特别需要 Lock / Condition 结构提供的独有特性时，才使用 Lock / Condition。

同步阻塞 synchronized(obj){...}

线程可以通过同步方法获得锁。通过进入一个同步阻塞也可以获得锁：

synchronized(obj){
    critical section
}//获得 obj 的锁

lock 对象被创建仅仅是用来使用每个 Java 对象持有的锁。

有时程序员使用一个对象的锁来实现额外的原子性，实际上称为客户端镇定。

监视器概念

锁和条件是线程同步的强大工具，但它们不是面向对象的。

监视器可以实现在不加锁的情况下，保证多线程的安全性。

监视器特性：

• 监视器只包含私有域的类。
• 每个监视器类的对象有一个相关的锁。
• 使用该锁对所有的方法进行加锁。
• 该锁可以有任意多个相关条件。

如果一个方法用 synchronized 关键字声明，那么，它表现的就像是一个监视器方法。

三个方法 java 对象不同于监视器，安全性下降：

• 域不要求必须是 private。
• 方法不要求必须是 synchronized。
• 内部锁对客户是可用的。

Volatile 域

如果向一个变量写入值，而这个变量接下来可能会被另一个线程读取，或者，从一个变量读取，而这个变量可能是之前被另一个线程写入的，此时必须使用同步。

volatile 关键字为实例域的同步访问提供了一种免锁机制。

如果一个域为 volatile ，那么编译器和虚拟机就知道该域是可能被另一个线程并发更新的。

private volatile boolean done;
public boolean isDone(){ return done; }
public void setDone(){ done = true; }

Volatile 不能提供原子性。不能确保读取、翻转和写入不被中断。

done = !done

final 变量

使用 final 的共享域可以安全地访问一个共享域。

final Map<String, Double> accounts = new HashMap<>();

线程会在构造函数完成构造只有才能看到这个 accounts 变量。

不使用 final ，不能保证其他线程看到更新后的值，可以只是看 null，而不是新构造的 HashMap。

原子性

java.util.concurrent.atomic包中有很多类使用了高效的机器级指令来保证其他操作的原子性。

死锁

有可能会因为每一个线程要等待更多的钱款存入而导致所有线程都被阻塞。这样的状态成为死锁。

线程局部变量

避免共享变量，使用 ThreadLocal 辅助类为各个线程提供各自的实例。

锁测试与超时

谨慎的申请锁：

if(myLock.tryLock()){
    try {
        ...
    } finally {
        myLock.unlock();
    }
} else {
    do something else
}

可以调用 tryLock 时，使用超市参数：

if (myLock.tryLock(100, TimeUnit.MILLISECONDS)) ...

TimeUnit 是一个枚举类型，超时参数可以让在等待期间被中断的线程抛出异常。允许程序打破死锁。

读/写锁

java.util.concurrent.locks 包定义了两个锁类。

读/写锁的必要步骤：

1. 构造一个 ReentrantReadWriteLock 对象

   private ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();

2. 抽取读锁和写锁

   private Lock readLock = rwl.readLock();

   private Lock readLock = rwl.writeLock();

3. 对所有的获取方法加读锁

   public double getTotalBalance(){
       readLock.lock();
       try {...}
       finally { readLock.unlock(); }
   }

4. 对所有的修改方法加写锁

   public void transfer(...){
       writeLock.lock();
       try {...}
       finally { writeLock.unlock();}
   }

为什么弃用 stop 和 suspend 方法

stop 方法天生就不安全。

suspend 方法会经常导致死锁。

阻塞队列

当视图向队列添加元素而队列已满，或是想从队列移出元素而队列为空的时候，阻塞队列导致线程阻塞。

线程安全的集合

高效的映射、集和队列

java.util.concurrent 包提供了映射、 有序集和队列的高效实现：

• ConcurrentHashMap
• ConcurrentSkipListMap
• ConcurrentSkipListSet
• ConcurrentLinkedQueue

这些集合使用复杂的算法，通过允许并发地访问数据结构的不同部分来使竞争极小化。

集合返回弱一致性的迭代器。迭代器不一定能反映出它们被构造之后的所有的修改，但是，它们不会将同一个值返回两次，也不会抛出 ConcurrentModificationException 异常。

ConcurrentHashMap 默认支持多达 16 个写着线程同时执行。如果同一时间写着超过 16 ，其他线程将会暂时阻塞。可以指定更大数目的构造器。

映射条目的原子更新

ConcurrentHashMap 只有为数不多的方法可以实现原子更新。

以下不是线程安全的：

Long oldValue = map.get(word);
Long newValue = oldValue == null ? 1 : oldValue + 1;
map.put(word, newValue);//Error 可能不会替换 oldValue

传统方法使用 replace 操作，它会以原子方式用一个新值替换原值，前提是之前没有其它线程把原值替换为其他值。

do {
    oldValue = map.get(word);
    newValue = oldValue == null ? 1 : oldValue + 1;
}while(!map.replace(word, oldValue, newValue));

或使用ConcurrentHashMap<String, AtomicLong>或在 java SE 8 中使用ConcurrentHashMap<String, LongAdder>：

map.putIfAbsent(word, new LongAdder());
map.get(word).increment();

map 函数接受键和相关联的值，它会计算新值。

map.compute(word, (k, v) -> v == null ? 1 : v + 1);

ConcurrentHashMap 中不允许有 null 值。

对并发散列映射的批操作 ConcurrentHashMap

• 搜索为每个键或值提供一个函数，直到函数生成一个非 null 的结果。然后搜索终止，返回这个函数的结果。
• 归约组合所有键或值，使用锁提供的一个累加函数。
• forEach为所有键或值提供一个函数。

每个操作都有 4 个版本：

• operationKeys：处理键
• operationValues：处理值
• operation：处理键或值
• operationEntries：处理 Map.Entry 对象

并发集视图

没有ConcurrentHashSet 类，但你可以使用 ConcurrentHashMap 创建视图。

Set<String> words = ConcurrentHashMap.<String>newKeySet();

//keySet 方法包含一个默认值，在为集增加元素时使用
Set<String> words = map.keySet(1L);
words.add("Java");

写数组的拷贝

CopyOnWriteArrayList 和 CopyOnWriteArraySet 是线程安全的集合，所有的修改线程对底层数组进行复制。

并行数组算法

静态 Arrays.parallelSort 方法可以对 一个基本类型值或对象的数组排序。

parallelPrefix 方法，它会用对应一个给定结合操作的前缀的累加结果替换 各个数组元素。

较早的线程安全集合

Vector 和 Hashtable 类就提供了线程安全的动态数组和散列表的 实现。

任何集合类都可以通过使用同步包装器（Collections .synchronized*）变成线程安全的

如果在另一个线程可能进行修改时要对集合进行迭代，仍然需要使用“ 客户端” 锁定：

synchronized (synchHashMap) {
    Iterator iter = synchHashMap.keySet().iterator();
    while (iter.hasNextO) . . 
} 

最好使用 java.util.conncurrent 包中定义的集合， 不使用同步包装器中的。特别是， 假如它 们访问的是不同的桶， 由于 ConcurrentHashMap 已经精心地实现了，多线程可以访问它而且 不会彼此阻塞。

有一个例外是经常被修改的数组列表。在那种情况下，同步的 ArrayList 可 以胜过 CopyOnWriteArrayList

Callable 与 Future

Runnable 封装一个异步运行的任务，可以把它想象成为一个没有参数和返回值的异步方法。

Callable 与 Runnable 类似，但是有返回值。Callable 接口是一个参数化的类型， 只有一个方法 call。

Future 保存异步计算的结果。可以启动一个计算，将 Future 对象交给某个线程，然后忘掉它。Future 对象的所有者在结果计算好之后就可以获得它。

get 方法的调用被阻塞， 直到计算完成。如果在计算完成之前， 第二个方法的调 用超时，拋出一个 TimeoutException 异常。如果运行该计算的线程被中断，两个方法都将拋 出 IntermptedException。如果计算已经完成， 那么 get 方法立即返回。

如果计算还在进行，isDone 方法返回 false 如果完成了， 则返回 true。

可以用 cancel 方法取消该计算。如果计算还没有开始，它被取消且不再开始。如果计算 处于运行之中，那么如果 maylnterrupt 参数为 true, 它就被中断。

执行器

如果程序中创建了大 量的生命期很短的线程，应该使用线程池 。

一个线程池中包含许多准备运行的 空闲线程。将 Runnable 对象交给线程池， 就会有一个线程调用 run 方法。 当 run 方法退出 时，线程不会死亡，而是在池中准备为下一个请求提供服务。

另一个使用线程池的理由是减少并发线程的数目。创建大量线程会大大降低性能甚至使 虚拟机崩溃。如果有一个会创建许多线程的算法， 应该使用一个线程数“ 固定的” 线程池以 限制并发线程的总数。

执行器（ Executor) 类有许多静态工厂方法用来构建线程池。

• newCachedThreadPool 必要时创建新线程；空闲线程会被保留 60 秒
• newFixedThreadPool 该池包含固定数量的线程；空闲线程会一直被保留
• newSingleThreadExecutor 只有一个线程的 “ 池”， 该线程顺序执行每一个提交的任务（类似于 Swing 事件分配线程）
• newScheduledThreadPool 用于预定执行而构建的固定线程池， 替代 java.util.Timer
• newSingleThreadScheduledExecutor 用于预定执行而构建的单线程 “ 池”

线程池

newCachedThreadPool

newFixedThreadPool

newSingleThreadExecutor

这 3 个方法返回实现了 ExecutorService 接口的 ThreadPoolExecutor 类的对象。

可用下面的方法之一将一个 Runnable 对象或 Callable 对象提交给 ExecutorService:

Future submit(Runnable task)
Future submit(Runnable task, T result)
Future submit(Callable task)

该池会在方便的时候尽早执行提交的任务。

调用 submit 时，会得到一个 Future 对象，可 用来查询该任务的状态。

• 第一个 submit 方法返回一个奇怪样子的 Future。可以使用这样一个对象来调用 isDone、 cancel 或 isCancelled。但是， get 方法在完成的时候只是简单地返回 null。
• 第二个版本的 Submit 也提交一个 Runnable， 并且 Future 的 get 方法在完成的时候返回指 定的 result 对象。
• 第三个版本的 Submit 提交一个 Callable, 并且返回的 Future 对象将在计算结果准备好的 时候得到它。

调用 shutdown 方法启动该池的关闭序列。被关闭的执 行器不再接受新的任务。当所有任务都完成以后，线程池中的线程死亡。

调用 shutdownNow 方法，该池取消尚未开始的所有任务并试图中断正在运行的线程。

下面总结了在使用连接池时应该做的事：

1. 调用 Executors 类中静态的方法 newCachedThreadPool 或 newFixedThreadPool。
2. 调用 submit 提交 Runnable 或 Callable 对象。
3. 如果想要取消一个任务， 或如果提交 Callable 对象， 那就要保存好返回的 Future 对象。
4. 当不再提交任何任务时，调用 shutdown。

预定执行

ScheduledExecutorService 接口具有为预定执行或重复执 行任务而设计的方法。 后两个方法实现了这个接口。

控制任务组

invokeAll 方法提交所有对象到一个 Callable 对象的集合中，并返回一个 Future 对象的列 表，代表所有任务的解决方案。

Fork-Join 框架

有些应用使用了大量线程， 但其中大多数都是空闲的。举例来说， 一个 Web 服务器可能会为每个连接分别使用一个线程。另外一些应用可能对每个处理器内核分别使用一个线程， 来完成计算密集型任务， 如图像或视频处理。Java SE 7中新引入了 fork-join 框架，专门用来 支持后一类应用。假设有一个处理任务， 它可以很自然地分解为子任务。

if (problemSize < threshold)
    solve problem directly
else {
    break problem into subproblems
    recursively solve each subproblem
    combine the results
}

可完成 Futrue

处理非阻塞调用的传统方法是使用事件处理器， 程序员为任务完成之后要出现的动作注册一个处理器。

同步器

java.util.concurrent 包包含了管理相互合作的线程集的类 。

信号量

一个信号量管理许多的许可证（permit)。

倒计时门栓

一个倒计时门栓（ CountDownLatch) 让一个线程集等待直到计数变为 0。倒计时门栓是 一次性的。一旦计数为 0, 就不能再重用了。

障栅

CyclicBarrier 类实现了一个集结点（rendezvous) 称为障栅（ barrier)。考虑大量线程运行 在一次计算的不同部分的情形。当所有部分都准备好时，需要把结果组合在一起。当一个线 程完成了它的那部分任务后， 我们让它运行到障栅处。一旦所有的线程都到达了这个障栅， 障栅就撤销， 线程就可以继续运行。

交换器

当两个线程在同一个数据缓冲区的两个实例上工作的时候， 就可以使用交换器 ( Exchanger) 典型的情况是， 一个线程向缓冲区填人数据， 另一个线程消耗这些数据。当它 们都完成以后，相互交换缓冲区。

同步队列

同步队列是一种将生产者与消费者线程配对的机制。当一个线程调用 SynchronousQueue 的 put 方法时，它会阻塞直到另一个线程调用 take 方法为止，反之亦然。与 Exchanger 的情 况不同， 数据仅仅沿一个方向传递，从生产者到消费者。 即使 SynchronousQueue 类实现了 BlockingQueue 接口， 概念上讲，它依然不是一个队 列。它没有包含任何元素，它的 size方法总是返回 0。