java并发编程实践

 一、基础概念总结

1.多线程程序可能存在的风险：
（1）安全性问题：多线程程序在没有充足同步的情况下，在特定的线程执行时序下，多个线程同时操作一块共享资源时，可能引发错误。
（2）活跃性问题：当多个线程存在竞争共享资源时，可能会引发死锁，饥饿（线程长时间得不到执行）及活锁（定义：不断的重复相同的操作，而且永远不会成功）或死循环等问题。
（3）性能问题：引入多线程后，如果设置不当可能发生线程的频繁的上下文切换，针对共享数据的同步抑制编译器的优化，或者线程内存缓存失效等带来的性能问题。

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

3.竞态条件定义：
在并发编程中，由于不恰当的执行时序而出现的不正确的结果是一种非常重要且槽糕的情况，引发这种槽糕情况的条件就叫做竞态条件。

4.原子操作的定义：
原子操作是指不会被线程调度机制打断的操作，这种操作一旦开始，就一直运行到结束，中间不会有任何的context switch（线程切换）。

5.锁机制:
(1) 内置锁：java提供了一种内置的锁机制来支持原子性，即同步代码块（Synchronized Block）。

synchronized（lock） {

...

}

每个java对象都可以做一个实现同步的锁，这些锁被称为内置锁（Intrinsic Lock）或监视器锁（Monitor Lock）。获得内置锁的唯一途径就是进入有这个锁保护的同步代码块或方法。
ps：java的内置锁相当于一种互斥体。
(2) 私有锁对象：将某个对象作为锁来使用。代码示例：

public class PrivateLock {

private final Object myLock = new Object();

void someMethod() {

synchronized(myLock) {

//..

}

}

}

ps:使用私有锁对象而不是对象的内置锁，有许多优点。私有锁的对象可以将锁封装起来，使客户代码无法得到锁，但客户代码可以通过公有方法来访问锁，以便参与到它的同步策略中。
(3)重入：
内置锁是可重入的，当一个线程已经获得了某个资源的锁，当它在针对该资源做其他操作时，仍然会成功完成。

6.可见性：
(1) 重排序：
编译器为了提升程序执行的性能（例如：降低占用CPU的时钟周期或内存的拷贝等使用），经常会对执行进行重排序，即不一定按照我们代码中写的顺序来执行，而是按照一种优化的方式执行，达到最终一致性的结果。
(2) 内存可见性：
线程在执行的时涉及两块内存，及每个线程自己的工作内存和主内存。针对共享变量的操作流程可以描述为，将共享变量从主内存拷贝到自己的工作内存 -> 在自己的工作内存对共享变量进行处理 -> 将处理完毕的工作变量从工作内存写回主内存。这一系列步骤并不是一个原子操作，及当中涉及到内存可见性问题。（ps：volatile通常用来处理这个问题）
(3) 非原子的64位操作：
在JVM规范中java内存模型要求 lock、unlock、read、load、assign、use、store、write这8个操作必须是原子的，但对没有volatile 修饰的long和double变量，JVM允许将这两个64bit的读操作或写操作分解为两个32bit的操作，这将也引发可见性的问题。
(4) 加锁与可见性：
加锁的含义不仅仅局限于互斥行为，还包括内存可见性。为了确保所有线程都看到共享变量的最新值，所有执行读操作或者写操作的线程都必须在同一个锁上同步。
(5) volatile变量:
java语言提供了一种稍弱的同步机制，即volatile变量，用来确保将变量的更新操作通知其他线程。当把变量声明为volatile类型后，编译器与运行时都会注意到这个变量是共享的，因为不会将该变量上的操作与其他内存操作一起重排序。另一方面，volatile变量不会被缓存在寄存器或者对其他处理器不可见的地方，因此在读取volatile类型的变量时总会返回最新的写入值。
ps:加锁机制既可以确保可见性又可以确保原子性，而volatile变量只能确保可见性。

7.发布与逸出：
（1）发布：发布一个对象指，使对象能够在当前作用域之外的代码中使用。例如将一个指向该对象的引用保存到其他代码可以访问的地方，或者在某一个非私有的方法中返回该引用，或者将引用传递到其他类的方法中。
（2）逸出：当某个不该被发布的对象被发布时，这种情况被称为逸出。

8.线程封闭：
(1)线程封闭概念：一种避免使用同步的方式就是不共享数据，如果尽在单线程内访问数据，就不需要同步。这种技术被称为线程封闭技术（Thread Confinement）。
(2) Ad-hoc线程封闭：
Ad-hoc线程封闭是指，维护线程封闭性的职责完全由程序承担。
(3)栈封闭:
栈封闭是线程封闭的一种特例，在线程封闭中，只能通过局部变量才能访问对象。即将代码全部放在方法体内是一种栈封闭的方式。
(4)ThreadLocal 类:
维持线程封闭性的一种更规范的方法是使用ThreadLocal，该类是线程将存储某个变量的一个独立的副本。

9.不变性:
(1) 不可变的对象一定是线程安全的。
(2) 当满足以下条件时，对象才是不可变的：
I.对象创建以后其状态不能修改。
II.对象的所有域都是final类型。
III.对象是正确创建的（在对象的创建期间，this引用没有逸出）。
ps:正如“除非需要更高的可见性，否则应该将所有的域都声明为私有域”是一个良好的编程习惯，“除非需要需要某个域是可变的，否则应该将其声明为final域”也是一个良好的编程习惯。

10.java监视器模式：
java监视器模式仅仅是一种代码约定，对于某个类的域变量，如果能够自始至终都使用同一种锁对象保护的同步代码来访问，则代码即为遵循java监视器模式的。（这里要和监视器锁的概念区别开来，监视器锁是 monitorenter和monitorexit指令实现的锁机制）

11.并发容器和同步容器：
(1) 比较典型的同步容器有:Vector,HashTable，SynchronizedList，SynchronizedMap，SynchronizedSet等SynchronizedXXX. 这些容器方法均使用synchronized来保证同步，及多线程对共享资源的访问都是串行化的，整体性能较并发容器较低。
(2) 比较典型的并发容器有：ConcurrentHashMap，ConcurrentSkipListMap，CopyOnWriteArrayList等，其同步是依赖AQS框架来实现（锁一般都是可重入锁）或者基于副本拷贝来实现。并发容器较同步容器性能更好，不是每次只能有一个线程访问容器，而是使用一种粒度更细的加锁机制来实现更大程度的共享。
ps:I.并发容器一般不会抛出 ConcurrentModificationException。II.类似ConcurrentHashMap的并发容器不能采用客户端加锁机制，因为并发容器没有采用synchronized内置锁而大多基于AQS框架（不是独占式的锁），所以使用客户端加锁机制来扩展并发容器的方法是不能实现的。

12.双端队列与工作密取：
简述：java6增加了两种容器类型，Dequeue和BlockingQueue，作为Queue和BlockingQueue的扩展，高效的实现了从队列头和队列的插入和删除。正如阻塞队列适用于生产者-消费者模式，双端队列适用于另一种相关模式，即工作密取。

13.阻塞和中断方法:
(1)阻塞：当线程阻塞时，通常被挂起，并处于阻塞状态（BLOCKED,WAITING或 TIMED_WAITING).阻塞操作和执行时间很长的普通操作的差别在于，被阻塞的线程必须等待某个不受它控制的事件发生后才能继续执行，例如等待IO操作完成，等待某个锁变成可用，或者等待外部计算的结束。当某个外部事件发生时，线程被置回RUNNABLE状态，并可以再次被调度执行。
ps：Object.wait和Thread.sleep都是本地方法，其实现完全由虚拟机来完成。
(2)中断：Thread提供了interrupt和interrupted方法，前者用于中断线程，后者用于恢复中断状态或判断线程是否中断。中断时一种协作机制，一个线程不能强制其他线程停止正在执行的操作而去执行其他的操作。当线程A中断线程B时，A仅仅是要求B在执行到某个可以暂停的地方停止正在执行的操作，前提是线程B愿意停下来。（ps：中断一般用于线程任务的取消工作.）

14.同步工具类：
(1) 任意对象：任何一个对象都可以作为同步工具类，因为每个对象都有一个内置锁，通过synchronized区块代码，可以利用任何对象的内置锁来同步或协调线程的控制流。
(2) 闭锁(Latch)：闭锁是一种同步工具类，可以延迟线程的进度直到其到达终止状态。闭锁的操作相当于一扇门：在闭锁到达结束状态之前，这扇门一直是关闭的，并且没有任何线程能够通过，当到达结束状态时，这扇门将允许所有线程通过。ps：CountDownLatch是一种灵活的闭锁实现。FutureTask也可以用作闭锁。
(3) 栅栏：栅栏类似于闭锁，它能阻塞一组线程直到某个事件发生。栅栏与闭锁的关键区别在于，所有线程必须同时到达栅栏位置，才继续执行。闭锁是等待事件，而栅栏是等待其他线程。说白了就是闭锁需要等待某个事件的发生才会执行阻塞的线程，而栅栏也是等待某个事件，只不过栅栏的事件是确定的，事件就是所有线程都到达栅栏。
ps:如果成功的通过栅栏，那么await将为每个线程返回一个唯一的索引号，我们可以利用这些索引来“选举”产生一个领导者线程。
(4) 信号量：计数信号量（Counting Semaphore）用来控制同时访问某个特定资源的操作数量，或者同时执行某个操作的数量。计数信号量还可用来实现某种资源池，或者对容器施加边界。
ps:计算信号量的一种简化形式是二值信号量，即初始值为1的Semaphore。二值信号量可以用作互斥体（mutex），并具备不可重入的加锁语义。谁拥有这个唯一的许可，谁就有用了互斥锁。也可以使用Semaphore将任何一种容器变成有边界的阻塞容器。

15.无限制创建线程的不足：
(1) 线程生命周期开销非常高：根据平台的不同，创建线程的开销也不同，在JVM进程中如果为每一个请求都创建一个线程开销还是很大的。创建过程本身就很耗资源。
(2) 资源消耗：活跃的线程会消耗系统资源，尤其是内存。如果可运行的线程数量多于可用处理器的数量，那么有些线程将闲置。大量的空闲线程会占用许多内存，给垃圾回收器带来压力，而且大量线程在竞争CPU资源时，还将产生其他性能开销。如果你已经拥有足够多的线程使所有CPU保持忙碌状态，那么再创建更多的线程反而会降低性能。
(3) 稳定性：在可创建线程的数量上存在一个限制。这个限制值将随着平台不同而不同，并且受多个因素制约，包括JVM的启动参数、Thread构造函数中请求的栈的大小。以及底层操作系统对线程的限制等。如果破坏了这些限制，那么很可能抛出OOM异常，并且从异常中恢复是比较困难的。

16.Executor框架：
简述：在java类库中，任务执行的主要抽象不是Thread，而是Executor：

public interface Executor {

void execute(Runnable command);

}

Executor提供了一种 标准的方法将任务的提交过程和执行过程解耦开来，并用Runnable来表示任务。Executor基于生产者-消费者模式，提交任务的操作相当于生产者，执行任务的线程相当于消费者，如果要在程序中实现一个生产者-消费者的设计，那么最简单的方式通常就是使用Executor。

17.线程池：
简述：线程池，从字面含义来看，是指管理一组同构工作线程的资源池。线程池与工作队列（work queue）密切相关，其中工作队列保存了所有等待执行的任务。工作者线程的任务很简单：从工作队列中获取一个任务，执行任务，然后返回线程池并等待下一个任务。
java类库提供了几种线程池的实现：通过调用Executors中的静态工厂方法之一来创建一个线程池。
(1) Executors.newFixedThreadPool方法: newFixedThreadPool将创建一个固定长度的线程池，每当提交一个任务时就创建一个线程，知道达到线程池的最大数量，这时线程池不再变化（如果某个线程由于发生了未预期的Exception而结束，那么线程池将补充一个线程）。
(2) Executors.newCachedThreadPool方法：newCachedThreadPool将创建一个可缓存的线程池，如果线程池的当前规模超过了处理请求时，那么将回收空闲的线程，而当需求增加时，则可以添加新的线程，线程池的规模不存在任何限制。
(3) Executors.newSingleThreadExecutor方法：newSingleThreadExecutor是一个单线程的Executor,它创建单个工作者线程来执行任务，如果这个线程异常结束，会创建另一个线程来替代。newSingleThreadExecutor能确保依照任务在队列中的顺序来串行执行。
(4) Executors.newScheduledThreadPool方法：newScheduleThreadPool创建一个固定长度的线程池，而且可以延迟或定时的方式来执行任务，类似于Timer。
ps:JVM只有在所有（非守护）线程全部终止后才会退出。因此，如果无法正确的关闭Executor，那么JVM将无法结束。只有当任务都是同类型的并且相互独立时，线程池的性能才能达到最佳，如果提交的任务依赖于其他任务，除非线程池无限大，否则可能造成死锁。

18.Executor的生命周期：
简述：为了解决生命周期的问题，Executor扩展了ExecutorService方法，添加了一些用于管理生命周期的方法。ExecutorService的生命周期有3中状态：运行、关闭和已终止。

19.延迟任务和并发任务：
简述：Timer在执行所有定时任务时只会创建一个线程。如果某个任务的执行时间过长，那么将破坏其他TimeTask的定时准确性。此时的TimeTask存在一定的缺陷，所以应该使用Executors.newScheduledThreadPool来实现延迟和并发任务。

20.线程泄露：
Timer的另一个问题是，如果TimeTask抛出了一个未检查的异常，那么Timer将表现出槽糕的行为。Timer线程并不捕获异常，因此当TimeTask抛出未检查的异常时将终止定时线程，此时Timer也不会恢复线程的执行，而是错误的认为整个Timer都被取消了。已经被调度但是还未执行的TimerTask也不会执行和调度。这个问题成为“线程泄露”。

21.异构任务并行化中存在缺陷：
只有当大量相互独立且同构的任务可以并发进行处理时，才能体现出将程序的工作负载分配到多个任务中带来的真正的性能提升。ps：web请求服务，就是一个同构的案例。

22.线程的中断状态:
每个线程都有一个boolean类型的中断状态。当中断线程时，这个线程的中断状态将被设置为true.静态的interrupted方法将清除当前线程的中断状态，并返回它之前的值，这也是清除中断状态的唯一方法。调用interrupt并不意味着立即停止目标线程正在运行的工作，而只是传递了请求的中断消息。即对中断的正确理解是：它并不会真正的中断一个正在运行的线程，而只是发出中断请求，然后由线程在下一个适合的时刻中断自己。
ps:在使用interrupted时应该小心，因为它只会清除当前线程的中断状态。如果在调用interrupted时返回了true，那么除非你想屏蔽这个中断，否则必须对它进行处理，可以抛出InterruptedException,或者通过再次调用interrupt来恢复中断状态.

23.UncaughtExceptionHandler:
在java API中同样提供了UncaughtExceptionHandler,它能检测出某个线程由于未捕获的异常而终结的情况。当一个线程由于未捕获异常而退出时，JVM会把这个时间报告给应用程序的UncaughtExceptionHandler异常处理器。

public interface UncaughtExceptionHandler {

void uncaughtException(Thread t, Throwable e);

}

24.使用ThreadLocal任务：
在线程池的线程中使用ThreadLocal才有意义，而在线程池中的线程不应该使用ThreadLocal在任务之间传递值。

25.线程饥饿死锁:
在线程池中，如果任务依赖于其他任务，那么可能产生死锁。在单线程的Executor中，如果一个任务将另一个任务提交到同一个Executor，并且等待这个被提交的任务的结果，那么通常会引发死锁。第二任务停留在工作队列中，并等待第一个任务完成，而第一个任务又无法完成，因为他在等待第二个任务完成。在更大的线程池中，如果所有正在执行任务的线程都由于等待其他处于工作队列的任务而阻塞，那么会发生同样的问题。这种现象称为线程饥饿死锁。

26.设置线程池的大小:
简述：线程池需要避免过大和过小设置，这两种极端。如果线程池设置的过大，那么大量的线程将在相对较少的CPU和内存资源上发生竞争，这不仅会导致更高的内存使用量，而且还可能耗尽资源。如果线程池过小，那么将导致许多空闲的处理器无法执行工作，从而降低吞吐率。所以合理的设置线程池的大小是很重要的。
(1) 计算密集型任务：对于计算密集型任务，在拥有N个处理器的系统上，当线程池的大小设置为 N+1时，通常实现最优的利用率。

(2) 混合型任务：当既有IO操作又有一定的计算操作的任务时，需要合理估算出线程的等待时间和计算时间的比值，用以估算线程池的容量。线程池大小的计算公式为：
N(thread) = N(cpu)U(cpu)(1+W/C)
N(thread):线程池的容量。
N(cpu):处理器的个数。
U(cpu):CPU的使用率。
W/C:任务等待时间和计算时间的比值。

二、多线程开发实施案例总结

1.在现有的线程安全类中添加功能：
题目：向现有的java线程安全的集合类中添加”若没有则添加“的同步方法。
(1) 扩展Vector集合类添加同步方法：

//正确的实现

@ThreadSafe

public class BetterVector<E> extends Vector<E> {

public synchronized boolean putIfAbsent(E x) {

boolean absent = !contains(x);

if(absent) {

add(x);

}

return absent;

}

}

ps：通过扩展方式添加的同步方法和原有线程安全集合类使用了同一种锁。
(2) 客户端加锁机制实现:
I.错误的实现：

//错误的案例

@NotThreadSafe

public class ListHelper<E> {

private 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;

}

}

ps:使用这种方式扩展java的线程安全集合不是线程安全的，虽然方法也声明为synchronized，但是这个锁和List集合使用的锁不是同一把锁，即contains声明的锁和putIfAbsent声明的锁不是同一把，所以不能保证操作的原子性，所以不是线程安全的。

II.正确的实现:

//正确的实现

@ThreadSafe

public class ListHelper<E> {

public List<E> list = Collections.synchronizedList(<new ArrayList<E>());

public boolean putIfAbsent(E x) {

synchronized(list) {

boolean absent = !list.contains(x);

if(absent) {

list.add(x);

}

return absent;

}

}

}

ps:这样即使用了正确的客户端加锁实现。

2.迭代器与ConcurrentModificationException:
(1) 案例：当迭代一个共享的集合容器时，当集合被其他线程修改时 即计数器发生变化，将导致迭代过程抛出ConcurrentModificationException。所以在堆共享集合进行迭代的地方，一定要使用客户端加锁机制来规避。

...

List<E> list = Collections.synchronizedList<new ArrayList<E>());

synchronized(list) {

for(E item : list) {

doSomething(item);

}

}

...

(2)隐藏迭代器:
案例：有些迭代器调用并不是显示的调用，而是java内部类中调用的，比如StringBuilder.append(Object)方法，这个方法会调用toString方法，toString方法会调用容器迭代器，此时如果不加锁加以控制仍然可能抛出ConcurrentModificationException。

...

private final Set<Integer> set = new HashSet<Integer>();

public synchronized void add(Integer i) {set.add(i);}

public synchronized void remove(Integer i) {set.remove(i);}

public synchronized void addTenThings() {

Random r = new Random();

for(int i = 0; i < 10; i++) {

add(r.nextInt());

}

System.out.println(set);

}

...

ps:如果addTenThings不加synchronized声明，可能抛出ConcurrentModificationException异常。

3.CountDownLatch的一个实用DEMO：
案例:通过CountDownLatch来控制多个线程的启动和结束。

public class TestHarness {

public long timeTasks(int nThreads,final Runnable task) throws InterruptedException {

final CountDownLatch startGate = new CountDownLatch(1);

final CountDownLatch endGate = new CountDownLatch(nThreads);

 

for(int i = 0; i < nThreads; i++) {

Thread t = new Thread(){

public void run() {

try {

startGate.await();

try {

task.run();

} finally {

endGate.counDown();

}

} catch(InterruptedException ignored) {}

}

};

t.start();

}

startGate = countDown();

endGate.await();

System.out.println("run over!");

}

简述：CountDownLatch的用法可以概括为当CountDownLatch内置的state的数值被设置为0时，则唤醒在别这个Latch上所有线程执行。ps:CountDownLatch也是基于AQS来实现的。

4.FutureTask一个异步任务的应用：
简述：FutureTask实现了Future语义，表示一种抽象的可生成结果的计算。FutureTask通过Callable来实现，相当于一个可生成结果的Runnable，并且可以处于三种状态：等待运行，正在运行和运行完成。
案例：使用FutureTask实现异步加载任务:

public class Preloader {

private final FutureTask<ProductInfo> future = new FutureTask<ProductInfo>(new Callable<ProductInfo>() {

public ProductInfo call() throws DataLoadException {

return loadProductInfo();

}

});

private final Thread thread = new Thread(future);

 

public void start() {thread.start();}

 

public productInfo get() throws DataLoadException,InterruptedException{

try{

reutrn future.get();

} catch(ExecutionException e) {

Throwable cause = e.getCause();

if(cause instanceof DataLoadException) {

throw (DataLoadException)cause;

} else {

...

}

}

}

}

ps:从DEMO来看，Callable 和 Runnable的区别就是Callable是一种可以拿到返回值的异步流程应用工具。

5.使用Semaphore为容器设置边界:

public class BoundedHashSet<T> {

private final Set<T> set;

private final Semaphore sem;

 

public BoundedHashSet（int bound) {

this.set = Collections.synchronizedSet(new HashSet<T>());

sem = new Semaphore(bound);

}

 

public boolean add(T o) throws InterruptedException {

sem.acquire();

boolean wasAdded = false;

try {

wasAdded = set.add(o);

return wasAdded;

}

finally {

if(!wasAdded) {

sem.release();

}

}

}

 

public boolean remove(Object o) {

boolean wasRemoved = set.remove(o);

if(wasRemoved) {

sem.release();

}

return wasRemoved();

}

}

ps：通过信号量将原本无边界的Set容器，变成了边界为Bound的有界容器。

1.CyclicBarrier控制所有线程都执行完毕后，才出发后面的Action:

2.public class CellularAutomata {

3. private final Board mainBoard;

4. private final CyclicBarrier barrier;

5. private final Worker[] workers;

6.

7. public CellularAutomata(Board board) {

8. this.mainBoard = board;

9. int count = Runtime.getRuntime().availableProcessors();

10. this.barrier = new CyclicBarrier(count,new Runnable() {

11. public void run() {

12. mainBoard.commitNewValues();

13. }});

14. this.works = new Worker[count];

15. for(int i = 0; i < count; i++) {

16. workers[i] = new Worker(mainBoard.getSubBoard(count,i));

17. }

18.}

19.

20.private class Worker implements Runnable {

21. private final Board board;

22. public Worker(Board board) { this.board = board; }

23.

24. public void run() {

25. while(!board.hasConverged()) {

26. for(int x = 0; x < board.getMaxX(); x++) {

27. for(int y = 0; y < board.getMaxY(); y++) {

28. board.setNewValue(x,y,coumputeValue(x,y));

29. }

30. }

31. try {

32. barrier.await();

33. } catch(InterruptedException ex) {

34. return;

35. } catch(BrokenBarrierException ex) {

36. return;

37. }

38. }

39. }

40.}

41.

42.public void start() {

43. for(int i = 0; i < workers.length; i++) {

44. new Thread(workers[i]).start();

45. }

46.}

ps:这个案例和前面那个CountDownLatch对比，其实两种同步工具类可以实现相同的功能，只是语义略有不同。

7.基于Executor的web服务器：

class TaskExecutionWebServer {

private static final int NTHREADS = 10;

private static final Executor exec = Executors.newFixedThreadPool(NTHREADS);

 

public static void main(String[] args) throws IOException {

ServerSocket socket = new ServerSocket(80);

while(true) {

final Socket connection = socket.accept();

Runnable task = new Runnable() {

public void run() {

handleRequest(connection);

}

}；

exec.execute(task);

}

}

}

ps:可见Executor这个框架封装了线程池。通过将任务的提交和执行解耦开来，从而无须太大的困难就可以为某种类型的任务执行和修改执行策略。可见Executor抽象程度更高，扩展性更好。

8.Callable和Future：
简述：Executor框架使用Runnable作为其基本的任务表现形式。Runnable是一种有很大局限的抽象，虽然run能写入到日志文件或者将结果放入某个共享的数据结构，但它不能返回一个值或抛出一个受检查的异常。但是Callable作为抽象的任务可以携带一个返回值。Future表示一个任务的生命周期，并提供了相应的方法来判断是否已经完成或取消，以及获取任务的结果和取消任务等。ExecutorService中的所有submit方法都将返回一个Future，从而将一个Runnable或者Callable提交给Executor，并得到一个Future用来获得任务的执行或取消任务。
实例：使用Future下载图片:

public class FutureRender {

private final ExecutorService executor = Executors.nnewFixedThreadPool（10）；

 

void renderPage(CharSequence source) {

final List<ImageInfo> imageInfos = scanForImageInfo(source);

 

Callable<List<ImageData>> task = new Callable<List<ImageData>>() {

public List<ImageData> call() {

List<ImageData> result = new ArrayList<ImageData>();

for(ImageInfo imageInfo : imageInfos) {

result.add(imageInfo.downloadImage());

}

reutrn result;

}

};

 

Future<List<ImageData>> future = executor.submit(task);

renderText(source);

 

try {

Future<List<ImageData> imageData = future.get();

for(ImageData data : imageData) {

renderImage(data);

}

} catch(InterruptedException e) {

Thread.currentThread().interrupt();

future.cancel(true);

} catch(ExecutionExecution e) {

throw launderTrowable(e.getCause());

}

}

}

三、技术细节概述

1.阻塞队列 BlockingQueue：
简述：BlockingQueue扩展了Queue，增加了可阻塞的插入和获取等操作。如果队列为空，那么获取元素的操作将一直阻塞，知道队列中出现一个可用的元素。如果队列已满（对于有界队列来说），那么插入元素的操作将一直阻塞，知道队列中出现可用的元素。
阻塞队列 BlockingQueue 提供了可阻塞的put和take方法，以及支持定时的offer和poll方法。

(1) ArrayBlockingQueue.put方法实现：

/**

* Inserts the specified element at the tail of this queue, waiting

* for space to become available if the queue is full.

*

* @throws InterruptedException {@inheritDoc}

* @throws NullPointerException {@inheritDoc}

*/

public void put(E e) throws InterruptedException {

if (e == null) throw new NullPointerException();

final E[] items = this.items;

final ReentrantLock lock = this.lock;

lock.lockInterruptibly();

try {

try {

while (count == items.length)

notFull.await();

} catch (InterruptedException ie) {

notFull.signal(); // propagate to non-interrupted thread

throw ie;

}

insert(e);

} finally {

lock.unlock();

}

}

 

insert方法：

/**

* Inserts element at current put position, advances, and signals.

* Call only when holding lock.

*/

private void insert(E x) {

items[putIndex] = x;

putIndex = inc(putIndex);

++count;

notEmpty.signal();

}

ps：当队列满时 会调用notFull.await() 阻塞等待。当线程被中断时，会调用signal唤醒一个等待在该对象上的线程。insert方法会调用notEmpty.signal唤醒一个等待的线程。

ArrayBlockingQueue.take方法实现和put相反：

public E take() throws InterruptedException {

final ReentrantLock lock = this.lock;

lock.lockInterruptibly();

try {

try {

while (count == 0)

notEmpty.await();

} catch (InterruptedException ie) {

notEmpty.signal(); // propagate to non-interrupted thread

throw ie;

}

E x = extract();

return x;

} finally {

lock.unlock();

}

}

 

extract方法：

/**

* Extracts element at current take position, advances, and signals.

* Call only when holding lock.

*/

private E extract() {

final E[] items = this.items;

E x = items[takeIndex];

items[takeIndex] = null;

takeIndex = inc(takeIndex);

--count;

notFull.signal();

return x;

}

ps：extract方法调用notFull.signal欢迎一个等待的线程。

简述：阻塞队列put和take的阻塞等待机制借助 ReentrantLock 来实现，notEmpty和notFull分别是 ReentrantLock的Condition。通过condition的await和signal来协调队列的阻塞和唤醒机制。

(2) BlockingQueue 的offer和 poll方法 区别于put和take，提供了阻塞定时的机制。即offer可以设定线程阻塞的时间，当阻塞等待超过指定的时间后，则自动中断阻塞。当不设置阻塞等待超时时间时，即不阻塞，当获取成功是返回true，获取失败（即队列空）则返回false；poll和offer逻辑刚好相反。
offer的两个重载方法：

/**

* Inserts the specified element at the tail of this queue, waiting

* up to the specified wait time for space to become available if

* the queue is full.

*

* @throws InterruptedException {@inheritDoc}

* @throws NullPointerException {@inheritDoc}

*/

public boolean offer(E e, long timeout, TimeUnit unit)

throws InterruptedException {

 

if (e == null) throw new NullPointerException();

long nanos = unit.toNanos(timeout);

final ReentrantLock lock = this.lock;

lock.lockInterruptibly();

try {

for (;;) {

if (count != items.length) {

insert(e);

return true;

}

if (nanos <= 0)

return false;

try {

nanos = notFull.awaitNanos(nanos);

} catch (InterruptedException ie) {

notFull.signal(); // propagate to non-interrupted thread

throw ie;

}

}

} finally {

lock.unlock();

}

}

ps：阻塞等待定时函数。

/**

* Inserts the specified element at the tail of this queue if it is

* possible to do so immediately without exceeding the queue's capacity,

* returning <tt>true</tt> upon success and <tt>false</tt> if this queue

* is full. This method is generally preferable to method {@link #add},

* which can fail to insert an element only by throwing an exception.

*

* @throws NullPointerException if the specified element is null

*/

public boolean offer(E e) {

if (e == null) throw new NullPointerException();

final ReentrantLock lock = this.lock;

lock.lock();

try {

if (count == items.length)

return false;

else {

insert(e);

return true;

}

} finally {

lock.unlock();

}

}

ps：当队列空时，则不阻塞，直接返回false。

2.CopyOnWriteArrayList：
简述：该并发容器采用写入时复制策略（Copy-On-Write），容器每次修改时，都会修改并重新发布一个新的容器副本，从而实现可变性。因为每次修改都复制整个数组，因此适用读多写少的场景。

CopyOnWriteArrayList.add方法：

/**

* Appends the specified element to the end of this list.

*

* @param e element to be appended to this list

* @return <tt>true</tt> (as specified by {@link Collection#add})

*/

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();

}

}

简述：CopyOnWriteArrayList同样借助了可重入锁 ReentrantLock来控制同步，每次写入建立了新的数据集合。

3.DelayQueue：
简述：DelayQueue是一个延时队列，在DelayQueue中，只有某个元素逾期后，才能从DelayQueue中执行take操作。从DelayQueue中返回的对象将根据他们的延迟时间进行排序。
DelayQueue.take方法：

public E take() throws InterruptedException {

final ReentrantLock lock = this.lock;

//如果当前线程未被中断，则获取锁

lock.lockInterruptibly();

try {

for (;;) {

E first = q.peek();

if (first == null) {

available.await();

} else {

long delay = first.getDelay(TimeUnit.NANOSECONDS);

if (delay > 0) {

long tl = available.awaitNanos(delay);

} else {

E x = q.poll();

assert x != null;

if (q.size() != 0)

available.signalAll(); // wake up other takers

return x;

 

}

}

}

} finally {

lock.unlock();

}

}

四、实践DEMO

1.DelayQueue的一个实用DEMO：
场景：下面的代码模拟一个考试的日子，考试时间为120分钟，30分钟后才可交卷，当时间到了，或学生都交完卷了者考试结束。线程的关闭参考Java编程思想中例子，将exec传给Student的一个内部类，通过他来关闭。

class Student implements Runnable,Delayed{

private String name;

private long submitTime;//交卷时间

private long workTime;//考试时间

public Student(String name, long submitTime) {

this.name = name;

workTime = submitTime;

//都转为转为ns

this.submitTime = TimeUnit.NANOSECONDS.convert(submitTime, TimeUnit.MILLISECONDS) + System.nanoTime();

}

public void run() {

System.out.println(name + " 交卷,用时" + workTime/100 + "分钟");

}

public long getDelay(TimeUnit unit) {

return unit.convert(submitTime - System.nanoTime(), TimeUnit.NANOSECONDS);

}

public int compareTo(Delayed o) {

Student that = (Student) o;

return submitTime > that.submitTime?1:(submitTime < that.submitTime ? -1 : 0);

}

public static class EndExam extends Student{

private ExecutorService exec;

public EndExam(int submitTime,ExecutorService exec) {

super(null,submitTime);

this.exec = exec;

}

public void run() {

exec.shutdownNow();

}

}

}

 

class Teacher implements Runnable{

private DelayQueue<Student> students;

 

public Teacher(DelayQueue<Student> students) {

this.students = students;

}

public void run() {

try {

System.out.println("考试开始……");

while (!Thread.interrupted()) {

students.take().run();

}

System.out.println("考试结束……");

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

 

public class DelayQueueDemo {

static final int STUDENT_SIZE = 45;

public static void main(String[] args) {

Random r = new Random();

DelayQueue<Student> students = new DelayQueue<Student>();

ExecutorService exec = Executors.newCachedThreadPool();

for(int i = 0; i < STUDENT_SIZE; i++){

students.put(new Student("学生" + i, 3000 + r.nextInt(9000)));

}

students.put(new Student.EndExam(12000,exec));//1200为考试结束时间

exec.execute(new Teacher(students));

}

}

 

转载自：http://blog.csdn.net/lantian0802/article/details/19285937