Java并发编程类学习总结

线程的定义

Runnable

定义线程的接口，只有一个run()方法，而且没有返回值。一般通过实现这个接口来定义线程。

public interface Runnable {
public abstract void run();
}

经常可以用匿名内部类实现：

new Thread(new Runnable(){
@Override
public void run() {
execute(arg);
}

}).start();

Thread

Thread实现了Runnable，是线程功能实现的类。可以通过继承Thread实现自己线程的功能，主要重写run()方法。

通过继承实现会牺牲一定的灵活性，因为Java只支持单继承。

注意启动线程的时候是调用Thread.start()，而不是Thread.run().

public class Thread implements Runnable

Callable

也是定义线程类型的接口，只有call()方法，有返回值。 与Runnable不同的是，它也返回线程执行后的结果，可以接受异常的检查。

public interface Callable<V> {
/**
* Computes a result, or throws an exception if unable to do so.
*
* @return computed result
* @throws Exception if unable to compute a result
*/
V call() throws Exception;
}

线程的包装

可以用这些类对线程进行包装，获取线程的运行的状态。

Future

表示异步计算处理后的结果。可以检查计算是否完成，是否可以取消。get()方法将一直阻塞等到计算的完成才返回结果。

public interface Future<V> {
boolean cancel(boolean mayInterruptIfRunning);
boolean isCancelled();
boolean isDone();
V get() throws InterruptedException, ExecutionException;
V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}

FutureTask

可取消的异步任务。主要实现了Future接口的功能。方法中包括开始和取消任务，查询任务是否结束，检索任务执行后的结果。

FutureTask 可以包装Callable和Runnable 。

FutureTask 定义如下，实现了RunnableFuture接口

public class FutureTask<V> implements RunnableFuture<V>

而RunnableFuture继承了Runnable和Future

public interface RunnableFuture<V> extends Runnable, Future<V> {
/**
* Sets this Future to the result of its computation
* unless it has been cancelled.
*/
void run();
}

示例代码

这里借用《Java并发编程实战》中缓存计算器的例子。例子要求把计算的结果缓存下来，如果下次还有还有相同的计算请求，则直接返回缓存的计算结果，否则开始计算，把计算结果加入缓存中。

计算接口:

public interface Computable<A, V> {
V compute(A arg) throws ExecutionException;
}

实现：

public class CacheComputable<A, V> implements Computable<A, V> {
private final Map<A, Future<V>> cache = new ConcurrentHashMap<>();
private final Computable<A, V> c;
public CacheComputable(Computable<A, V> c) {
this.c = c;
}

@Override
public V compute(final A arg) throws ExecutionException {
Future<V> f = cache.get(arg);
Callable<V> eval = new Callable<V>(){
@Override
public V call() throws Exception {
return c.compute(arg);
}

};

if (f == null){
FutureTask<V> ft = new FutureTask<>(eval);
f = ft;
cache.put(arg, f);
ft.run();//开始调用c.compute
}

try {
return f.get();
} catch (InterruptedException e) {
e.printStackTrace();
throw new ExecutionException(e.getCause());
}
}
}

线程的执行

Executor

任务运行的执行器。可以解耦任务提交和计算的过程。换句话说，只管提交给执行器执行，而不用关系执行的内部过程。

任务被提交后是立即执行的。

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

ExecutorService

ExecutorService继承了Executor接口。提供了管理任务终止的功能，并且能生成Future跟踪任务异步执行的结果。

调用shutdown方法后，将继续执行之前提交的任务直到完成，但是拒绝任务的提交。submit()方法扩展了execute方法，任务提交后提供Future返回，可以通过返回Future控制任务的生命周期。invokeAny和invokeAll适合大批量的提交。

public interface ExecutorService extends Executor

ThreadFactory

一个Thread生成的工厂接口，代替用构造器创建Thread，可以灵活的按自己需求创建Thread的子类。

public interface ThreadFactory {
Thread newThread(Runnable r);
}

DefaultThreadFactory

DefaultThreadFactory是ThreadFactory接口的默认实现。其中用到了原子变量类来生成线程名。

static class DefaultThreadFactory implements ThreadFactory {
private static final AtomicInteger poolNumber = new AtomicInteger(1);
private final ThreadGroup group;
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final String namePrefix;
DefaultThreadFactory() {
SecurityManager s = System.getSecurityManager();
group = (s != null) ? s.getThreadGroup() :
Thread.currentThread().getThreadGroup();
namePrefix = "pool-" +
poolNumber.getAndIncrement() +
"-thread-";
}
public Thread newThread(Runnable r) {
Thread t = new Thread(group, r,
namePrefix + threadNumber.getAndIncrement(),
0);
if (t.isDaemon())
t.setDaemon(false);
if (t.getPriority() != Thread.NORM_PRIORITY)
t.setPriority(Thread.NORM_PRIORITY);
return t;
}
}

ThreadPoolExecutor

ThreadPoolExecutor是一个线程池，扩展实现了ExecutorService的方法。它用线程池中可用的线程来处理提交的任务。正常情况下建议使用Executors。

ThreadPoolExecutor根据基本线程数和最大线程数自动调整当前线程数；当基本线程数和最大线程数相等时，就是一个固定数量的线程池。当线程数小于基本线程数，任务到达时，即使有空闲的线程，也会创建新的线程服务当前的任务。当最大线程数设置为无边界，如Integer.MAX_VALUE，线程池会随意调整当前并发任务的数量。

线程池初始化为0个线程，当任务到达时线程才被创建，但是你可以重写方法prestartCoreThread或者prestartAllCoreThreads。

默认用DefaultThreadFactory创建线程，可以传入自定义ThreadFactory。

队列机制：

当前线程数少于基本线程数时，执行器总是倾向于创建线程处理任务而不是将任务加入队列中等待处理；

当前线程大于或者等于基本线程数，执行器总是倾向于将任务加入队列中等待处理而不是创建线程处理任务；

如果任务不能加入队列，一个新的线程会被创建，前提是创建后的线程数不能大于最大线程数，否则这个任务会被拒绝处理；

public class ThreadPoolExecutor extends AbstractExecutorService {

Executors

一个功能强大的工具类，通过静态工厂方法来生成ExecutorService、ScheduledExecutorService、ThreadFactory和Callable的实例。

Executors在创建ExecutorService时调用了ThreadPoolExecutor的方法。

CompletionService

虽然ExecutorService可以在执行任务的时候，可以用Future来跟踪执行的结果。但是对于多个返回的执行结果，却需要去轮询哪个执行过程已经结束，这无疑浪费了资源。

CompletionService这个接口就是为了解决这个问题，take()方法会阻塞直到返回最先获取的执行结果Future。

public interface CompletionService<V> {
Future<V> submit(Callable<V> task);
Future<V> submit(Runnable task, V result);
Future<V> take() throws InterruptedException;
Future<V> poll();
Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException;
}

ExecutorCompletionService

ExecutorCompletionService是CompletionService接口的实现。内部采用了BlockingQueue来保存执行结果的Future，BlockingQueue是个阻塞队列，可以很好处理并发问题。

在构造ExecutorCompletionService的时候，必须传入Executor，内部任务的执行还是依赖Executor。

public class ExecutorCompletionService<V> implements CompletionService<V> {
private final Executor executor;
private final AbstractExecutorService aes;
private final BlockingQueue<Future<V>> completionQueue;
...

Exchanger

两个线程可以用Exchanger直接交换数据，而不用其他容器。

官方实例：

class FillAndEmpty {

Exchanger exchanger = new Exchanger();

DataBuffer initialEmptyBuffer = ... a made-up type

DataBuffer initialFullBuffer = ...

class FillingLoop implements Runnable {

public void run() {

DataBuffer currentBuffer = initialEmptyBuffer;

try {

while (currentBuffer != null) {

addToBuffer(currentBuffer);

if (currentBuffer.isFull())

currentBuffer = exchanger.exchange(currentBuffer);

}

} catch (InterruptedException ex) { ... handle ... }

}

}

class EmptyingLoop implements Runnable {

public void run() {

DataBuffer currentBuffer = initialFullBuffer;

try {

while (currentBuffer != null) {

takeFromBuffer(currentBuffer);

if (currentBuffer.isEmpty())

currentBuffer = exchanger.exchange(currentBuffer);

}

} catch (InterruptedException ex) { ... handle ...}

}

}

void start() {

new Thread(new FillingLoop()).start();

new Thread(new EmptyingLoop()).start();

}

}

}

线程安全的类

同步容器类

Vector和Hashtable是重量级线程安全的，不建议使用。另外还可以用Collections.synchronizedXXXX工厂方法同步类。

并发容器

ConcurrentHashMap、ConcurrentLinkedQueue、ConcurrentLinkedDeque、ConcurrentSkipListSet和CopyOnWriteArrayList等等。他们都是线程安全的。

阻塞容器

ArrayBlockingQueue：基于数组实现的一个阻塞队列，在创建ArrayBlockingQueue对象时必须制定容量大小。并且可以指定公平性与非公平性，默认情况下为非公平的，即不保证等待时间最长的队列最优先能够访问队列。

LinkedBlockingQueue和LinkedBlockingDeque：基于链表实现的一个阻塞队列，在创建LinkedBlockingQueue对象时如果不指定容量大小，则默认大小为Integer.MAX_VALUE。

PriorityBlockingQueue：以上2种队列都是先进先出队列，而PriorityBlockingQueue却不是，它会按照元素的优先级对元素进行排序，按照优先级顺序出队，每次出队的元素都是优先级最高的元素。注意，此阻塞队列为无界阻塞队列，即容量没有上限（通过源码就可以知道，它没有容器满的信号标志），前面2种都是有界队列。

DelayQueue：基于PriorityQueue，一种延时阻塞队列，DelayQueue中的元素只有当其指定的延迟时间到了，才能够从队列中获取到该元素。DelayQueue也是一个无界队列，因此往队列中插入数据的操作（生产者）永远不会被阻塞，而只有获取数据的操作（消费者）才会被阻塞。

原子类

AtomicBoolean – 原子布尔

AtomicInteger – 原子整型

AtomicIntegerArray – 原子整型数组

AtomicLong – 原子长整型

AtomicLongArray – 原子长整型数组

AtomicReference – 原子引用

AtomicReferenceArray – 原子引用数组

AtomicMarkableReference – 原子标记引用

AtomicStampedReference – 原子戳记引用

AtomicIntegerFieldUpdater – 用来包裹对整形 volatile 域的原子操作

AtomicLongFieldUpdater – 用来包裹对长整型 volatile 域的原子操作

AtomicReferenceFieldUpdater – 用来包裹对对象 volatile 域的原子操作

同步工具

内置锁（synchronized）

每个Java对象都对应有一个内置锁。有两种方式：第一种直接修饰方法；第二种synchronized后跟加锁的对象。

示范：

public synchronized void doSomething(Object obj){
synchronized(obj){
//execute action
}
}

CountDownLatch

CountDownLatch翻译为闭锁，是一种同步工具。CountDownLatch强调的是一个线程（或多个）需要等待另外的n个线程干完某件事情之后才能继续执行。

CountDownLatch使用countDown事件去控制并发。一般在用CountDownLatch的时候，对于同一个CountDownLatch对象，有两类线程：一类是在await，等待某个countDown事件发生，然后再执行后续操作；另一类完成了前期的准备工作，发起countDown事件。

参考我之前写的文章http://blog.csdn.net/csujiangyu/article/details/44236205。

CyclicBarrier

CyclicBarrier翻译为栅栏，也是一种同步工具。它允许多个线程（或者说一组线程）在一个屏障点相互等待，直到每个线程到到达屏障点。CyclicBarrier本身是用线程的await()方法控制并发，每个线程都会参与其中，等到最后到达屏障点的线程后才执行后续操作。

参考我之前写的文章http://blog.csdn.net/csujiangyu/article/details/44338307

信号量

用来控制访问或者操作某个特定资源的线程数量。常用来实现某种资源池，或者对容器施加边界。

如果要实现一个有界缓存，最多只能有指定个数的线程访问缓存，可以这么实现：

public class BoundedCache<K, V> {
private static final int DEFAULT_SIZE = 10;

private Map<K, V> cache;
private final Semaphore sem;

public BoundedCache(int capacity){
this.sem = new Semaphore(capacity);
this.cache = new ConcurrentHashMap<>() ;
}

public BoundedCache(){
this(DEFAULT_SIZE);
}

public void put(K key, V value) throws InterruptedException{
sem.acquire();

try{
cache.put(key, value);
}finally{
sem.release();
}
}

public V get(K key) throws InterruptedException{
sem.acquire();
try{
return cache.get(key);
}finally{
sem.release();
}

}
}

ReentrantLock

一种显示锁，不同于内置锁，需要显示声明和使用。具体可参照我之前写的问题http://blog.csdn.net/csujiangyu/article/details/44002609

Condition

显示的Condition对象是一种更灵活的选择，提供了更丰富的功能：在每个锁上可以存在多个条件等待，条件等待可以是中断的或不可中断的，基于时限的等待，以及公平的或非公平的队列操作。可调用Lock.newCondition生成Condition对象。一个Lock可以有多个关联的Condition。

实现有界缓存：

public class ConditionBoundBuffer<V> {
private final V[] buf;
private int tail;
private int head;
private int count;
private Lock lock = new ReentrantLock();
private Condition notFull = lock.newCondition();
private Condition notEmpty = lock.newCondition();

@SuppressWarnings("unchecked")
public ConditionBoundBuffer(int capacity) {
buf = (V[]) new Object[capacity];
}
private void doPut(V v) {
buf[tail] = v;
if (++tail == buf.length)
tail = 0;
count++;
}
private V doTake() {
V v = buf[head];
if (++head == buf.length)
head = 0;
count--;
return v;
}
public void put(V v) throws InterruptedException {
lock.tryLock();
try {
while (count == buf.length) {
notFull.await();// 等待直到notfull
}
doPut(v);
notEmpty.signal();
} finally {
lock.unlock();
}
}
public V get() throws InterruptedException {
lock.tryLock();
try {
while (count == 0) {
notEmpty.await();// 等待直到notEmpty
}
V v = doTake();
notFull.signal();
return v;
} finally {
lock.unlock();
}
}
}