Java多线程编程

使用多线程的原因有: 提高资源利用率, 公平的共享计算机资源, 编写多任务时的便利性.

线程允许在同一个进程中同时存在多个程序控制流, 线程会共享进程范围内的资源, 每个线程都有各自的程序计数器, 栈以及局部变量等.

在大多数现代操作系统中, 都是以线程为基本的调度单位, 而不是进程. 如果没有明确的协同机制, 那么线程将彼此独立执行.由于同一个进程中的所有线程都将共享进程的内存地址空间, 因此这些线程都能访问相同的变量并在同一个堆上分配对象, 这就需要实现一种比在进程间共享数据粒度更细的数据共享机制. 如果没有明确的同步机制来协同对共享数据的访问, 那么会造成不可预测的结果.

线程的优势:

1. 发挥多处理器的强大能力: 如果设计正确, 多线程程序可以通过提高处理器资源的利用率来提升系统吞吐率.
2. 建模的简单性: 可以将复杂并且异步的工作流进一步分解为一组简单并且同步的工作流, 每个工作流在一个单独的线程中运行, 并在特定的同步位置进行交互.
3. 异步事件的简化处理: 为每个套接字连接分配各自的线程并且使用同步I/O, 那么会降低开发难度.
4. 响应更灵敏的用户界面: 现代的GUI框架中, 都采用一个事件分发线程来提高响应的灵敏度.

线程带来的风险:

1. 安全性问题: 由于多个线程要共享相同的内存地址空间, 并且是并发运行, 因此它们可能会访问或修改其他线程正在使用的变量, 线程会由于无法预料的数据变化而发生错误. 当多个线程同时访问和修改相同的变量时, 将会在串行编程模型中引入非串行因素, 而这种非串行性是很难分析的, 要使多线程程序的行为可以预测, 必须对共享变量的访问操作进行协同.

   如果没有同步, 无论是编译器, 硬件还是运行时, 都可以随意安排操作的执行时间和顺序, 例如对寄存器或者处理器中的变量进行缓存, 而这些被缓存的变量对于其他线程来说是暂时(设置永久)不可见的. 虽然这些有助于实现更优的性能, 但为开发人员带来了负担, 因为开发人员必须找出这些数据在哪些位置被多个线程共享, 只有这样才能使这些优化措施不破坏线程安全性.

2. 活跃性问题: 当某个操作无法继续执行下去时, 就会发生活跃性问题, 如死锁, 饥饿, 活锁等, 活跃性关注的目标是某件正确的事情最终会发生. 导致活跃性问题的错误同样难以分析, 因为它们依赖于不同 线程的事件发生时序, 因此在开发和测试中并不总是能够重现.

3. 性能问题: 活跃性意味着某件正确的事情最终会发生, 而性能意味着希望正确的事情尽快发生. 在设计良好的并发应用程序中, 线程能提升程序的性能. 但无论如何, 线程会带来某种程度的运行时开心. 在多线程程序中, 当线程调度器临时挂起活跃线程并转而运行另一个线程时, 就会频繁地出现上下文切换操作, 这种操作将带来极大的开销, 保存和恢复执行上下文, 丢失局部性, 并且CPU时间将更多地花在线程调度而不是线程运行上. 当线程共享数据时, 必须使用同步机制, 而这些机制往往会抑制某些编译器优化, 使内存缓存区中的数据无效, 以及增加共享内存总线的同步流量.

当框架在应用程序中引入并发性时, 通常不可能将并发性局限于框架代码, 因为框架本身会回调应用程序的代码, 而这些代码将访问应用程序的状态. 同样, 对线程安全性的需求也不能局限于被调用的代码, 而是要延伸到需要访问这些代码所访问的程序状态的所有代码路径, 因此对线程安全性的需求将在程序中蔓延开.

框架通过在框架线程中调用应用程序代码将并发性引入到程序中. 在代码中将不可避免地访问应用程序状态, 因此所有访问这些状态的代码都必须是线程安全的.

线程安全性

要编写线程安全的代码, 其核心在于要对状态访问操作进行管理, 特别是对共享的和可变的状态的访问.

对象的状态是指存储在状态变量中的数据, 在对象的状态中包含了可能任何影响其外部可见行为的数据.

共享意味着变量可以由多个线程同时访问, 而可变意味着变量的值在对象生命周期内可以发生变化.

如果当多个线程访问同一个可变的状态变量时没有使用合适的同步, 那么程序就会出现错误. 有三种方式可以修复这个问题:

1. 不在线程之间共享该状态变量
2. 将状态变量修改为不可变的变量
3. 在访问状态变量时使用同步.

当设计线程安全的类时, 良好的面向对象技术, 不可修改性以及明晰的不变性规范都能起到一定的帮助作用. 程序状态的封装性越好, 就越容易实现程序的线程安全性, 并且代码的维护人员也越容易保持这种方式.

在某些情况下, 良好的面向对象设计与实际情况的需求不一致, 在这些情况下, 可能需要牺牲一些良好的设计原则, 以换取性能或者对遗留代码的向后兼容. 编写并发应用程序的正确编程方法是: 首先使代码正确运行, 然后再提高代码的速度.

线程安全性的定义: 当多个线程访问某个类时, 不管运行时环境采用何种调度方式或者这些线程将如何交替执行, 并且在主调代码中不需要任何额外的同步或协同, 这个类都能表现出正确的行为, 那么就称这个类是线程安全的.

无状态对象: 不包含状态信息, 也不包含对其他类中域的引用, 而且计算过程中的临时状态仅存在于线程栈上的局部变量中且只能由正在执行的线程访问. 线程之间没有共享状态, 因此无状态对象一定是线程安全的.

竞态条件

在并发编程中, 由于不恰当的执行时序而出现不正确的的结果的情况, 叫做竞态条件. 当某个计算的正确性取决于多个线程的交替执行时序时, 就会发生竞态条件. 最常见的竞态条件是先检查后执行, 即通过一个可能失效的观测结果来决定下一步的动作.

先检查后执行的一种常见情况就是延迟初始化. 延迟初始化的目的是将对象的初始化操作推迟到实际被使用时才进行, 同时要确保只被初始化一次.

要避免竞态条件, 需要以原子方式(不可分割)执行, 就必须在某个线程修改该变量时, 通过某种方式防止其他线程使用这个变量, 从而确保其他线程只能在修改操作完成之前或之后读取和修改状态, 而不是在修改状态的过程中.

在实际情况中, 应尽可能使用现有的线程安全类来管理类的状态. 与非线程安全的对象相比, 判断线程安全对象的可能状态及其状态转换情况要更为容易, 从而更容易维护和验证线程安全性.

同时需要注意, 当在不变性条件中涉及多个变量时, 各个变量之间不是彼此独立的, 而是某个变量的值会对其他变量的值产生约束, 因此当更新某一个变量时, 需要在同一个原子操作中对其他变量同时进行更新, 即使这些变量都是线程安全对象.

要保持状态的一致性, 就需要在单个原子操作中更新所有相关的状态变量.

同步代码块

Java提供一种内置的锁机制来支持原子性: synchronized 同步代码块和同步方法. 同步代码块包含两部分: 一个作为锁的对象引用, 一个作为由这个锁保护的代码块. 每个Java对象都可以做一个实现同步的锁, 这些锁被称为内置锁或监视器锁. Java内置锁相当于一种互斥体, 是互斥锁, 这意味着最多只有一个线程能持有这种锁. 由于每次只能有一个线程执行内置锁保护的代码块, 因此这个锁保护的同步代码块会以原子方式执行.

Java内置锁是可重入的, 即线程试图获得一个已经由它自己持有的锁, 那么这个请求就会成功. 重入的一种实现方法是为每个锁关联一个计数值和一个所有者线程. 重入进一步提升了加锁行为的封装性, 简化了面向对象并发代码的开发, 也会一定程度上防止死锁情况的发生.

一种常见的错误是认为只有在写入共享变量是才需要使用同步. 因为可见性问题, 其他线程不一定能看到新写入的共享数据.

对象的内置锁与其状态之间没有内在的联系, 虽然大多数类都将内置锁用做一种有效的加锁机制, 但对象的域并不一定要通过内置锁来保护. 线程在获取对象的锁之后, 只能阻止其他线程获得同一个锁, 并不能阻止其他线程访问该对象及其对象的域.

一种常见的加锁约定是将所有的可变状态都封装在对象内部, 并通过对象的内置锁对所有访问可变状态的代码路径进行同步, 使得在该对象上不会发生并发访问. 但如果在添加新的方法或代码路径时忘记了使用同步, 那么这种加锁约定会很容易被破坏.

并非所有数据都需要锁的保护, 只有被多个线程同时访问的可变数据才需要通过锁来保护. 对于每个包含多个变量的不变性条件, 其中涉及的所有变量都需要有同一种锁保护.

如果不加区别地滥用synchronized, 可能导致程序中出现过多的同步, 还可能导致活跃性问题或性能问题.

使用synchronized对整个方法进行同步, 虽然能够确保线程安全性, 但有时付出的代价却很高, 如性能严重下降. 可以通过缩小同步代码块的作用范围, 不仅可以保证性能下降不那么严重, 还可以保证线程安全性. 应尽量将不影响共享状态且执行时间较长的操作从同步代码块中分离出去, 从而在这些操作的执行过程中, 其他线程可以访问共享状态,

要判断同步代码块的合理大小, 需要在各种设计之间进行权衡, 包括安全性(必须满足), 简单性和性能. 通常在简单性和性能之间存在着相互制约的因素, 当实现某个同步策略时, 一定不要盲目地为了性能而牺牲简单性, 这可能会破坏安全性.

当使用锁时, 应该清楚代码块中实现的功能, 以及在执行该代码块时是否需要很长的时间. 无论是执行计算密集的操作, 还是在执行某个可能阻塞的操作, 如果持有锁的时间过长, 那都会带来活跃性问题或性能问题. 当执行时间较长的计算或者可能无法快速完成的操作时(例如网络I/O), 一定不要持有锁.

对象的共享

可见性

关键字synchronized不仅能用于实现原子性或者确定临界区, 而且还能保证内存可见性. 有时不仅希望防止某个线程正在使用对象状态而另一个线程在同时修改该状态, 而且希望确保当一个线程修改了对象状态后, 其他线程能够看到发生的状态变化.

通常无法确保执行读操作的线程能适时地看到其他线程写入的值, 为了确保多个线程之间对内存写入操作的可见性, 必须使用同步机制.

还有种现象是重排序, 在没有同步的情况下, 编译器, 处理器以及运行时等都可能对操作的执行顺序进行一些意想不到的调整, 在缺乏足够同步的多线程程序中, 要想对内存操作的顺序进行判断, 几乎无法得出正确的结论.

缺乏同步的程序中可能产生错误结果的一种情况是失效数据. 当一个线程读取时可能会得到失效的值, 更糟糕的是失效值可能不会同时出现: 一个线程可能获得一个变量的最新值, 而获得另一个变量的失效值. 失效值可能会导致一些严重的安全性问题, 活跃性问题或令人困惑的故障.

Java内存模型要求, 变量的读取操作和写入操作必须是原子操作, 但对于非volatile类型的long和double变量, JVM允许将64位的读操作或写操作分解为两个32位的操作. 这样会造成对该变量的读操作和写操作在不同的线程中执行, 那么可能会读取到某个值的高32位和另一个值的低32位. 在多线程程序中使用共享且可变的long和double等类型的变量是不安全的, 除非用关键字volatile来声明它们或者用锁保护起来.

加锁的含义不仅仅局限于互斥行为, 还包括内存可见性. 为了确保所有线程都能看到共享变量的最新值, 所有执行读操作或者写操作的线程都必须在同一个锁上同步.

volatile变量

Java提供了一种较弱的同步机制即volatile变量, 用来确保将变量的更新操作通知其他线程. 当把变量声明为volatile类型后, 编译器与运行时都会注意到这个变量是共享的, 因此不会将该变量的操作与其他内存操作一起重排序. volatile变量不会被缓存在寄存器或者对其他处理器不可见的地方, 因此在读取volatile类型的变量时总会返回最新写入的值.

虽然volatile变量很方便, 但存在一些局限性, 如不足以确保递增(++)/递减(--)操作的原子性. 仅当volatile变量能简化代码的实现以及对同步策略的验证时, 才应该使用. 如果在验证正确性时需要对可见性进行复杂的判断, 那么就不要使用volatile变量. volatile变量的正确使用方式包括: 确保它们自身状态的可见性, 确保它们所引用对象的状态的可见性, 以及标识一些重要的程序生命周期事件的发生.

加锁机制既可以确保可见性又可以确保原子性, 而volatile变量只能确保可见性和防止重排序. 当且仅当满足以下所有条件时, 才应该使用volatile变量:

1. 对变量的写入操作不依赖变量的当前值, 或者能确保只有单个线程更新变量的值.
2. 该变量不会与其他状态变量一切纳入不变性条件中.
3. 在访问变量时不需要加锁.

发布与逸出

发布一个对象的意思是指使对象能够在当前作用域之外的代码中使用. 许多情况下, 要确保对象及其内部状态不被发布. 而在某些情况下, 又需要发布某个对象, 但如果在发布时需要确保线程安全性, 则可能需要同步. 发布内部状态会破坏封装性, 并使得程序难以维持不变性条件. 如果在对象构造完成之前就发布该对象, 就会破坏线程安全性.

当某个不应该发布的对象被发布时, 这种情况被成为逸出. 如下面的代码:

class UnsafeStates {
    private String[] states = new String[] {
        "AK", "AL", ...
    };
    
    public String[] getStates() {
        return states;
    }
}

按照上面代码的方式发布states会出现问题, 因为任何调用者都能修改这个数组的内容, 数组states已经逸出了它所在的作用域, 因为这个本应是私有的变量已经被发布了.

当发布一个对象时, 在该对象的非私有域中引用的所有对象同样会被发布. 一般来说, 如果一个已经发布的对象能够通过非私有的变量引用和方法调用到达其他的对象, 那么这些其他对象也都会被发布.

当把一个对象传递给某个外部方法时, 就相当于发布了这个对象. 无法知道哪些代码会执行, 也不知道外部方法中究竟会发布这个对象还是会保留对象的引用并在随后由另一个线程使用.

无论其他的线程会已发布的引用执行何种操作, 其实都不重要, 因为误用该引用的风险始终存在. 当某个对象逸出后, 你必须假设有某个类或线程可能会误用该对象. 这正是使用封装的最主要原因: 封装能够使得对程序的正确性进行分析变得可能, 并使得无意中破坏设计的约束条件变得更难.

最后一种发布对象或其内部状态的机制是发布一个内部的类实例. 如下面的代码:

public class ThisEscape {
    public THisEscape(EventSource source) {
        source.registerListener(
            new EventListener() {
                public void onEvent(Event e) {
                    doSomething(e);
                }
            }
        );
    }
}

当ThisEscape发布EventListener时, 也隐含地发布了ThisEscape实例本身, 因为在内部类的实例中包含了对ThisEscape实例的隐含引用, 即this引用在构造函数中逸出.

当且仅当对象的构造函数返回时, 对象才处于可预测的和一致的状态. 因此当从对象的构造函数中发布对象时, 只是发布了一个尚未构造完成的对象, 这种对象被认为是不正确的构造. 不要在构造过程中使this引用逸出.

在构造过程中this引用逸出的一个常见错误是在构造函数中启动一个线程. 当对象在其构造函数中创建一个线程时, 无论是显式创建还是隐式创建, this引用都会被新创建的线程共享. 在对象未完全构造之前, 新的线程就可以看见它. 在构造函数中创建线程并没有错误, 但最好不要立即启动它, 而是通过一个start或initialize方法来启动.

在构造函数中调用一个可改写的实例方法, 既不是私有方法也不是final方法时, 同样会导致this引用在构造过程中逸出.

如果想在构造函数中注册一个事件监听器或启动线程, 那么可以使用一个私有的构造函数和一个公共的工厂方法, 从而避免不正确的构造构成, 如下面的代码:

public class SafeListener {
    private final EventListener listener;
    
    private SafeListener() {
        listener = new EventListener() {
            public void onEvent(Event e) {
                doSomething(e);
            }
        };
    }
    
    public static SafeListener newInstance(EventSource source) {
        SafeListener safe = new SafeListener();
        source.registerListener(safe.listener);
        return safe;
    }
}

线程封闭

当访问共享的可变数据时, 通常需要同步. 一种避免使用同步的方式就是不共享. 如果仅在单线程内访问, 就不需要同步, 这种技术被称为线程封闭, 它是实现线程安全性的最简单方式之一. 如Netty框架和ThreadLocal变量. 但程序员仍然需要负责确保封闭在线程中的对象不会从线程中逸出.

栈封闭是线程封闭的一种特例, 在栈封闭中, 只能通过局部变量才能访问对象, 局部变量的固有属性之一就是封闭在执行线程中, 它们位于执行线程的栈中, 其他线程无法访问到. 如果在线程内部上下文中使用非线程安全的对象, 那么该对象仍然是线程安全的.

维持线程封闭性的一种更规范方法是使用ThreadLocal. 这个类能使线程中的某个值与保存值的对象关联起来. ThreadLocal提供了get和set等接口或方法, 这些方法为每个使用该变量的线程都存有一份独立的副本, 因此get总是返回由当前执行线程在调用set时设置的最新值.

ThreadLocal对象通常用于防止可变的单实例变量或全局变量进行共享, 每个线程都会拥有属于自己的变量.

当某个频繁执行的操作需要一个临时对象, 如一个缓冲区, 而同时又希望避免在每次执行时都重新分配该临时对象, 就可以使用这项技术.

ThreadLocal对象将自身作为key存放在Thread对象的ThreadLocalMap中, 而ThreadLocalMap并非Map接口的实现, 而是由弱引用的ThreadLocal类型的key和value组成的Entry数组构成.

在实现应用程序框架时大量使用了ThreadLocal. 比如通过将事务上下文保存在静态的ThreadLocal对象中, 当框架代码需要判断当前运行的是哪一个事务时, 只需从这个ThreadLocal对象中读取事务上下文. 这种机制很方便, 因为它避免了在调用每个方法时都传递执行上下文信息.

虽然ThreadLocal类似于全局变量, 但是能降低代码的可重用性, 并在类之间引入隐含的耦合性, 因此在使用时要格外小心. 且需要调用remove方法进行清理不用的内存占用.

不变性

满足同步的另一种方法是使用不可变对象, 如果某个对象在创建后其状态就不能被修改, 那么这个对象就称为 不可变对象. 线程安全性是不可变对象的固有属性之一, 其不变性条件是由构造函数创建的, 只要它们的状态不改变, 那么这些不变性条件就能得以维持.

不可变对象很简单, 只有一种状态, 并且该状态由构造函数来控制. 不可变对象更安全, 不可变对象不会被恶意代码或有问题的代码破坏, 可以安全地共享和发布这些对象.

当满足以下条件时, 对象才是不可变的:

1. 对象创建后其状态不能修改.
2. 对象的所有域都是final类型.
3. 对象是正确创建的, 在对象的创建期间, this引用没有逸出.

在不可变对象内部仍可以使用可变对象来管理它们的状态, 可变对象可封装在不可变对象内部使外部无法对其进行修改.

保存在不可变对象中的程序状态仍然可以更新, 即通过将一个保存新状态的实例来替换原有的不可变对象.

Java内存模型中, final域所引用的对象是可变的, 那么被引用的对象可以被修改的; final域能确保初始化过程的安全性, 从而可以不受限制地访问不可变对象, 并在共享这些对象时无需同步.

即使对象是可变的, 通过将对象的某些域声明为final类型, 仍然可以简化对状态的判断, 因此限制对象的可变性也就相当于限制了该对象可能的状态集合.

除非需要更高的可见性, 否则应将所有的域都声明为私有域, 是一个良好的编程习惯; 除非需要某个域是可变的, 否则应将其声明为final域, 也是一个良好的编程习惯.

对于在访问和更新多个相关变量时出现的竞争条件问题, 可以通过将这些变量全部保存在一个不可变对象中来消除. 如果是一个可变对象, 就必须使用锁来确保原子性. 如果是一个不可变对象, 那么当线程获得了对该对象的引用后, 就不必担心另一个线程会修改对象的状态. 如果要更新这些变量, 那么可以创建一个新的容器对象, 但其他使用原有对象的线程仍然会看到对象处于一致的状态.

安全发布

在某些情况下希望在多个线程之间共享对象, 此时必须确保对象能够安全地发布, 不正确的发布会导致其他线程看到尚未创建完成的对象. 尚未创建完成地对象不拥有完整性, 某个观察该对象的线程将看到对象处于不一致的状态, 然后看到对象的状态突然发生变化, 即使线程在对象发布后还没有修改过它.

即使某个对象的引用对其他线程是可见的, 也并不意味着对象状态对于使用该对象的线程来说一定是可见的. 因此为了确保对象状态能呈现出一致的视图, 就必须使用同步.

另一方面, 即使在发布不可变对象的引用时没有使用同步, 也仍然可以安全地访问该对象. 为了维持这种初始化安全性地保证, 必须满足不可变地所有需求: 状态不可修改, 所有域都是final类型, 以及正确的构造过程.

这种保证还将延伸到被正确创建对象中所有地final类型的域. 在没有额外同步的情况下, 也可以安全地访问final类型的域. 然而, 如果final类型的域所指向的是可变对象, 那么在访问这些域所指向的对象的状态时仍然需要同步.

任何线程都可以在不需要额外同步的情况下安全地访问不可变对象, 即使在发布这些对象时没有使用同步.

要安全地发布一个对象, 对象的引用以及对象的状态必须同时对其他线程可见, 一个正确构造的对象可通过以下方式来安全地发布:

1. 在静态初始化函数中初始化一个对象引用.
2. 将对象的引用保存到volatile类型的域或者AtomicReference对象中.
3. 将对象的引用保存到某个正确构造对象的final类型域中.
4. 将对象的引用保存到一个由锁保护的域中.

将对象放入到线程安全的容器内满足上述的最后一条需求. 线程安全库中的容器类提供了安全发布保证.

类库中的数据传递机制, 如Future和Exchanger同样能实现安全发布.

要发布一个静态构造的对象, 最简单和最安全的方式是使用静态的初始化器. 静态初始化器是由JVM在类的初始化阶段执行, 由于在JVM内部存在着同步机制, 因此通过这种方式初始化的任何对象都可以被安全地发布.

如果对象从技术上来看是可变的, 但其状态在安全发布后不会再改变, 那么这种对象称为事实不可变对象. 在这些对象发布后, 程序只需将它们视为不可变对象即可. 在没有额外的同步的情况下, 任何线程都可以安全地使用被安全发布的事实不可变对象.

对于可变对象, 不仅在发布对象时需要使用同步, 而且在每次对象访问时同样需要使用同步来确保后续修改操作的可见性.

对象的发布需求取决于它的可变性:

• 不可变的对象可以通过任意机制来发布.
• 事实不可变对象必须通过安全方式来发布.
• 可变对象必须通过安全方式来发布, 并且必须是线程安全的或者由某个锁保护起来.

在并发程序中使用和共享对象时, 可以使用一些实用的策略, 包括:

1. 线程封闭: 线程封闭的对象只能由一个线程拥有, 对象被封闭在该线程中, 并且只能由这个线程修改.
2. 只读共享: 在没有额外同步的情况下, 共享的只读对象可以由多个线程并发访问, 但任何线程都不能修改它. 共享的只读对象包括不可变对象和事实不可变对象.
3. 线程安全共享: 线程安全的对象在其内部实现同步, 因此多个线程可以通过对象的公有接口来进行访问而不需要进一步的同步.
4. 保护对象: 被保护的对象只能通过持有特定的锁来访问. 保护对象包括封装在其他线程安全对象中的对象, 以及已发布的并且由某个特定锁保护的对象.

对象的组合

本节介绍一些组合模式, 通过这些模式能够使一个类更容易成为线程安全的, 并且在维护这些类时不会无意中破坏类的线程安全性保证.

设计线程安全的类

通过使用封装技术, 可以使得在不对整个程序进行分析的情况下就可以判断一个类是否是线程安全的.

在设计线程安全类的过程中, 需要包含以下三个基本要素:

1. 找出构成对象状态的所有变量.
2. 找出约束状态变量的不变性条件.
3. 建立对象状态并发访问管理策略.

要分析对象的状态, 首先从对象的域开始. 如果对象中所有域都是基本类型的变量, 那么这些域将构成对象的全部状态. 对于含有n个基本类型的对象, 其状态就是这些域构成的n元组.

同步策略定义了如何在不违背对象不变条件或后验条件的情况下对其状态的访问操作进行协同. 同步策略规定了如何将不变性, 线程封闭与加锁机制等结合起来以维护线程的安全性, 并且还规定了哪些变量由哪些锁来保护.

要确保类的线程安全性, 就需要确保它的不变条件不会在并发访问的情况下被破坏, 这就需要对其状态进行判断. 对象与变量都有一个状态空间, 即所有可能的取值. 状态空间越小, 就越容易判断线程的状态. final类型的域使用得越多, 就越能简化对象可能状态的分析过程.

在许多类中都定义了一些不可变条件, 用于判断状态是有效的还是无效的. 同样在操作中还会包含一些后验条件来判断状态迁移是否是有效的.

由于不变性条件以及后验条件在状态及状态转换上施加了各种约束, 因此就需要额外的同步与封装. 如果某些状态是无效的, 那么必须对底层的状态变量进行约束, 否则客户代码可能会使对象处于无效状态. 如果在某个操作中存在无效的状态转换, 那么该操作必须是原子的.

在类中可以包含同时约束多个状态变量的不变性条件. 这些相关的变量必须在单个原子操作中进行读取或更新, 不能首先更新一个变量, 然后释放锁并再次获得锁, 然后再更新其他的变量. 因为释放锁后, 可能会使对象处于无效状态. 如果在一个不变性条件中包含多个变量, 那么在执行任何访问相关变量的操作时, 都必须持有保护这些变量的锁.

如果不了解对象的不变性条件与后验条件, 那么就不能确保线程安全性. 要满足在状态变量的有效值或状态转换上的各种约束条件, 就需要借助于原子性与封装性.

如果在某个操作中包含有基于状态的先验条件, 那么这个操作就被称为依赖状态的操作. 在并发程序中要一直等到先验条件为真, 然后在执行该操作. 要实现依赖状态的操作, 更简单的方法是使用现有库中的类.

许多情况下, 所有权和封装性总是相互关联的, 对象封装它拥有的状态, 反之对象对它封装的状态拥有所有权. 状态变量的所有者将决定采用何种加锁协议来维持变量状态的完整性, 所有权意味着控制权. 然而如果发布了某个可变对象的引用, 那么就不再拥有独占的控制权, 最多是共享控制权. 对于从构造函数或者从方法中传递进来的对象, 类通常并不拥有这些对象, 除非这些方法是被专门设计为转移传递进来的对象的所有权.

实例封闭

封装简化了线程安全类的实现过程, 提供了一种实例封闭机制. 当一个对象被封装到另一个对象中时, 能够访问封装对象的所有代码路径都是已知的, 与对象可以由整个程序访问的情况相比, 这更易于对代码进行分析. 通过将封闭机制与合适的加锁策略结合起来, 可以确保以线程安全的方式来使用非线程安全的对象.

将数据封装在对象内部, 可以将数据的访问限制在对象的方法上, 从而更容易确保线程在访问数据时总能持有正确的锁.

被封闭对象一定不能超出它们既定的作用域. 对象可以封闭在类的一个实例中(类的私有成员), 或者封闭在某个作用域内(作为一个局部变量), 或者封闭在线程内(在线程中把对象从一个方法传递到另一个方法).

实例封闭是构建线程安全的一个最简单方式, 它使得在锁策略的选择上拥有了更多的灵活性. 可以使用对象的内置锁来保护状态, 但对于其他形式的锁来说, 只要自始至终都使用同一个锁, 就可以保护状态. 实例封闭还使得不同的状态变量可以由不同的锁来保护.

Java平台的类库由很多线程封闭的示例. 如Collections.synchronizedXXX方法. 这些工厂方法通过装饰器模式将容器类封装在一个同步的包装器对象中, 而包装器能将接口中的每个方法都实现为同步方法, 并将调用请求转发到底层的容器对象上. 只要包装器对象拥有对底层容器对象的唯一引用, 那么它就是线程安全的, 对底层容器对象的所有访问必须通过包装器类进行.

当然如果将一个本该被封闭的对象发布出去, 那么也能破坏封闭性. 如果一个对象本应该封闭在特定的作用域内, 那么让该对象逸出作用域就是一个错误. 当发布其他对象时, 可能会间接地发布被封闭对象, 同样会使被封闭对象逸出.

封闭机制更易于构造线程安全类, 因为当封闭类的状态时, 在分析类的线程安全性时就无需检查整个程序.

从线程封闭原则及其逻辑推论可以得出Java监视器模式. 遵循Java监视器模式的对象会把对象的所有可变状态都封装起来, 并由对象自己的内置锁来保护. 如下面的代码使用私有锁来保护状态:

public class PrivateLock {
    private final Object myLock = new Object();
    void someMethod() {
        synchronized(myLock) {
            //...
        }
    }
}

使用私有的锁对象而不是对象的内置锁, 有许多优点. 私有的锁对象可以将锁封装起来, 使客户代码无法得到锁, 但客户代码可以通过公有方法来访问锁, 以便参与到它的同步策略中. 如果客户代码错误地获得了另一个对象地锁, 那么可能会产生活跃性问题. 此外, 也无需检查整个程序来验证锁是否被正确地使用.

线程安全性的委托

大多数对象都是组合对象, 当从头开始构建一个类, 或者将多个非线程安全的类组合为一个类时, Java监视器模式是非常有用的. 但是如果类中的各个组件都已经是线程安全的, 需要视情况进行分析.

下面是委托于线程安全Map的代码示例:

@Immutable
public class Point {
    private final int x, y;

    public Point(int x, int y) {
        this.x = x;
        this.y = y;
    }
}
@ThreadSafe
public class DelegatingVehicleTracker {

    private final ConcurrentMap<String, Point> locations;

    private final Map<String, Point> unmodifiedMap;

    public DelegatingVehicleTracker(Map<String, Point> points) {
        locations = new ConcurrentHashMap<>(points);
        // unmodifiableMap方法只是包装了原对象,原对象的内容改变,unmodifiedMap也会相应进行改变
        unmodifiedMap = Collections.unmodifiableMap(locations);
    }

    public Map<String, Point> getLocations() {
        return unmodifiedMap;
    }

    public Point getLocation(String id) {
        return locations.get(id);
    }

    public void setLocation(String id, int x, int y) {
        if (locations.replace(id, new Point(x, y)) == null) {
            throw new IllegalArgumentException("Invalid vehicle name: " + id);
        }
    }
}

上述代码中, 该使用委托的车辆追踪器中返回的是一个不可修改但却实时的车辆位置视图. 如果需要一个不发生变化的车辆视图, 那么getLocations可以返回对locations这个Map的浅拷贝. 由于Map的内容是不可变的, 因此只需要复制Map的结构, 而不用复制它的内容. 如下代码所示:

    public Map<String, Point> getLocations() {
        return Collections.unmodifiableMap(new HashMap<>(locations));
    }

还可以将线程安全性委托给多个状态变量, 只要这些变量是彼此独立的, 即组合而成的类并不会在其包含的多个状态变量上增加任何不变性条件. 如下面的代码:

public class VisualComponent {
    private final List<KeyListener> keyListeners = new CopyOnWriteArrayList<>();
    private final List<MouseListener> mouseListeners = new CopyOnWriteArrayList<>();
    public void addKeyListener(KeyListener listener) {
        keyListeners.add(listener);
    }
    public void addMouseListener(MouseListener listener) {
        mouseListeners.add(listener);
    }
}

但大多数组合对象不会像上述代码的VisualComponent这么简单, 在它们的状态变量之间存在着某些不变性条件. 如下面的代码中的NumberRange, 其约束条件是第一个数值要小于第二个数值:

public class NumberRange {
    // 刚开始是[0, 10]
    private final AtomicInteger lower = new AtomicInteger(0);
    private final AtomicInteger upper = new AtomicInteger(0);
    // 一个线程设置lower为5
    public void setLower(int i) {
        if (i > upper.get()) {
            throw new IllegalArgumentException("can't set lower to " + i + " > upper");
        }
        lower.set(i);
    }
    // 另一个线程设置upper为4
    public void setUpper(int i) {
        if (i < lower.get()) {
            throw new IllegalArgumentException("can't set upper to " + i + " > upper");
        }
        upper.set(i);
    }

    public boolean isInRange(int i) {
        return i >= lower.get() && i <= upper.get();
    }
}

NumberRange不是线程安全的, 没有维持对下界和上界进行约束的不变性条件.

如果某个类含有复合操作, 那么仅靠委托并不足以实现线程安全性. 在这种情况下, 这个类必须提供自己的加锁机制以保证这些复合操作都是原子操作, 除非整个复合操作都可以委托给状态变量.

如果一个类是由多个独立且线程安全的状态变量组成, 并且在所有的操作中都不包含无效状态转换, 那么可以将线程安全性委托给底层的状态变量.

当把线程安全性委托给某个对象的底层状态变量时, 可以发布这些变量从而使其他类能修改它们的情况取决于在类中对这些变量施加了哪些不变条件.

如果一个状态变量是线程安全的, 并且没有任何不变性条件来约束它的值, 在变量的操作上也不存在任何不允许的状态转换, 那么就可以安全地发布这个变量.

在现有的线程安全类中添加功能

Java类库中包含许多有用的基础模块类, 通常应该优先选择重用这些现有类而不是创建新的类. 虽然现有的线程安全类能支持大部分的操作, 但有时却需要在不破坏线程安全性的情况下添加一个新的操作. 比如一个线程安全的链表, 需要提供一个若没有则添加的原子操作.

要添加一个新的原子操作, 最安全的方法是直接修改原始的类, 但通常无法做到, 因为可能无法访问或修改类的源代码. 要想修改原始的类, 就需要理解代码中的同步策略, 这样增加的功能才能与原有的设计保持一致.

另一种方法是扩展这个类, 但并非所有的类都向子类公开. 扩展方法比直接将代码添加到类中更加脆弱, 因为现在的同步策略实现被分布到多个单独维护的源代码文件中. 如果底层的类改变了同步策略并选择了不同的锁来保护它的状态变量, 那么子类会被破坏, 因为在同步策略改变后它无法再使用正确的锁来控制对基类状态的并发访问.

第三中策略是扩展类的功能, 如下面的代码:

public class ListHelper<E> {
    public List<E> list = Collections.synchronizedList(new ArrayList<E>());
    
    public synchronized boolean putIfAbsent(E x) {
        boolean absent = !list.contains(x);
        if (absent) {
            list.add(x);
        }
        return absent;
    }
}

但是上面的代码是错误的. 因为ListHelper同步方法putIfAbsent使用的锁不是Collections.synchronizedList方法返回的List使用的锁, 因此putIfAbsent不是原子操作. 要想使这个方法能正确执行, 必须使List在实现客户端加锁或外部加锁时使用同一个锁. 需修改putIfAbsent为下面的代码:

    public boolean putIfAbsent(E x) {
        synchronized(list) {
            boolean absent = !list.contains(x);
            if (absent) {
                list.add(x);
            }
            return absent;
        }
    }

通过添加一个原子操作类扩展类是脆弱的, 因为它将类的加锁代码分布到多个类中. 然而客户端加锁却更加脆弱, 因为它将类的加锁代码放到与该类完全无关的其他类中.

客户端加锁机制与扩展类机制有许多共同点, 二者都是将派生类的行为与基类的实现耦合在一起, 正如扩展会破坏实现的封装性, 客户端加锁同样会破坏同步策略的封装性.

当为现有的类添加一个原子操作时, 有一种更好的方法: 组合. 如下面的代码:

public class ImprovedList<T> implements List<T> {

    private final List<T> list;

    public ImprovedList(List<T> list) {
        this.list = list;
    }

    public synchronized boolean putIfAbsent(T x) {
        boolean absent = !list.contains(x);
        if (absent) {
            list.add(x);
        }
        return absent;
    }
    ...
}

ImprovedList通过自身的内置锁增加了一层额外的加锁, 它并不关系底层的List是否是线程安全的, 即使List不是线程安全的或者修改了它的加锁实现, ImprovedList也会提供一致的加锁机制来实现线程安全性, 虽然额外的同步层可能导致轻微的性能损失, 但与模拟另一个对象的加锁策略相比, ImprovedList更为健壮.

将同步策略文档化

在文档中说明客户代码需要了解的线程安全性保证, 以及代码维护人员需要了解的同步策略.

Java同步策略的关键字的使用是程序设计的要素之一, 应该将其文档化. 在编写设计决策文档之后的几周或几个月内, 需要在忘记之前将它们记录下来.

在设计同步策略时考虑的多个方面, 锁的保护说明都是严格的实现细节, 应该将它们文档化以便于日后的维护.

最起码应该保证将类中的线程安全性文档化. 如果希望客户代码能够在类中添加新的原子操作, 那么就需要在文档中说明需要获得哪些锁才能实现安全的原子操作. 如果使用锁来保护状态, 那么也要将其写入文档以便日后维护. 如果要使用更复杂的方法来维护线程安全性, 那么一定要将它们写入文档, 因为维护者很难发现它们.

基础构建模块

Java平台类库包含了丰富的并发基础构建模块.

同步容器类

同步容器类包括Vector和Hashtable, 还包括同步封装器类是由Collections.synchronizedXXX等工厂方法创建的. 这些类实现线程安全的方式是: 将它们的状态封装起来, 并对每个公有方法都进行同步, 使得每次只有一个线程能访问容器的状态.

同步容器类都是线程安全的, 但在某些情况下可能需要额外的客户端加锁来保护复合操作, 包括迭代, 跳转以及条件运算等. 由于同步容器类要遵守同步策略, 即支持客户端加锁, 因此可能会创建一些新操作, 只要知道应该使用哪一个锁, 那么这些新操作就与容器的其他操作一样都是原子操作.

Java中对容器进行迭代的标准方式是使用Iterator, 然而如果有其他线程并发地修改容器, 那么即使是使用迭代器也无法避免在迭代期间对容器加锁. 在设计同步容器类的迭代器时并没有考虑到并发修改地问题, 并且它们表现出的行为是及时失败的. 这意味着, 当它们发现容器在迭代过程中被修改时, 就会抛出一个ConcurrentModificationException异常.

如果不希望在迭代期间对容器加锁, 那么一种替代方法就是克隆容器, 并在副本上进行迭代. 由于副本被封闭在线程内, 因此其他线程不会在迭代期间对其进行修改, 这样就避免了抛出ConcurrentModificationException异常. 在克隆容器时存在显著的性能开销. 这种方式的好处取决于多个因素, 包括容器的大小, 在每个元素上执行的工作, 迭代操作相对于容器其他操作的调用频率, 以及在响应事件和吞吐量等方面的需求.

在某些情况下, 迭代器会隐藏起来, 比如调用容器类的toString, equals, hashCode方法时, 会调用迭代器. 如果状态与保护它的同步代码之间相隔越远, 那么开发人员就越容易忘记在访问状态时使用正确的同步.

正如封装对象的状态有助于维持不变性条件一样, 封装对象的同步机制同样有助于确保实施同步策略.

并发容器

Java提供了多种并发容器类来改进同步容器的性能.

ConcurrentHashMap用来替代同步且基于散列的Map. ConcurrentHashMap使用一种粒度更细的加锁机制来实现更大程度的共享, 这种机制称为分段锁. 这种机制下, 任意数量的读取线程可并发地访问Map, 执行读取操作的线程和执行写入操作的线程可以并发地访问Map, 并且一定数量地写入线程可以并发地修改Map. ConcurrentHashMap带来的结果是在并发访问环境下有更高的吞吐量, 而在单线程环境中只损失非常小的性能.

ConcurrentHashMap提供的迭代器不会抛出ConcurrentModificationException异常, 因此不需要在迭代过程中对容器加锁, ConcurrentHashMap返回的迭代器具有弱一致性, 而并非及时失败.. 弱一致性的迭代器可以容忍并发的修改, 当创建迭代器时会遍历已有的元素, 并可以(但是不保证)在迭代器被构造后将修改操作反映给容器.

尽管有以上的改进, 但仍然有一些需要权衡的因素, 对于一些需要在整个Map上进行计算的方法, 如size和isEmpty, 这些方法的语义被略微减弱了以反映容器的并发特性. 由于size返回的结果在计算时可能已经过期了, 它实际上只是一个估计值, 因此允许size返回一个近似值而不是一个精确值. 事实上size和isEmpty方法在并发环境的用处很小, 因为它们的返回值总在不断变化. 因此这些操作的需求被弱化了, 以换取其他更重要操作的性能优化.

与Hashtable和synchronizedMap相比, ConcurrentHashMap有着更多的优势以及更少的劣势, 因此在大多数情况下, 用ConcurrentHashMap来代替同步Map能进一步提高代码的可伸缩性. 只有当应用程序需要加锁Map以进行独占访问时, 才应该放弃使用ConcurrentHashMap.

常见的复合操作, 如若没有则添加, 若相等则移除, 若相等则替换等, 都已经实现为原子操作并且在ConcurrentMap的接口中声明.

CopyOnWriteArrayList用于在遍历操作为主要操作的情况下代替同步的List. 在某些情况下提供了更好的并发性能, 并且在迭代期间不需要对容器进行加锁或复制.

写入时复制容器的线程安全性在于, 只要正确地发布一个事实不可变的对象, 那么在访问该对象时就不再需要进一步的同步. 在每次修改时, 都会创建并重新发布一个新的容器副本, 从而实现可变性. 写入时复制容器的迭代器保留一个指向底层基础数组的引用, 这个数组当前位于迭代器的起始位置, 由于它不会被修改, 因此在对其进行同步时只需确保数组内容的可见性, 因此多个线程可以同时对这个容器进行迭代, 而不会彼此干扰或者与修改容器的线程相互干扰. 写入时容器返回的迭代器不会抛出ConcurrentModificationException异常, 并且返回的元素与迭代器创建时的元素完全一致, 而不必考虑之后修改操作所带来的影响.

显然每当修改容器时都会复制底层数组, 这需要一定的开销, 特别是容器的规模较大时. 仅当迭代操作远远多于修改操作时, 才应该使用写入时复制容器.

Queue的实现ConcurrentLinkedQueue是一个先进先出队列, 以及PriorityQueue是非并发的优先级队列.

BlockingQueue扩展了Queue, 增加了可阻塞的插入和获取等操作.

ConcurrentSkipListMap和ConcurrentSkipListSet分别作为同步的SortedMap和SortedSet的并发替代品.

阻塞队列和生产者消费者模式

阻塞队列提供了可阻塞的put和take方法, 以及支持定时的offer和poll方法, 还有抛出异常的add和remove方法. 队列可以是有界的也可以是无界的.

阻塞队列支持生产者-消费者这种设计模式. 该模式将找出需要完成的工作与执行工作这两个过程分离开来, 并把工作项放入一个待完成列表中以便在随后处理, 而不是找出后立即处理. 生产者-消费者模式能够简化开发过程, 因为它消除了生产者类和消费者类之间的代码依赖性. 此外, 该模式还将生产数据的过程与使用数据的过程解耦开来以简化工作负载的管理, 因为这两个过程在处理数据的速率上有所不同.

在构建高可靠的应用程序时, 有界队列是一种强大的资源管理工具, 它们能抑制并防止生产过多的工作项, 使应用程序在负荷过载的情况下变得更加健壮.

BlockingQueue有多种实现, LinkedBlockingQueue和ArrayBlockingQueue是FIFO队列, 二者分别与LinkedList和ArrayList类似, 但比同步List有更好的并发性能. PriorityBlockingQueue是一个按优先级排序的队列, 可以根据元素的自然顺序来比较元素, 也可以使用Comparator来比较.

SynchronousQueue实际上不是一个队列, 因为它不会为队列中元素维护存储空间, 与其他队列不同的是, 它维护一组线程, 这些线程在等待这把元素加入或移除队列. 由于可以直接交付工作, 从而降低了将数据从生产者移动到消费者的延迟. 直接交付方式会将更多关于任务状态的信息反馈给生产者, 当交付被接受时, 生产者就知道消费者已经得到了任务. 仅当有足够多的消费者, 并且总是有一个消费者准备好获取交付的工作时, 才适合使用同步队列.

生产者-消费者模式能带来许多性能优势, 生产者和消费者可以并发地执行. 如果一个是I/O密集型, 另一个是CPU密集型, 那么并发执行的吞吐率要高于串行执行的吞吐率. 如果生产者和消费者的并行度不同, 那么将它们紧密耦合在一起会把整体并行度降低为二者中更小的并行度.

对于可变对象, 阻塞队列和生产者-消费者模式在一起促进了线程封闭, 从而将对象所有权从生产者交付给消费者. 线程封闭对象只能由单个线程拥有, 但可以通过安全地发布该对象来转移所有权. 在转移所有权后, 也只有另一个线程能获得这个对象的访问权限, 并且发布对象的线程不会再访问它. 这种安全的发布确保了对象状态对于新的所有者来说是可见的, 并且由于最初的所有者不会再访问它, 因此对象将被封闭在新的线程中, 新的所有者线程可以对该对象做任意修改, 因为它具有独占的访问权.

可以使用其他的发布机制来传递对象的所有权, 但必须确保只有一个线程能接受被转移的对象, 除了阻塞队列外, 还可以通过ConcurrentMap的原子方法remove, 或者AtomicReference的原子方法compareAndSet.

Deque是一个双端队列, 实现了在队列头和队列尾的高效插入和移除, 具体实现包括ArrayDeque和LinkedBlockingDeque. 双端队列适用于工作窃取模式.

在工作窃取设计中, 每个消费者都有各自的双端队列. 如果一个消费者完成了自己双端队列中的全部工作, 那么它可以从其他消费者双端队列末尾秘密地窃取工作. 密取工作模式比传统地生产者-消费者模式具有更高地可伸缩性, 因为工作者线程不会在单个共享的任务队列上发生竞争. 在大多数时候, 它们都只是访问自己的双端队列, 从而极大地减少了竞争. 当工作者线程需要访问另一个队列时, 它会从队列的尾部而不是从头部获取工作, 因此进一步降低了队列上的竞争程度.

工作窃取非常适用于既是消费者也是生产者问题: 当执行某个工作时可能导致出现更多的工作. 当一个工作线程找到新的任务单元时, 它会将其放到自己队列的末尾或者在工作共享设计模式下放入其他工作者线程的队列中. 当双端队列为空时, 它会在另一个线程的队列队尾查找新的任务, 从而确保每个线程都保持忙碌状态.

阻塞方法和中断方法

线程可能会阻塞, 当线程阻塞时, 它通常被挂起, 并处于某种阻塞状态. 被阻塞的线程必须等待某个不受它控制的事件发生后才能继续执行.

当某方法抛出InterruptedException时, 表示该方法是一个阻塞方法, 如果这个方法被中断, 那么它将努力提前结束阻塞状态.

Thread提供了interrupt方法用于中断线程或者查询线程是否已经被中断. 每个线程都有一个布尔类型的属性, 表示线程的中断状态, 当中断线程时将设置这个状态.

中断是一种协作机制. 一个线程不能强制其他线程停止正在执行的操作而去执行其他的操作. 当线程A中断B时, A仅仅是要求B在执行到某个可以暂停的地方停止正在执行的操作, 前提是如果线程B愿意停止下来. 最常使用中断的情况就是取消某个操作, 方法对中断请求的响应度越高, 就越容易及时取消那些执行时间很长的操作.

当在代码中调用了一个将抛出InterruptedException异常的方法时, 你自己的方法也就变成了一个阻塞方法, 并且必须处理对中断的响应, 有两种基本选择:

1. 传递InterruptedException, 避开这个异常通常是最明智的策略, 只需把InterruptedException传递给方法调用者. 传递InterruptedException的方法包括: 根本不捕获该异常, 或者捕获该异常然后在执行某种简单的清理工作后再次抛出这个异常.
2. 恢复中断, 有时候不能抛出InterruptedException, 必须捕获InterruptedException, 并通过调用当前线程上的interrupt方法恢复中断状态, 这样在调用栈中更高层的代码将看到引发了一个中断.

然而出现InterruptedException时不应该做的事情是捕获它而不做出任何响应. 这将使调用栈上更高层的代码无法对中断采取处理措施, 因为线程被中断的证据已丢失.

同步工具类

同步工具类可以是任何一个对象, 只要它根据其自身的状态来协调线程的控制流. 除了阻塞队列, 还有信号量, 栅栏, 以及闭锁等同步工具类.

所有的同步工具类都包含一些特定的结构化属性, 它们封装了一些状态, 这些状态将决定执行同步工具类的线程是继续执行还是等待, 此外还提供了一些方法对状态进行操作, 以及一些方法用于高效地等待同步工具类进入到预期的状态.

闭锁可以延迟线程的进度直到其到达终止状态. 闭锁的作用相当于一扇门, 在闭锁到达结束状态之前, 这扇门是一直关闭的, 并且没有任何线程能通过. 当达到结束状态时, 这扇门会打开并允许所有的线程通过. 当闭锁到达结束状态后, 将不会再改变状态, 因为这扇门将永远保持打开状态. 闭锁可以用来确保某些获得直到其他活动都完成后才继续执行.

CountDownLatch是一种灵活的闭锁实现, 可以使一个或多个线程等待一组事件发生. 闭锁状态包括一个计数器, 该计数器初始化为一个正数, 表示需要等待的事件数量. countDown方法递减计数器, 表示有一个事件以及发生了, 而await方法等待计数器达到零, 这表示所有需要等待的事件都已经发生. 如果计数器的值非零, 那么await会一直阻塞直到计数器为零, 或者等待中的线程中断, 或者等待超时.

FutureTask可以用作闭锁. FutureTask表示的计算是通过Callable来实现的, 相当于一种可生成结果的Runnable, 并且可以处于三种状态: 等待运行, 正在运行, 运行完成. 执行完成表示计算的所有可能结束方式, 包括正常结束, 由于取消而结束和由于异常而结束等. 当FutureTask进入完成状态后, 会永远停止在这个状态上.

Future.get的行为取决于任务的状态, 如果任务已经完成, 那么get会立即返回结果, 否则get将阻塞直到任务进入完成状态, 然后返回结果或抛出异常. FutureTask将计算结果从执行计算的线程传递到获取这个结果的线程, 而FutureTask的规范确保了这种传递过程能实现结果的安全发布.

信号量中管理着一组虚拟的许可, 许可的初始数量可通过构造函数来指定它. 在执行操作时可以首先获得许可, 并在使用以后释放许可. 如果没有许可, 那么acquire将阻塞直到有许可或者直到被中断或者操作超时. release方法将返回一个许可给信号量.

计数信号量可以用来控制访问某个特定资源的操作数量, 或者同时执行某个指定操作的数量, 还可以用来实现某种资源池, 或者对容器施加边界. 计算信号量的一种简化形式是二值信号量, 即初始值为1的信号量. 二值信号量可以用作互斥体, 并具备不可重入的加锁语义: 谁拥有这个唯一的许可, 谁就拥有了互斥锁.

栅栏类似于闭锁, 能阻塞一组线程直到某个事件发生. 栅栏和闭锁的关键区别在于, 所有线程必须同时到达栅栏位置, 才能继续执行, 闭锁用于等待事件, 而栅栏用于等待其他线程.

CyclicBarrier可以使一定数量的参与方反复地在栅栏位置汇集, 其在并行迭代算法中非常有用: 这种算法通常将一个问题拆分成一系列相互独立的子问题.

当线程到达栅栏位置时将调用await方法, 这个方法将阻塞直到所有线程都到达栅栏位置. 如果所有线程都到达了栅栏位置, 那么栅栏将打开, 此时所有线程都被释放, 而栅栏将被重置以便下次使用.

如果对await的调用超时, 或者await阻塞的线程被中断, 那么栅栏就被认为是打破了, 所有阻塞的await调用都将终止并抛出BrokenBarrierException.

如果成功地通过栅栏, 那么await将为每个线程返回一个唯一的到达索引号. 可以利用这些索引号来选举产生一个领导线程, 并在下一次迭代中由该领导线程执行一些特殊的工作.

CyclicBarrier还可以使你将一个栅栏操作传递给构造函数, 这是一个Runnable, 当成功通过栅栏时会在一个子任务线程中执行它, 但在阻塞线程被释放之前是不能执行的.

另一种形式的栅栏是Exchanger, 它是一种两方栅栏, 各方在栅栏位置上交换数据. 当两方执行不对称的操作时, Exchanger会非常有用.

例如当一个线程向缓冲区写入数据, 而另一个线程从缓冲区读取数据. 这些线程可以使用Exchanger来汇合, 并将满的缓冲区与空的缓冲区交换. 当两个线程通过Exchanger交换对象时, 这种交换就把两个对象安全地发布给另一方.

数据交换的时机取决于应用程序的响应需求, 最简单的方案是当缓冲区被填满时, 由填充任务进行交换, 当缓冲区为空时, 由清空任务进行交换. 这样会把需要交换的次数降至最低, 但如果新数据的到达率不可预测, 那么一些数据的处理过程就将延迟. 另一个方法是不仅当缓冲被填满时进行交互, 并且当缓冲被填充到一定程度并保持一定时间后, 也进行交换.

构建高效且可伸缩的结果缓存

本节将开发一个高效且可伸缩的缓存, 用于改进一个高计算开销的函数.

@FunctionalInterface
public interface Computable<A, V> {
    V compute(A arg) throws InterruptedException;
}
public class ExpensiveFunction implements Computable<String, BigInteger> {
    @Override
    public BigInteger compute(String arg) throws InterruptedException {
        // 使用sleep模拟高开销计算
        TimeUnit.SECONDS.sleep(10);
        return new BigInteger(arg);
    }
}
public class CacheMemory<A, V> implements Computable<A, V> {
    private final ConcurrentMap<A, Future<V>> cache = new ConcurrentHashMap<>();
    private final Computable<A, V> c;
    public CacheMemory(Computable<A, V> c) {
        this.c = c;
    }
    @Override
    public V compute(A arg) throws InterruptedException {
        while (true) {
            Future<V> f = cache.get(arg);
            if (f == null) {
                Callable<V> eval = () -> c.compute(arg);
                FutureTask<V> ft = new FutureTask<>(eval);
                f = cache.putIfAbsent(arg, ft);
                if (f == null) {
                    f = ft;
                    ft.run();
                }
            }
            try {
                return f.get();
            } catch (CancellationException e) {
                cache.remove(arg, f);
            } catch (ExecutionException e) {
                e.printStackTrace();
                cache.remove(arg, f);
                throw new RuntimeException(e);
            }
        }
    }
}

使用ConcurrentHashMap容器来提升并发性能. 使用FutureTask表示计算过程, 可以避免其他线程重复计算. 使用putIfAbsent方法实现如果没有则添加的原子操作, 并且如果添加成功则返回null, 否则返回原来的value.

使用while循环的原因是避免Future造成的缓存污染问题. 如果某个计算被取消或者失败, 那么在计算这个结果时将指明计算过程被取消或者失败. 为了避免这种情况, 如果发现计算被取消, 那么将把Future从缓存中移除. 如果检测到RuntimeException, 那么也会移除Future, 这样将来的计算才可能成功.

上述的实现没有解决缓存逾期的问题, 可以通过使用FutureTask的子类来解决. 推荐使用Caffeine来实现.

任务执行

大多数并发应用程序是围绕任务执行来构造的, 任务通常是一些抽象的且离散的工作单元. 通过把应用程序的工作分解到多个任务中, 可以简化程序的组织结构, 提供一种自然的事务边界来优化错误恢复过程, 以及提供一种自然的并行工作结构来提升并发性.

在线程中执行任务

当围绕任务执行来设计应用程序结构时, 第一步就是要找出清晰的任务边界. 在理想情况下, 各个任务之间是相互独立的, 任务并不依赖于其他任务的状态, 结果或者边界效应. 独立性有助于实现并发, 如果存在足够多的处理资源, 那么这些独立的任务都可以并行执行. 为了在调用与负载均衡等过程中实现更高的灵活性, 每项任务还应该表示应用程序的一小部分处理能力.

在正常的负载下, 服务器应用程序应该同时表现出良好的吞吐量和快速的响应性. 应用程序提供商希望程序支持尽可能多的用户, 从而降低每个用户的服务成本, 而用户则希望获得尽快的响应. 而且当负荷过载时, 应用程序的性能应该是逐渐降低, 而不是直接失败. 要实现上述目标, 应该选择清晰的任务边界以及明确的任务执行策略.

大多数服务器应用程序都提供了一种自然的任务边界选择方式, 以独立的客户请求为边界. 将独立的请求作为任务边界, 既可以实现任务的独立性, 又可以实现合理的任务规模.

在应用程序中可以通过多种策略来调度任务, 最简单的策略就是在单个线程中串行地执行各项任务. 但是串行处理机制通常都无法提供高吞吐率或快速响应性, 服务器的资源利用率低.

可以通过为每个请求创建一个新的线程来提供服务, 从而实现更高的响应性. 但这种方法存在一些缺陷, 尤其是当需要创建大量的线程时:

1. 线程生命周期的开销非常高. 线程的创建过程都会需要时间, 延迟处理的请求. 如果请求的到达率非常高且请求的处理过程是轻量级的, 那么为每一个请求创建一个新线程将消耗大量的计算资源.
2. 资源消耗. 活跃的线程会消耗系统资源, 尤其是内存. 如果可运行的线程数量多于可用处理器的数量, 那么有些线程将闲置. 大量线程给垃圾回收器带来压力, 且在竞争CPU资源时还将产生上下文切换等性能开销. 如果已经有足够多的线程使所有CPU保持忙碌状态, 那么再创建更多的线程反而会降低性能.
3. 稳定性. 可创建的线程数量上存在一个限制. 这个限制随着平台的不同而不同, 并且受多个因素制约, 如JVM的启动参数, Thread构造函数中请求的栈大小, 以及底层操作系统对线程的限制等. 如果破坏了这些限制, 那么很可能抛出OutOfMemoryError异常, 要想从这种错误中恢复过来是非常危险的.

在一定的范围内, 增加线程可以提高系统的吞吐率, 但如果超出了这个范围, 再创建更多的线程只会降低程序的执行速度, 如果过多的创建线程, 那么整个应用程序将崩溃. 为每个任务分配一个线程这种方法的问题在于, 它没有限制可创建线程的数量, 只限制了远程用户提交HTTP请求的速率. 如果有某个恶意的用户或者过多的用户, 都会使Web服务器的负载达到某个阈值, 从而使服务器崩溃.

Executor框架

任务是一组逻辑工作单元, 而线程则是使任务异步执行的机制. 线程池简化了线程的管理工作, 并且Java类库提供了一种灵活的线程池实现作为Executor框架的一部分.

public interface Executor {
    void execute(Runnable command);
}

虽然Executor是一个简单的接口, 但它却为灵活且强大的异步任务执行框架提供了基础, 该框架能够支持多种不同类型的任务执行策略. 它提供了一种标准的方法将任务的提交过程与执行过程解耦开来, 并用Runnable表示任务. Executor的实现还提供了对生命周期的支持, 以及统计信息收集, 应用程序管理机制和性能监视等机制.

Executor基于生产者-消费者模式, 提交任务的操作相当于生产者(生产待完成的工作单元), 执行任务的线程则相当于消费者(执行完这些工作单元). 如果要在程序中实现一个生产者-消费者的设计, 那么最简单的方式就是使用Executor.

通过将任务的提交与执行解耦开来, 从而轻易地为某种类型的任务进行指定和修改执行策略. 各种执行策略都是一种资源管理工具, 最佳策略取决于可用的计算资源以及对服务质量的需求. 通过限制并发任务的数量, 可用确保应用程序不会由于资源耗尽而失败, 或者由于在稀缺资源上发生竞争而严重影响性能, 通过将任务的提交任务的执行策略分离开来, 有助于在部署阶段选择与可用硬件资源最匹配的执行策略.

线程池是指管理一组同构工作线程的资源池. 线程池与工作队列是密切相关的, 其中在工作队列中保存了所有等待执行的任务. 工作者线程的任务很简单: 从工作队列中获取一个任务, 执行任务, 然后返回线程池并等待下一个任务.

在线程池中执行任务比为每个任务分配一个线程优势更多. 通过重用现有的线程而不是创建新线程, 可以在处理多个请求时分摊在线程创建和销毁过程中产生的巨大开销. 另一个额外的好处是, 当请求到达时, 工作线程通常已经存在, 因此不会由于等待创建线程而延迟任务的执行, 从而提高了响应性. 通过适当调整线程池的大小, 可以创建足够多的线程以便使处理器保持忙碌状态, 同时还可以防止过多线程相互竞争资源而使应用程序耗尽内存或失败.

使用Executors中的静态工厂方法来创建线程池:

1. newFixedThreadPool: 创建一个固定长度的线程池. 每当提交一个任务时就创建一个线程, 直到达到线程池的最大数量.
2. newCachedThreadPool: 创建一个可缓存的线程池. 如果线程池的规模超过了处理需求时, 那么将回收空闲的线程, 而当需求增加时, 则可以添加新的线程, 线程池的规模不存在任何限制.
3. newSingleThreadExecutor: 是一个单线程的Executor. 创建单个工作者线程来执行任务, 如果这个线程异常结束, 会创建另一个线程来代替. 该Executor能确保依照任务在队列中的顺序来串行执行.
4. newScheduledThreadPool: 创建了一个固定长度的线程池, 而且以延迟或定时的方式来执行任务.

JVM只有在所有非守护线程全部终止或才会退出, 因此如果无法正确地关闭Executor, 那么JVM将无法结束.

为了解决执行服务的生命周期问题, Executor扩展了ExecutorService接口, 添加了一些关于生命周期管理的方法. ExecutorService的生命周期有3种状态: 运行, 关闭和终止. ExecutorService在初始创建时处于运行状态. shutdown方法将执行平缓的关闭过程: 不再接受新的任务, 同时等待已经提交的任务执行完成(包括那些还未开始执行的任务). shutdownNow方法将执行粗暴的关闭过程: 它将尝试取消所有运行中的任务, 并且不再启动队列中尚未开始执行的任务.

在ExecutorService关闭后提交的任务将由拒绝执行处理器来处理, 它会抛弃任务, 或者使得execute方法抛出一个未检查的RejectedExecutionException异常. 等所有任务都完成后, ExecutorService进入终止状态. 可以调用awaitTermination来等待ExecutorService到达终止状态, 或者通过调用isTerminated来轮询ExecutorService是否已经终止. 通常在awaitTermination之后会立即调用shutdown, 从而产生同步地关闭ExecutorService的效果.

Timer类负责管理延迟任务以及周期任务, 但是其存在一些缺陷, 因此应该考虑使用ScheduledThreadPoolExecutor来代替它.

Timer在执行所有定时任务时只会创建一个线程, 如果某个任务的执行时间过长, 那么将破坏其他TimerTask的定时精确性. 线程池能弥补这个缺陷, 可以提供多个线程来执行延时任务和周期任务.

Timer的另一个问题是, 如果TimerTask抛出了一个未检查异常, 那么Timer线程并不捕获异常却终止定时线程. 这种情况下Timer也不会恢复线程的执行, 而是会错误地任务整个Timer都被取消了. 因此已经被调度但尚未执行地TimerTask将不会再执行, 新的任务也不能调度.

如果要构建自己的调度服务, 那么可以使用DelayQueue, 它实现了BlockingQueue, 并为ScheduledThreadPoolExecutor提供调度功能. DelayQueue管理着一组Delayed对象, 每个Delayed对象都有一个相应的延迟时间, 在DelayQueue中, 只有某个元素逾期后, 才能从DelayQueue中执行take操作. 从DelayQueue中返回的对象将根据它们的延迟时间进行排序.

找出可利用的平行性

Executor框架使用Runnable作为其基本的任务的表示形式. Runnable是一种有很大局限的抽象, 它不能返回一个值或抛出一个受检查的异常. 而Callable是一种更好的抽象, 其认为主入口点将返回一个值, 并可能抛出一个异常.

由于有些任务可能要执行很长时间, 因此通常希望能够取消这些任务. 在Executor框架中, 已提交但尚未开始的任务可以取消, 但对于那些已经开始执行的任务, 只有当它们能响应中断时才能取消. 取消一个已经完成的任务不会有任何影响.

Future表示一个任务的生命周期, 并提供了相应的方法来判断是否已经完成或取消, 以及获取任务的结果或取消任务等. 在Future规范中包含的隐含意义是任务的生命周期只能前进, 不能后退. ExecutorService中的所有submit方法都将返回一个Future.

Future的get方法的行为取决于任务的状态. 如果任务已经完成, 那么get会立即返回或抛出一个异常. 如果任务没有完成, 那么get将阻塞并直到任务完成. 如果任务抛出了异常, 那么get将该异常封装为ExecutionException并重新抛出, 可通过getCause来获得被封装的初始异常. 如果任务被取消, 那么get将抛出CancellationException.

在将Runnable或Callable提交到Executor的过程中, 包含了一个安全发布过程, 即将Runnable或Callable从提交线程发布到最终执行任务的线程. 类似地, 在设置Future结果的过程中也包含了一个安全发布, 即将这个结果从计算它的线程发布到任何通过get获得它的线程.

只有大量相互独立且同构的惹我你可以并发进行处理时, 才能体现出将程序的工作负载分配到多个任务中带来的真正性能提升. 因为如果一个任务的速度远远高于另一个任务的速度, 那么程序的最终性能与串行执行时的性能差别不大, 而代码却变得更复杂了. 虽然做了很多工作来并发执行异构任务以提高并发度, 但从中获得的并发性却是十分有限的.

CompletionService将Executor和BlockingQueue的功能融合在一起, 可以将Callable任务提交给它来执行, 然后使用类似于队列操作的take和poll等方法来获得已完成的结果, 而这些结果会在完成时被封装为Future. ExecutorCompletionService实现了CompletionService, 并将计算部分委托给一个Executor.

多个ExecutorCompletionService可以共享一个Executor, 因此可以创建一个对于特定计算私有, 又能共享一个公共Executor的ExecutorCompletionService. 因此CompletionService的作用相当于一组计算的句柄, 这与Future作为单个计算的句柄是非常类似的. 通过记录提交给CompletionService的任务数量, 并计算出已经获得的已完成结果的数量, 即使使用一个共享的Executor, 也能知道已经获得了所有任务结果的时间.

有时候, 如果某个任务无法在指定的时间内完成, 那么将不再需要它的结果, 此时可以放弃这个任务. 在有限的时间内执行任务的主要困难在于: 要确保得到答案的时间不会超过限定的时间, 或者在限定的时间内无法获得答案. 在支持时间限制的Future.get中支持这种需求: 当结果可用时, 它将立即返回, 如果在指定时限内没有计算出结果, 那么将抛出TimeoutException.

在使用限时任务时需要注意, 当这些任务超时后应该立即停止, 从而避免为继续计算一个不再使用的结果而浪费计算资源, 要实现这个功能, 可使用Future. 如果一个限时的get方法抛出了TimeoutException, 那么可以通过Future来取消任务.

如果任务相互独立且同构, 可以创建N个任务, 将其提交到一个线程池, 保留N个Future, 并使用限时的get方法通过Future串行地获取每一个结果, 此时可以使用invokeAll方法.

invokeAll方法地参数为一组任务, 并返回一组Future. invokeAll按照任务集合中迭代器的顺序将所有的Future添加到返回的集合中, 从而使调用者能将各个Future与其表示的Callable关联起来. 当所有任务都执行完毕时, 或者调用线程被中断时, 又或者超过指定时限时, invokeAll将返回. 当超过指定时限后, 任何还未完成的任务都会取消. 当invokeAll返回时, 每个任务要么正常地完成, 要么被取消, 而客户端代码可以调用get或isCancelled来判断究竟是何种情况.

取消与关闭

有时候需要提前结束任务或线程, 或许是因为用户取消了操作, 或者应用程序需要被快速关闭. Java提供了中断, 这是一种协作机制, 能够使一个线程终止另一个线程的当前工作.

这种协作式的方法是必要的, 很少希望某个任务, 线程或服务立即停止, 因为这种立即停止会使共享的数据结构处于不一致的状态. 使用协作方式时, 当需要停止时, 首先会清除当前正在执行的工作, 然后再结束. 这提供了更好的灵活性, 因为任务本身的代码比发出取消请求的代码更清楚如何执行清除工作.

任务取消

如果外部代码能在某个操作正常完成之前将其置入完成状态, 那么这个操作就可以称为可取消的.

Java采用协作机制, 其中一种协作机制能设置某个已请求取消标志, 而任务将定期地查看该标志, 如果设置了这个标志, 那么任务将提前结束. 如果是使用变量作为标志, 那么该变量必须是volatile类型的.

一个可取消的任务必须拥有取消策略, 在这个策略中详细定义取消操作的How, When和What. 即其他代码如何How请求取消该惹我你, 任务在何时When检查是否已经请求了取消, 以及在响应取消请求时应该执行哪些What操作.

如果任务中调用了一个阻塞方法, 那么设置请求取消标志的做法会产生一个严重的问题, 任务可能会永远不会检查取消标志, 因此永远不会结束.

线程中断是一种协作机制, 线程可以通过这种机制来通知另一个线程, 告诉它在合适的或者可能的情况下停止当前工作, 并转而执行其他的工作. 每个线程都有一个boolean类型的中断状态. 当中断线程时, 这个线程的中断状态将被设置为true. 在Thread中包含了中断线程以及查询中断线程状态的方法. 实例interrupt方法中断目标线程, 实例isInterrupted方法返回目标线程的中断状态, 静态的interrupted方法将清除当前线程的中断状态, 并返回之前的值.

Java库中的阻塞方法都会检查线程何时中断, 并且在发现中断时提前返回. 它们在响应中断时执行的操作包括: 清除中断状态, 抛出InterruptedException, 表示阻塞操作由于中断而提前结束. JVM并不能保证阻塞方法检测到中断的速度, 但在实际情况中响应速度还是非常快的.

当线程在非阻塞状态下中断时, 它的中断状态将被设置, 然后根据将被取消的操作来检查中断状态以判断发生了中断. 调用interrupt并不意味着立即停止目标线程正在进行的工作, 而只是传递了请求中断的消息.

对中断操作的正确理解是: 它并不会真正地中断一个正在运行的线程, 而只是发出中断请求, 然后由线程在下一个合适的时刻中断自己. 设计良好的方法需要使调用代码对中断请求进行处理, 设计糟糕的方法可能会屏蔽中断请求, 从而导致调用栈中的其他代码无法对中断请求做出响应.

在使用静态的interrupted方法时应该小心, 因为它会清除当前线程的中断状态. 如果在调用interrupted时返回了true, 即线程之前被中断, 那么除非想屏蔽这个中断, 否则必须对它进行处理. 可以抛出InterruptedException, 或者通过再次调用interrupt来恢复中断状态. 通常中断是实现取消的最合理方式.

线程同样应该包含中断策略. 中断策略规定线程如何解释某个中断请求, 即当线程发现中断请求时, 应该做哪些工作, 哪些工作单元对于中断来说是原子操作, 以及以多快的速度来响应中断.

最合理的中断策略是某种形式的线程级取消操作或服务级取消操作: 尽快退出, 在必要时进行清理, 通知某个所有者该线程已退出. 一个中断请求可以有一个或多个接收者, 如中断线程池中的某个工作者线程, 同时意味着取消当前任务和关闭工作者线程.

有些任务会在线程池中拥有的线程中执行, 对于非线程所有者的代码应该小心地保存中断状态, 这样拥有线程的代码才能对中断做出响应. 这就是大多数可阻塞的库函数只是抛出InterruptedException作为中断响应的原因, 因为它们为任务或库代码实现了最合理的取消策略: 尽快退出执行流程, 并把中断信息传递给调用者, 从而使调用栈中的上层代码可以采取进一步的操作.

当检查到中断请求时, 任务并不需要放弃所有的操作, 它可以推迟处理中断请求, 并直到某个更合适的时机. 因此需要记住中断请求, 并在完成当前任务后抛出InterruptedException或者表示已收到中断请求. 这项技术能够确保在更新过程中发生中断时, 数据结构不会被破坏. 如果除了将InterruptedException传递给调用者外还需要执行其他操作, 那么应该在捕获InterruptedException之后恢复中断状态.

通过推迟中断请求的处理, 开发人员能制定更灵活的中断策略, 从而使应用程序在响应性和健壮性之间实现合理的平衡.

任务不应该对执行该任务的线程的中断策略做出任何假设, 同样执行取消操作的代码也不一般应该对线程的中断策略做出假设. 线程应该只能由其所有者中断, 所有者可以将线程的中断策略信息封装到某个合适的取消机制中.

只有实现了线程中断策略的代码才可以屏蔽中断请求, 在常规的任务或库代码中都不应该屏蔽中断请求.

对一些不支持取消但仍可以调用可中断阻塞方法的操作, 它们必须在循环中调用这些方法, 并在发现中断后重新尝试. 在这种情况下, 它们应该在本地保存中断状态, 并在返回前恢复状态而不是在捕获InterruptedException时恢复状态. 如下面的代码:

public Task getNextTask(BlockingQueue<Task> queue) {
    boolean interrupted = false;
    try {
        while (true) {
        try {
            return queue.take();
        } catch (InterruptedException e) {
            interrupted = true;
            // 下次循环进行重试
        }
    	}
    } finally {
        if (interrupted) {
            Thread.currentThread().interrupt();
        }
    }
}

如果过早地设置中断状态, 就可能引起无限循环, 因为大多数可中断的阻塞方法都会在入口处检查中断状态, 并且发现该状态已被设置时会立即抛出InterruptedException.

如果代码中不会调用可中断的阻塞方法, 那么仍然可以通过在任务代码中轮询当前线程的中断状态来响应中断. 要选择合适的轮询频率, 就需要在效率和响应性之间进行权衡.

可以使用Future来实现取消, Future有一个cancel方法, 该方法带有一个boolean类型的参数mayInterruptRunning, 表示取消是否成功. 如果mayInterruptRunning为true并且任务当前正在某个线程中运行, 那么这个线程能被中断. 如果这个参数为false, 那么意味着任务还没有启动, 就不要运行它, 这种方式应该用于那些不处理中断的任务.

如果执行任务的线程在标准的Executor创建的, 并实现了一种中断策略使得任务可以通过中断被取消, 那么可以通过Future来取消任务, 此时可以设置mayInterruptRunning为true. 当尝试取消任务时, 不宜直接中断线程池, 因为并不知道当中断请求到达时正在运行什么任务, 只能通过任务的Future来实现取消. 当Future.get抛出InterruptedException或TimeoutException时, 如果知道不再需要结果, 那么就可以调用Future.cancel来取消任务.

对于执行不可中断操作而被阻塞的线程, 只能通过设置线程的中断状态来停止这些线程, 但要求必须知道线程阻塞的原因.

1. Java.io包中的同步Socket I/O: 通过关闭底层的套接字, 可以使得由于执行read或write等方法而被阻塞的线程抛出一个SocketException.
2. Java.io包中的同步I/O: 当中断一个正在InterruptibleChannel上等待的线程时, 将抛出ClosedByInterruptedException并关闭链路, 还会使得其他在这条链路上阻塞的线程同样抛出该异常. 当关闭一个InterruptibleChannel时, 将导致所有在链路操作上阻塞的线程都抛出AsynchronousCloseException.
3. Selector的异步I/O: 如果一个线程在调用Selector.select方法时阻塞了, 那么调用close或wakeup方法会使线程抛出ClosedSelectorException并提前返回.
4. 获取某个锁: 如果一个线程由于等待某个内置锁而阻塞, 那么将无法响应中断, 因为线程认为它肯定会获得锁, 所以就不会例会中断请求. 在Lock类中提供了lockInterruptibly方法, 该方法允许在等待一个锁的同时仍能响应中断.

下面的代码给出了如何封装非标准的取消操作:

public class ReaderThread extends Thread {
    private final Socket socket;
    private final InputStream in;

    public ReaderThread(Socket socket) throws IOException {
        this.socket = socket;
        this.in = socket.getInputStream();
    }

    @Override
    public void interrupt() {
        try {
            socket.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            super.interrupt();
        }
    }
}

ReaderThread重写了interrupt方法使其既能处理标准的中断, 也能关闭底层的套接字.

还可以通过ThreadPoolExecutor.newTaskFor方法来进一步优化ReaderThread中封装非标准取消的技术. 当把一个Callable提交给ExecutorService时, submit方法会返回一个Future, 可通过这个Future来取消任务. newTaskFor方法是一个工厂方法, 它将创建Future来代表任务. newTaskFor还能返回一个RunnableFuture接口, 该接口扩展了Future和Runnable, 并由FutureTask实现.

public interface CancellableTask<T> extends Callable<T> {
    void cancel();
    RunnableFuture<T> newTask();
}
public class CancellingExecutor extends ThreadPoolExecutor {
    public CancellingExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    @Override
    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        if (callable instanceof CancellableTask) {
            return ((CancellableTask<T>) callable).newTask();
        } else {
            return super.newTaskFor(callable);
        }
    }
}
public abstract class SocketUsingTask<T> implements CancellableTask<T> {
    private Socket socket;

    protected synchronized void setSocket(Socket s) {
        this.socket = s;
    }

    @Override
    public synchronized void cancel() {
        try {
            if (socket != null) {
                socket.close();
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public RunnableFuture<T> newTask() {
        return new FutureTask<T>(this) {
            @Override
            public boolean cancel(boolean mayInterruptIfRunning) {
                try {
                    SocketUsingTask.this.cancel();
                } finally {
                    return super.cancel(mayInterruptIfRunning);
                }
            }
        };
    }
}

SocketUsingTask实现了CancellableTask, 并定义了Future.cancel来关闭套接字和调用super.cancel. 如果SocketUsingTask通过自己的Future来取消, 那么底层的套接字将被关闭并且线程将被中断. 因此它提高了任务对取消操作的响应性, 不仅能够在调用可中断方法的同时确保响应取消操作, 而且还能调用可阻塞的套接字I/O方法.

停止基于线程的服务

应用程序通常会创建多个线程的服务, 例如线程池, 并且这些服务的生命周期通常比创建它们的方法的生命周期更长. 如果应用程序准备退出, 那么这些服务所拥有的线程也需要结束, 由于无法通过抢占式的方法来停止线程, 因此这些服务需要自行结束.

正确的封装原则是: 除非拥有某个线程, 否则不能对该线程进行操控. 在线程API中, 并没有对线程所有权给出正式的定义, 线程由Thread对象表示, 并且像其他对象一样可以被自由地共享. 然而线程有一个相应的所有者, 即创建该线程的类. 因此线程池是其工作者线程的所有者, 如果要中断这些线程, 那么应该使用线程池.

与其他封装对象一样, 线程的所有权是不可传递的, 应用程序可以拥有服务, 服务可以拥有工作者线程, 但应用程序并不能拥有工作者线程, 因此应用程序不能直接停止工作者线程, 而服务应该提供生命周期方法来关闭自己以及它所拥有的线程. 对于持有线程的服务, 只要服务的存在时间大于创建线程方法的时间, 那么就应该提供生命周期方法.

ExecutorService提供了两种关闭方法, 使用shutdown正常关闭, 以及使用shutdownNow强行关闭. 在进行强行关闭时, shutdownNow首先关闭当前正在执行的任务, 然后返回所有尚未启动的任务清单.

这两种关闭方式的差别在于各自的安全性和响应性, 强行关闭的速度更快, 但风险更大, 因为任务很可能在执行到一半时被结束, 而正常关闭虽然速度慢, 但却更安全, 因为ExecutorService会一直等待队列中的所有任务都执行完毕后才关闭.

另一种关闭生产者-消费者服务的方式是使用毒丸对象. 毒丸是指一个放在队列上的对象, 含义是: 当得到这个对象, 立即停止. 在FIFO先进先出队列中, 毒丸对象将确保消费者在关闭之前首先完成队列中的所有工作, 在提交毒丸对象之前提交的所有工作都会被处理, 而生产者在提交了毒丸对象后, 将不会再提交任何工作.

只有在生产者和消费者的数量都已知的情况下, 才可以使用毒丸对象. 可以扩展到多个生产者和多个消费者. 然而当生产者和消费者的数量较大时, 这种方法将变得难以使用. 此时只有在无界队列中, 毒丸对象才可能可靠地工作.

如果某个方法A需要处理一批任务, 并且当所有任务都处理完成后才返回, 那么可以通过一个私有的Executor来简化服务的生命周期管理, 其中该Executor的生命周期是由这个方法A来控制的.

当通过shutdownNow来强行关闭ExecutorService时, 会尝试取消正在执行的任务, 并返回所有已提交但尚未开始的任务, 从而将这些任务写入日志或者保存起来以便之后进行处理. 但无法在关闭过程中知道正在执行的任务的状态, 除非任务本身会执行某种检查. 要知道哪些任务还没有完成, 不仅需要知道哪些任务还没有开始, 还需要知道当Executor关闭时哪些任务正在执行.

下面的代码给出了如何在关闭过程中判断正在执行的任务的方法:

public abstract class TrackingExecutor extends AbstractExecutorService {

    private final ExecutorService exec;

    private final Set<Runnable> tasksCancelledAtShutdown = Collections.synchronizedSet(new HashSet<>());

    public TrackingExecutor(ExecutorService exec) {
        this.exec = exec;
    }

    public List<Runnable> getCancelledTasks() {
        if (!exec.isTerminated()) {
            throw new IllegalStateException();
        }
        return new ArrayList<>(tasksCancelledAtShutdown);
    }
    @Override
    public void execute(Runnable command) {
        exec.execute(() -> {
            try {
                command.run();
            } finally {
                if (isShutdown() && Thread.currentThread().isInterrupted()) {
                    tasksCancelledAtShutdown.add(command);
                }
            }
        });
    }
}

在TrackingExecutor中存在一个不可避免的竞态条件, 从而产生误报问题: 一些被认为已取消的认为实际上已经执行完成. 这个问题的原因在于, 在任务执行最后一条指令以及线程池及将任务记录为结束的两个时刻之间, 线程池可能被关闭. 如果任务是幂等的, 那么就不会存在问题.

处理非正常的线程终止

如果并发程序中的某个线程发生故障, 那么通常不会导致程序停止运行并输出栈追踪信息到控制台上. 此外当线程发生故障时, 应用程序可能看起来仍然在工作, 所以这个故障可能会被忽略.

导致线程提前死亡的最主要原因是RuntimeException. 由于这些异常表示除了某种编程错误或者其他不可修复的错误, 因此它们通常不会被捕获, 它们不会再调用栈中逐层转递, 而是默认地再控制台中输出栈追踪信息, 并终止线程.

任何代码都可能抛出一个RuntimeException, 每当调用另一个方法时, 都要对它的行为保持怀疑, 不要盲目地认为它一定会正常返回, 或者一定会抛出在方法原型中声明地某个已检查异常.

在线程池内部, 如果任务抛出了一个未检查异常, 那么它将使线程终结, 但会首先通知框架该线程已经终结. 然后框架可能会用新的线程来代替这个工作线程, 也可能不会, 因为线程池正在关闭, 或者当前已有足够多的线程能满足需要.

在Thread的API中提供了UncaughtExceptionHandler, 它能检测出某个线程由于未捕获的异常而终结的情况. 当一个线程由于未捕获异常而退出时, JVM会把这个事件报告给应用程序提供的UncaughtExceptionHandler异常处理器, 如果没有提供任何异常处理器, 那么默认的行为是将栈追踪信息输出到System.err.

异常处理器如何处理未捕获异常, 取决于对服务质量的需求. 最常见的响应方式是将一个错误信息以及相应的栈追踪信息写入应用程序日志中. 在运行时间较长的应用程序中, 通常会为所有线程的未捕获异常指定同一个异常处理器, 并且该处理器至少会将异常信息记录到日志中.

要为线程池中的所有线程设置一个UncaughtExceptionHandler, 需要为ThreadPoolExecutor的构造函数中提供一个ThreadFactory. 标准线程池允许当发生未捕获异常时结束线程, 但由于使用了一个try-finally代码块来接收通知, 因此当线程结束时, 将有新的线程来代替它. 如果没有提供捕获异常处理器或其他的故障通知机制, 那么任务会悄悄失败, 从而导致极大的混乱. 如果希望在任务由于发生异常而失败时获得通知, 并且执行一些特定于任务的恢复操作, 那么可以将任务封装在能捕获异常的Runnable或Callable中, 或者改写ThreadPoolExecutor的afterExecute方法.

只有通过execute提交的任务, 才能将它抛出的异常交给未捕获异常处理器, 而通过submit提交的任务, 无论是抛出的未检查异常还是已检查异常, 都将被任务是任务返回状态的一部分. 如果一个由submit提交的任务由于抛出了异常而结束, 那么这个异常将被Future.get封装在ExecutionException中重新抛出.

JVM关闭

JVM既可以正常关闭, 也可以强行关闭. 正常关闭包括: 最后一个正常非守护线程结束, 或者调用了System.exit, 或者通过其他特定于平台的方法关闭时(例如发送了sigint信息或键入ctrl-c). 强行关闭包括: 通过调用Runtime, halt或者在操作系统中杀死JVM进程.

在正常关闭中, JVM首先调用所有已注册的关闭钩子. 关闭钩子是指通过Runtime.addShutdownHook注册的但尚未开始的线程. JVM并不能保证关闭钩子的调用顺序. 在关闭应用程序线程时, 如果有守护或非守护线程仍然在运行, 那么这些线程接下来将与关闭进程并发执行. 当所有的关闭钩子都执行结束时, 如果runFinalizersOnExit为true, 那么JVM将运行终结器, 然后再停止. JVM并不会停止或中断任何在关闭时仍然运行的应用程序线程. 当JVM最终结束时, 这些线程将强行结束. 如果关闭钩子或终结器没有执行完成, 那么正常关闭进程将挂起. 当被强行关闭时, 只是关闭JVM, 而不会运行关闭钩子.

关闭钩子应该是线程安全的, 它们在访问共享数据时必须使用同步机制, 并且小心地避免发生死锁, 这与其他并发代码的要求相同. 而且关闭钩子不应该对应用程序的状态或者JVM的关闭原因做出任何假设, 因此在编写关闭钩子的代码时必须考虑周全. 关闭钩子必须尽快退出, 因为它们会延迟JVM的结束时间, 而用户可能希望JVM能尽快终止.

关闭钩子不应该依赖那些可能被应用程序或其他关闭钩子关闭的服务.可以对所有服务使用同一个关闭钩子, 并且在该关闭钩子中执行一系列的关闭操作, 这确保了关闭操作在单个线程中串行执行, 从而避免了关闭操作之间出现竞态条件或死锁等问题. 当应用程序需要维护多个服务之间的显式依赖信息时, 这项技术可以确保关闭操作按照正确的顺序执行.

有时候希望创建一个线程来执行一些辅助工作, 但又不希望这个线程阻碍JVM的关闭, 这种情况下可使用守护线程. 线程可分为两种: 普通线程和守护线程. 默认情况下, 主线程创建的所有线程都是主线程.

普通线程与守护线程之间的差异仅在于当线程退出操作时发生的操作. 当一个线程退出时, JVM会检查其他正在运行的线程, 如果这些线程都是守护线程, 那么JVM会正常退出操作. 当JVM停止时, 所有仍然存在的守护线程都将被抛弃.

应尽可能少地使用守护胡线程, 守护线程通常不能用来代替应用程序管理各个服务的生命周期.

Java垃圾回收器对那些定义了finalize方法的对象会进行特色处理: 在回收器释放它们后, 调用它们的finalize方法, 从而保证一些持久化的资源被释放.

由于终结器可以在某个JVM管理的线程中运行, 因此终结器访问的任何状态都可能被多个线程访问, 这样就必须对其访问操作进行同步. 终结器并不能保证它们将在何时运行甚至是否会运行, 并且复杂的终结器通常还会在对象上产生巨大的性能开销. 避免使用终结器. 大多数情况下通过使用finally代码块或显示的close方法, 能够比使用终结器更好地管理资源.

线程池的使用

本节介绍对线程池进行配置与调优的一些高级选项, 并分析在使用任务执行框架时需要注意的各种危险, 以及一些使用Executor的高级示例.

在任务与执行策略之间的隐性耦合

虽然Executor框架可以将任务的提交与任务的执行策略解耦开来, 为指定和修改执行策略都提供了相当大的灵活性, 但并非所有的任务都能适用所有的执行策略, 有些类型的任务需要明确地指定执行策略, 包括:

依赖性任务: 大多数行为正确的任务都是独立的, 不依赖于其他任务的执行时序, 执行结果或其他效果. 当在线程池中执行独立的任务时, 可以随意地改变线程池地大小和配置, 这些修改只会对执行性能产生影响. 然而, 如果提交给线程池地任务需要依赖其他任务, 那么就隐含地给执行策略带来了约束, 此时必须小心地维持这些执行策略以避免产生活跃性问题.

使用线程封闭机制的任务: 对象可以封闭在任务线程中, 使得该线程中执行的任务在访问对象时不需要同步, 这种情形将任务与执行策略之间形成隐式的耦合: 任务要求其执行所在的Executor是单线程的. 如果将Executor从单线程环境改为线程池环境, 那么将会失去线程安全性.

对响应时间敏感的任务: 如果将一个运行时间较长的任务提交到单线程的Executor中, 或者将多个运行时间较长的任务提交到一个只包含少量线程的线程池中, 那么将降低由该Executor管理的服务的响应性.

使用ThreadLocal的任务: 只有当线程本地值的生命周期受限于任务的生命周期时, 在线程池中的线程中使用ThreadLocal才有意义, 而在线程池中不应该使用ThreadLocal在任务之间传递值.

只有当任务都是同类型的并且相互独立时, 线程池的性能才能达到最佳. 如果将运行时间较长的与运行时间较短的任务混合在一起, 那么除非线程池特别大, 否则将可能造成拥塞. 如果提交的任务依赖于其他任务, 那么除非线程池无限大, 否则将可能造成死锁.

如果线程池中的任务需要无限期地等待一些必须由池中的其他任务才能提供的资源或条件, 那么除非线程池足够大, 否则将发生线程饥饿死锁.

如果任务阻塞的时间过长, 那么即使不出现死锁, 线程池的响应性也会变得糟糕. 执行时间较长的任务不仅会造成线程池堵塞, 甚至还会增加执行时间较短任务的服务时间. 如果线程池中的线程数量远小于在稳定状态下执行时间较长任务的数量, 那么最后可能所有的线程都会运行这些执行时间较长的任务, 从而影响整体的响应性. 可以通过限定任务等待资源时间的技术来缓解执行时间较长任务造成的影响.

设置线程池的大小

线程池的理想大小取决于被提交任务的类型以及所部署系统的特性. 在代码中通常不会固定线程池的大小, 而应该通过某种配置机制来提供, 或者根据Runtime.availableProcessors来动态计算.

如果线程池过大, 那么大量的线程将在相对很少的CPU和内存资源上发生竞争, 这不仅会导致更高的内存使用量, 而且还可能耗尽资源. 如果线程池过小, 那么将导致许多空闲的处理器无法执行工作, 从而降低吞吐率.

对于计算密集型的任务, 在拥有N个处理器的系统上, 当线程池的大小为N+1时, 通常能实现最优的利用率. 对于包含I/O操作或者其他阻塞操作的任务, 由于线程并不会一直执行, 因此线程池的规模应该更大. 要正确地设置线程池的大小, 必须估算任务的等待时间与计算时间的比值.

当然CPU周期并不是唯一影响线程池大小的资源, 还包括内存, 文件句柄, 套接字句柄和数据库连接等. 计算这些资源对线程池的约束条件是更容易的: 计算每个任务对该资源的需求量, 然后用该资源的可用总量除以每个任务的需求量, 所得结果就是线程池的大小的上限.

当任务需要某种通过资源池来管理的资源是, 例如数据库连接, 那么线程池和资源池的大小将会相互影响.

配置ThreadPoolExecutor

如果默认的执行策略不能满足需求, 那么可以通过ThreadPoolExecutor的构造函数来实例化一个对象, 并根据自己的需求来定制.

线程池的基本大小(CorePoolSize): 是线程池的目标大小, 即在没有任务执行时线程池的大小, 并且只有在工作队列满了的情况下才会创建出超出这个数量的线程.

线程池的最大大小(MaximumPoolSize): 表示可同时活动的线程数量上限, 如果某个线程的空闲时间超过了存活时间, 那么将被标记为可回收的, 并且当线程池的当前大小超过了基本大小时, 这个线程将被终止.

调节线程池的基本大小和存活时间, 可以帮助线程池回收空闲线程占有的资源, 从而使得这些资源可以用于执行其他工作. newFixedThreadPool工厂方法将线程池的基本大小和最大大小设置为参数中指定的值, 而且创建线程池不会超时. newCachedThreadPool工厂方法将线程池的最大大小设置为Integer.MAX_VALUE, 而将基本大小设置为零, 并将超时设置为1分钟, 这种方法创建出来的线程池可被无限扩展, 并且当需求降低时会自动收缩.

如果无限制地创建线程, 那么将导致不稳定, 可通过采用固定大小的线程池来解决这个问题. 然而这个方案并不完整, 在高负载情况下, 应用程序仍可能耗尽资源, 只是出现问题的概率较小. 如果新请求的到达速率超过了线程池的处理速度, 那么新到来的请求将累积起来, 这些请求任务在线程池的队列中等待, 而不会像线程那样去竞争CPU资源.

即使请求的平均到达速率很稳定, 也仍然会出现请求突增的情况, 尽快队列有助于缓解任务的突增问题, 但如果任务持续高速地到来, 那么最终还是会抑制请求的到达率以避免耗尽内存. 甚至在耗尽内存之前, 响应性能也将随着任务队列的增长而变得越来越糟.

newFixedThreadPool和newSingleThreadExecutor在默认情况下使用一个无界队列, 如果任务持续快速地到达, 并且超过了线程池处理地速度, 那么队列将无限制地增加, 耗尽内存.

一种更稳妥的资源管理策略是使用有界队列, 有界队列有助于避免资源耗尽的情况发生, 当队列填满后, 新的任务通过饱和策略进行处理. 在使用有界队列时, 队列的大小与线程池的大小必须一起调节, 如果线程池较小而队列较大, 那么有助于减少内存使用量, 降低CPU的使用率., 同时还可以减少上下文切换, 但付出的代价是可能会限制吞吐量.

对于非常大的或者无界的线程池, 可以通过使用SynchronousQueue来避免任务排队, 以及直接将任务从生产者移交给工作者线程. SynchronousQueue不是一个真正的队列, 而是一种在线程之间进行移交的机制. 要将一个元素放入SynchronousQueue中, 必须有另一个线程正在等待接受这个元素, 如果没有线程正在等待, 并且线程池的当前大小小于最大值, 那么线程池将创建一个新的线程, 否则根据饱和策略, 这个任务将被拒绝. 使用直接移交将更高效, 因为任务会直接移交给执行它的线程, 而不是被首先放在队列中, 然后由工作者线程从队列中提前该任务, 只有当线程池是无界的或者可以拒绝任务时, SynchronousQueue才有实际价值. 在newCachedThreadPool工厂方法中就使用了SynchronousQueue.

如果想控制任务执行顺序, 可以使用PriorityBlockingQueue, 这个队列根据优先级来安排任务. 任务的优先级是通过自然顺序或Comparator来定义的.

当有界队列被填满后, 饱和策略开始发挥作用. ThreadPoolExecutor的饱和策略可以通过调用setRejectedExecutionHandler来修改. 如果一个任务被提交到一个已被关闭的Executor时也会用到饱和策略. JDK提供了几种不同的RejectedExecutionHandler实现:

中止Abort策略是默认的饱和策略, 该策略将抛出未检查的RejectedExecutionException, 调用者可以捕获这个异常, 然后根据需求编写自己的处理代码. 当新提交的任务无法保存到队列中等待执行时, 抛弃Discard策略会悄悄抛弃该任务. 抛弃最旧的策略则会抛弃下一个将被执行的任务, 然后尝试重新提交新的任务. 如果工作队列是一个优先级队列, 那么抛弃最旧的策略将导致抛弃优先级最高的任务, 因此最好不要将抛弃最旧的饱和策略和优先级队列放在一起使用.

调用者运行策略实现了一种调节机制, 该策略既不会抛弃任务, 也不会抛出异常, 而是将某些任务回退到调用者, 从而降低任务的流量. 它不会在线程池的某个线程中执行新提交的任务, 而是在一个调用了execute的线程中执行该任务.

当工作队列被填满后, 没有预定义的饱和策略来阻塞execute, 可以通过使用Semaphore信号量来限制任务的到达率. 如下面的代码:

public class BoundedExecutor {
    private final Executor exec;

    private final Semaphore semaphore;

    public BoundedExecutor(Executor exec, int bound) {
        this.exec = exec;
        this.semaphore = new Semaphore(bound);
    }

    public void submitTask(final Runnable command) throws InterruptedException {
        semaphore.acquire();
        try {
            exec.execute(() -> {
                try {
                    command.run();
                } finally {
                    semaphore.release();
                }
            });
        } catch (RejectedExecutionException e) {
            semaphore.release();
        }
    }
}

该方法使用了一个无界队列, 因为不能限制队列的大小和任务的到达率, 并设置信号量的上界设置为线程池的大小加上可排队任务的数量, 这是因为信号量需要控制正在执行的和等待执行的任务数量.

每当线程池需要创建一个线程时, 都是通过线程工厂方法来完成的. 默认的线程工厂方法将创建一个新的, 非守护的线程, 并且不包含特殊的配置信息. 通过指定一个线程工厂方法, 可以定制线程池的配置信息. 在ThreadFactory中只定义了一个方法newThread, 每当线程池需要创建一个新线程时都会调用这个方法.

在调用完ThreadPoolExecutor的构造函数后, 仍然可以通过setter方法来修改大多数传递给它的构造函数的参数. 如果Executor是通过Executors中的工厂方法创建的, 那么可以将结果的类型转换为ThreadPoolExecutor以访问设置器.

在Executors中包含一个unconfigurableExecutorService工厂方法, 该方法对一个现有的ExecutorService进行包装, 使其只暴露出ExecutorService的方法, 因此不能对它进行配置. 可以在自己的Executor中使用这项技术来防止执行策略被修改.

扩展ThreadPoolExecutor

ThreadPoolExecutor是可扩展的, 提供了几个可以在子类化中改写的方法. 在执行任务的线程中将调用beforeExecute和afterExecute等方法, 在这些方法中还可以添加日志, 计时, 监视或统计信息收集的功能. 无论任务是从run中正常返回, 还是抛出一个异常返回, afterExecute都会被调用. 如果任务在完成后带有一个Error, 那么就不会调用afterExecute. 如果beforeExecute抛出一个RuntimeException, 那么任务将不被执行, 并且afterExecute也不会被调用.

在线程池完成关闭操作时调用terminated, 也就是在所有任务都已经完成并且所有工作者线程也已经关闭后. terminated可以用来释放Executor在其生命周期里分配的各种资源, 此外还可以执行发生通知, 记录日志等操作.

递归算法的并行化

如果在循环体中包含了一些密集计算, 或者需要执行可能阻塞的I/O操作, 那么只要每次迭代是独立的, 都可以对其进行并行化. 如果循环中的迭代操作都是独立的, 并且不需要等待所有的迭代操作都完成再进行执行, 那么就可以使用Executor将串行循环转化为并行循环.

void processInParallel(Executor exec, List<Element> elements) {
    // 使用execute提交任务后会立即返回
    // 如果需要提交一个任务集并等待它们完成, 那么可以使用invokeAll, 并且在所有任务都执行完成后调用CompletionService来获取结果
    for (final Element e: elements) {
        exec.execute(() -> process(e));
    }
}

当串行循环中的各个迭代操作直接彼此独立, 并且每个迭代操作执行的工作量比管理一个新任务时带来的开销更多, 那么这个串行循环就适合并行化.

在一些递归设计中同样可以才有循环并行化的方法. 一种简单的情况是在每个迭代操作中都不需要来自后续递归迭代的结果.

这项技术的一种强大应用就是解决一些谜题, 这些谜题都需要找出一系列的操作从初始状态转换到目标状态. 下面将谜题定义为: 包含了一个初始位置, 一个目标位置, 以及用于判断是否是有效移动的规则集. 规则集包含两部分: 计算从指定位置开始的所有合法移动, 以及每次移动的结果位置.

/
 * 谜题的抽象接口
 *
 * @param <P> 位置类
 * @param <M> 移动类
 */
public interface Puzzle<P, M> {
    // 初始位置
    P initialPosition();
    // 判断是否是目标
    boolean isGoal(P position);
    // 合法的移动集合
    Set<M> legalMoves(P position);
    // 进行移动
    P move(P position, M move);

    // 节点类
    class Node<P, M> {
        final P pos;
        final M move;
        final Node<P, M> prev;

        public Node(P pos, M move, Node<P, M> prev) {
            this.pos = pos;
            this.move = move;
            this.prev = prev;
        }
        // 一定轨迹列表
        List<M> asMoveList() {
            List<M> solution = new LinkedList<>();
            for (Node<P, M> n = this; n.move != null; n = n.prev) {
                solution.add(0, n.move);
            }
            return solution;
        }
    }
}
/
 * 串行的谜题解决方案
 *
 * @param <P> 位置类
 * @param <M> 移动类
 */
public class SequentialPuzzleSolver<P, M> {
    private final Puzzle<P, M> puzzle;

    private final Set<P> seen = new HashSet<>();

    public SequentialPuzzleSolver(Puzzle<P, M> puzzle) {
        this.puzzle = puzzle;
    }

    public List<M> solve() {
        P pos = puzzle.initialPosition();
        return this.search(new Puzzle.Node<>(pos, null, null));
    }
    // 搜寻目标,并返回移动轨迹
    private List<M> search(Puzzle.Node<P, M> node) {
        if (!seen.contains(node.pos)) {
            seen.add(node.pos);
            if (puzzle.isGoal(node.pos)) {
                return node.asMoveList();
            }
            for (M move : puzzle.legalMoves(node.pos)) {
                P pos = puzzle.move(node.pos, move);
                Puzzle.Node<P, M> child = new Puzzle.Node<>(pos, move, node);
                List<M> result = this.search(child);
                if (result != null) {
                    return result;
                }
            }
        }
        return null;
    }
}

/
 * 串行的谜题解决方案
 *
 * @param <P> 位置类
 * @param <M> 移动类
 */
public class ConcurrentPuzzleSolver<P, M> {
    private final Puzzle<P, M> puzzle;

    private final ExecutorService exec;

    private final ConcurrentMap<P, Boolean> seen;

    final ValueLatch<Puzzle.Node<P, M>> solution = new ValueLatch<>();

    public ConcurrentPuzzleSolver(Puzzle<P, M> puzzle) {
        this.puzzle = puzzle;
        this.exec = Executors.newCachedThreadPool();
        this.seen = new ConcurrentHashMap<>();
    }

    public List<M> solve() throws InterruptedException {
        try {
            P pos = puzzle.initialPosition();
            exec.execute(newTask(pos, null, null));
            Puzzle.Node<P, M> solutionNode = solution.getValue();
            return solutionNode == null ? null : solutionNode.asMoveList();
        } finally {
            exec.shutdown();
        }
    }

    protected Runnable newTask(P pos, M move, Puzzle.Node<P, M> node) {
        return new SolverTask(pos, move, node);
    }

    class SolverTask extends Puzzle.Node<P, M> implements Runnable {
        public SolverTask(P pos, M move, Puzzle.Node<P, M> prev) {
            super(pos, move, prev);
        }

        @Override
        public void run() {
            // 已经找到答案或者已经遍历了这个位置
            if (solution.isSet() || seen.putIfAbsent(pos, true) != null) {
                return;
            }
            // 找到目标添加到solution中
            if (puzzle.isGoal(pos)) {
                solution.setValue(this);
            } else {
                // 否则继续执行
                for (M move : puzzle.legalMoves(pos)) {
                    exec.execute(newTask(puzzle.move(pos, move), move, this));
                }
            }
        }
    }
}

public class ValueLatch<T> {

    private T value = null;

    private final CountDownLatch done = new CountDownLatch(1);

    public boolean isSet() {
        return done.getCount() == 0;
    }

    public synchronized void setValue(T newValue) {
        if (!isSet()) {
            value = newValue;
            done.countDown();
        }
    }

    public T getValue() throws InterruptedException {
        done.await();
        synchronized (this) {
            return value;
        }
    }
}

public class PuzzleCountingSolver<P, M> extends ConcurrentPuzzleSolver<P, M> {
    private final AtomicInteger taskCount = new AtomicInteger(0);
    
    public PuzzleCountingSolver(Puzzle<P, M> puzzle) {
        super(puzzle);
    }

    @Override
    protected Runnable newTask(P pos, M move, Puzzle.Node<P, M> node) {
        return new CountingSolverTask(pos, move, node);
    }

    class CountingSolverTask extends SolverTask {

        public CountingSolverTask(P pos, M move, Puzzle.Node<P, M> prev) {
            super(pos, move, prev);
            taskCount.incrementAndGet();
        }

        @Override
        public void run() {
            try {
                super.run();
            } finally {
                if (taskCount.decrementAndGet() == 0) {
                    solution.setValue(null);
                }
            }
        }
    }
}

public void transferMoney(Account fromAccount, Account toAccount, DollarAmount amount) throw InsufficientFundsException {
    synchronized(fromAccount) {
        synchronized(toAccount) {
            if (fromAccount.getBalance().compareTo(amount) < 0) {
                throw new InsufficientFundsException();
            } else {
                fromAccount.debit(amount);
                toAccount.credit(amount);
            }
        }
    }
}

public String getStoogeNames() {
    List<String> stooges = new Vector<>();
    stooges.add("Moe");
    stooges.add("Larry");
    stooges.add("Curly");
    return stooges.toString();
}

基于对象状态请求锁
while (前置条件没有满足) {
    释放锁
    等待直到前置条件满足
    (可选的)如果被打断或超时
    重新请求锁
}
执行操作
释放锁

void stateDependentMethod() throws InterruptedException {
    // 必须通过一个锁来保护条件谓词
    synchronized (lock) {
        while (!conditionPredicate()) {
            lock.wait();
        }
        // 现在对象处于合适的状态
    }
}

public class ThreadGate {

    private boolean isOpen;
    private int generation;

    public synchronized void close() {
        isOpen = false;
    }

    public synchronized void open() {
        ++generation;
        isOpen = true;
        notifyAll();
    }

    public synchronized void await() throws InterruptedException {
        int arrivalGeneration = generation;
        while (!isOpen && arrivalGeneration == generation) {
            wait();
        }
    }
}

boolean acquire() throws InterruptedException {
    while (当前状态不允许获取操作) {
        if (需要阻塞获取操作) {
            如果当前线程不在队列中,则将其插入队列
            阻塞当前线程
        } else {
            返回失败
        }
    }
    可能更新同步器的状态
    如果线程位于队列中,则将其移出队列
    返回成功
}

void release() {
    更新同步器的状态
    if (新的状态允许某个被阻塞的线程获取成功) {
        解除队列中一个或多个线程的阻塞状态
    }
}

public class OneShotLatch {

    private final Sync sync = new Sync();

    public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(0);
    }

    public void signal() {
        sync.releaseShared(0);
    }

    private static class Sync extends AbstractQueuedSynchronizer {
        @Override
        protected int tryAcquireShared(int arg) {
            // 如果闭锁是开的(state==1)那么这个操作将成功,否则将失败
            return getState() == 1 ? 1 : -1;
        }

        @Override
        protected boolean tryReleaseShared(int arg) {
            // 打开闭锁
            setState(1);
            // 现在其他线程可以获取该闭锁
            return true;
        }
    }
}

protected boolean tryAcquire(int ignored) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        // 使用CAS操作竞争获取锁
        if (compareAndSetState(0, 1)) {
            // 记录获取到锁的当前线程
            owner = current;
            return true;
        }
        // 判断是否是当前线程拥有锁
    } else if (current == owner) {
        // 可重入
        setState(c + 1);
        return true;
    }
    return false;
}

protected int tryAcquireShared(int acquires) {
    while (true) {
        int available = getState();
        int remaining = available - acquires;
        if (remaining < 0 || compareAndSetState(available, remaining)) {
            return remaining;
        }
    }
}

protected boolean tryReleaseShared(int releases) {
    while (true) {
        int p = getState();
        if (compareAndSetState(p, p + releases)) {
            return true;
        }
    }
}

public class CasNumberRange {

    private static class IntPair {
        // 不变性条件 lower <= upper
        final int lower;
        final int upper;

        public IntPair(int lower, int upper) {
            this.lower = lower;
            this.upper = upper;
        }
    }

    private final AtomicReference<IntPair> values = new AtomicReference<>(new IntPair(0, 0));

    public int getLower() {
        return values.get().lower;
    }

    public int getUpper() {
        return values.get().upper;
    }

    public void setLower(int i) {
        while (true) {
            IntPair oldPair = values.get();
            if (i > oldPair.upper) {
                throw new IllegalArgumentException();
            }
            IntPair newPair = new IntPair(i, oldPair.upper);
            if (values.compareAndSet(oldPair, newPair)) {
                return;
            }
        }
    }

    public void setUpper(int i) {
        while (true) {
            IntPair oldPair = values.get();
            if (i < oldPair.lower) {
                throw new IllegalArgumentException();
            }
            IntPair newPair = new IntPair(oldPair.lower, i);
            if (values.compareAndSet(oldPair, newPair)) {
                return;
            }
        }
    }
}

public class ConcurrentStack<E> {

    AtomicReference<Node<E>> top = new AtomicReference<>();

    public void push(E item) {
        Node<E> newHead = new Node<>(item);
        Node<E> oldHead;
        do {
            oldHead = top.get();
            newHead.next = oldHead;
        } while (!top.compareAndSet(oldHead, newHead));
    }

    public E pop() {
        Node<E> oldHead;
        Node<E> newHead;
        do {
            oldHead = top.get();
            if (oldHead == null) {
                return null;
            }
            newHead = oldHead.next;
        } while (!top.compareAndSet(oldHead, newHead));
        return oldHead.item;
    }


    private static class Node<E> {
        public final E item;
        public Node<E> next;

        public Node(E item) {
            this.item = item;
        }
    }
}

public class MyConcurrentLinkedQueue<E> {
    private static class Node<E> {
        final E item;
        final AtomicReference<Node<E>> next;

        public Node(E item, Node<E> next) {
            this.item = item;
            this.next = new AtomicReference<>(next);
        }
    }

    // 哨兵节点
    private final Node<E> dummy = new Node<>(null, null);
    // 头节点
    private final AtomicReference<Node<E>> head = new AtomicReference<>(dummy);
    // 尾节点
    private final AtomicReference<Node<E>> tail = new AtomicReference<>(dummy);

    public boolean put(E item) {
        Node<E> newNode = new Node<>(item, null);
        while (true) {
            Node<E> curTail = tail.get();
            Node<E> tailNext = curTail.next.get();
            if (curTail == tail.get()) {
                // 队列处于中间状态,即尾节点在操作执行更新时指向倒数第二个元素的情况
                if (tailNext != null) {
                    tail.compareAndSet(curTail, tailNext);
                } else {
                    // 处于稳定状态,尝试插入新节点
                    if (curTail.next.compareAndSet(null, newNode)) {
                        // 插入成功,继续推进尾节点;如果不成功的话,会处于中间状态
                        tail.compareAndSet(curTail, newNode);
                        return true;
                    }
                }
            }
        }
    }
}

    private class Node<E> {
        private final E item;
        private volatile Node<E> next;

        public Node(E item, Node<E> next) {
            this.item = item;
        }
    }

private static AtomicReferenceFieldUpdater<Node, Node> nextUpdater = AtomicReferenceFieldUpdater.newUpdater(Node.class, Node.class, "next");

public class UnsafeLazyInitialization {
    private static Resource resource;
    public static Resource getInstance() {
        if (resource == null) {
            resource = new Resource();
        }
        return resource;
    }
}

public class SafeLazyInitialization {
    private static class ResourceHolder {
        public static Resource resource = new Resource();
    }
    public static Resource getInstance() {
        return ResourceHolder.resource;
    }
}

public class DoubleCheckLocking {
    private static Resource resource;
    public static Resource getInstance() {
        if (resource == null) {
            synchronized(DoubleCheckLocking.class) {
                if (resource == null) {
                    resource = new Resource();
                }
            }
        }
        return resource;
    }
}