Java Concurrency In Practice

线程安全

定义

A class is thread-safe if it behaves correctly when accessed from multiple threads, regardless of the scheduling or interleaving of the execution of those threads by the runtime environment, and with no additional synchronization or other coordination on the part of the calling code.

A Stateless Servlet

独立的, 无状态的, 这种程序一定是线程安全的

@ThreadSafe
public class StatelessFactorizer implements Servlet {
    public void service(ServletRequest req, ServletResponse resp) {
        BigInteger i = extractFromRequest(req);
        BigInteger[] factors = factor(i);
        encodeIntoResponse(resp, factors);
    }
}

竞争条件(Race Conditions), 原子性问题

做了如下改动, 这个代码就是非线程安全的, 因为他存在Race Conditions

++count, 是典型的read-modify-write operation, 非原子的, 在多线程情况下, 你无法保证在read到write的过程中, count不被改动, 所有运行结果取决于执行顺序, 这就产生竞争条件

多个进程并发访问和操作同一数据且执行结果与访问的特定顺序有关, 称为竞争条件(Race Conditions)

@NotThreadSafe
public class UnsafeCountingFactorizer implements Servlet {
    private long count = 0;

    public long getCount() { return count; }

    public void service(ServletRequest req, ServletResponse resp) {
        BigInteger i = extractFromRequest(req);
        BigInteger[] factors = factor(i);
        ++count; //非原子操作

        encodeIntoResponse(resp, factors);
    }

}

可见性问题

直接看例子, 下面的代码会得到什么结果, 正常的期望是子线程结束, 并打印出42
但是结果会出人意外, 每次都不一定, 有可能子线程根本就不终止, 或打印出0
这就是典型的可见性问题, 由于缺乏同步机制, 你在主线程中修改number和ready的值, 在子线程中不一定是可见的

对应打印0, 可能会更加令人意外, 看上去ready=true被先执行了, 实际就是这样的
这是"reordering”现象, 在单线程中, 如果reordering不会影响最终结果, 那么编译器就有可能会做优化调整执行顺序

public class NoVisibility {
    private static boolean ready;
    private static int number;

    private static class ReaderThread extends Thread {
        public void run() {
            while (!ready)
                Thread.yield();
            System.out.println(number);
        }
    }

    public static void main(String[] args) {
        new ReaderThread().start();
        number = 42;
        ready = true;
    }
}

Nonatomic 64-bit Operations

对于大部分情况下, 即使不加锁, 还是可以满足"out-of-thin-air safety”(最底限的安全性), 即虽然有可能读到过期数据, 但毕竟是真实存在的数据
但对于64-bit numeric variables (double and long) that are not declared volatile, 却是例外, 会产生更大的问题
因为java会将一个64位的操作分成2个32位的操作, 这就存在在不同线程中和读到写到一半的long或double

悲观锁, synchronized

显而易见, 最常用的方法就是用锁来解决线程安全问题, 虽然锁是一种极高耗费的操作
在Java中锁可以同时解决原子性问题和可见性问题
通过锁来解决原子性问题, 比较显而易见, 而解决可见性问题就稍微阴晦一些,所以通过下面的图说明

Java对锁做了封装, 可以方便的使用synchronized来解决, 当然你也可以显式的使用lock(不推荐,除非特殊情况)

A synchronized block has two parts: a reference to an object that will serve as the lock, and a block of code to be guarded by that lock.

synchronized (lock) {
    // Access or modify shared state guarded by lock
}

Every Java object can implicitly act as a lock for purposes of synchronization; these built-in locks are called intrinsic locks or monitor locks.
The lock is automatically acquired by the executing thread before entering a synchronized block and automatically released when control exits the synchronized block, whether by the normal control path or by throwing an exception out of the block. The only way to acquire an intrinsic lock is to enter a synchronized block or method guarded by that lock.

需要注意的是, 这种锁虽然很方便, 但是他锁的是那段逻辑, 和你需要保护的那个属性是没有关系的, 其他线程可以通过其他的逻辑任意修改这个属性
所以是否达成线程安全, 是需要开发者设计和保证的

下面给出如何利用synchronized来解决线程安全问题
需要平衡安全和效率, 所以实现没有对整个service函数加锁, 而是只是对其中部分需要互斥逻辑加锁

@ThreadSafe
public class CachedFactorizer implements Servlet {
    @GuardedBy("this") private BigInteger lastNumber;
    @GuardedBy("this") private BigInteger[] lastFactors;
    @GuardedBy("this") private long hits;
    @GuardedBy("this") private long cacheHits;

    public synchronized long getHits() { return hits; } //读数据也加锁,避免读到写操作的中间数据或过期数据
    public synchronized double getCacheHitRatio() {
        return (double) cacheHits / (double) hits;
    }

    public void service(ServletRequest req, ServletResponse resp) {
        BigInteger i = extractFromRequest(req);
        BigInteger[] factors = null;
        synchronized (this) { //部分加锁,降低锁粒度,提高效率
            ++hits;
            if (i.equals(lastNumber)) {
                ++cacheHits;
                factors = lastFactors.clone();
            }
        }
        if (factors == null) {
            factors = factor(i);
            synchronized (this)  {
                lastNumber = i;
                lastFactors = factors.clone();
            }
        }
        encodeIntoResponse(resp, factors);
    }
}

Reentrancy, 可重入性

Reentrancy means that locks are acquired on a per-thread rather than per-invocation basis. 意味着同一个线程可以多次lock这个锁, 只是增加锁计数
看下面的例子, 如果不支持Reentrancy, 就会导致死锁

public class Widget {
    public synchronized void doSomething() { //第二次锁同一对象, 死锁
        ...
    }
}

public class LoggingWidget extends Widget {
    public synchronized void doSomething() { //第一次锁对象
        System.out.println(toString() + ": calling doSomething");
        super.doSomething();
    }
}

Volatile Variables

锁是一个最平庸的方案, 但问题是比较低效, 所以Java也支持一些更高效的方法来解决线程安全问题

Volatile, 解决可见性问题
底层应该是通过memory barrier来实现的
被声明为Volatile的变量,
被写入后, 会将change立即从cache更新到memory中, 等同于离开synchronized区域(从可见性上看)
被读时, 如果已经被其他线程改变, 也会先从memory中同步最新数据, 等同于进入synchronized区域(从可见性上看)

但需要注意的是, Volatile只解决可见性问题, 而不能解决原子性问题, 这个还需要其CAS配合, 称为lock-free技术

另外, memory barrier还能保证有序性, 在memory barrier之前写的指令一定先于之后写的指令

我看到的lock-free比较典型的应用是, disruptor, 参考LMAX Disruptor 原理

并发容器

解决并发问题最简单, 最安全的方法, 就是直接使用线程安全的容器和类来管理状态

同步容器

在谈并发容器前, 先看看早期的java版本中提供同步容器, Vector and Hashtable, 他们是线程安全的
问题是, 对整个数据结构都只能串行操作, 这个效率是非常低下的, 尤其当size比较大的时候

并且也无法应对复合操作, 比如put-if-absent, read-modify-write, 仍然要在客户端加锁, 效率更低, 因为加了两遍锁

比如下面的deleteLast, 很有可能会抛ArrayIndexOutOfBoundsException异常

public static void deleteLast(Vector list) {
    int lastIndex = list.size() - 1;
    list.remove(lastIndex);
}

所以仍然需要加上锁, 很低效

public static void deleteLast(Vector list) {
    synchronized (list) {
        int lastIndex = list.size() - 1;
        list.remove(lastIndex);
    }
}

对于比较新的collection类而言, 这些问题仍然存在, 因为它本身就不是线程安全的, 尤其对于大量使用的Iterator
因为它是fail-fast, 即在迭代过程中发现容器被修改, 会抛出unchecked ConcurrentModificationException
而且在collection的操作中处处都隐含了对Iterator的使用, 比如for each, toString, hashCode and equals

所以下面就介绍一些并发容器, 以方便并发处理

ConcurrentHashMap

ConcurrentHashMap is a concurrent replacement for a synchronized hash-based Map, Java 6 adds ConcurrentSkipListMap and ConcurrentSkipListSet, which are concurrent replacements for a synchronized SortedMap or SortedSet (such as treeMap or treeSet wrapped with synchronizedMap).

ConcurrentHashMap用于替换现有的同步hashmap, 并且在Java 6里面增加了ConcurrentSkipListMap, ConcurrentSkipListSet替换现有的同步的treeMap,treeSet, 从名字上看, 应该是用skiplist来替换红黑树

ConcurrentHashMap关键是使用更细粒度的分离锁, 即把整个hashtable, 分成segment, 并独立局部加锁(参考, http://www.infoq.com/cn/articles/ConcurrentHashMap)

并且在读的时候不用加锁, 原因是对get要访问的共享变量都使用了voliate, 保证读到最新数据
ConcurrentHashMap还提高了复合操作的接口, 把他们原子化, 你可以直接使用

public interface ConcurrentMap<K,V> extends Map<K,V> {
    // Insert into map only if no value is mapped from K
    V putIfAbsent(K key, V value);

    // Remove only if K is mapped to V
    boolean remove(K key, V value);

    // Replace value only if K is mapped to oldValue
    boolean replace(K key, V oldValue, V newValue);

    // Replace value only if K is mapped to some value
    V replace(K key, V newValue);
}

CopyOnWriteArrayList

CopyOnWriteArrayList is a concurrent replacement for a synchronized List that offers better concurrency in some common situations and eliminates the need to lock or copy the collection during iteration. (Similarly, CopyOnWriteArraySet is a concurrent replacement for a synchronized Set.)

CopyOnWrite技术, 用于很多地方, 只有读操作的时候保留一份copy, 只有当写操作发生的时候才创建新的copy, 用于change, 这样旧的copy仍然可以被并发访问
当然copy是需要开销的, 所以比较适合大量读操作的场景
具体实现参考, http://www.molotang.com/articles/558.html
读的时候没有问题, 不过要注意在用iterator时候, 它其实是做了snapshot, 新的改动是读不到的, 并且迭代过程中不能change, 否则会报异常
写的时候需要加锁, 是同步的, 先加锁，拷贝一份新数组对象，在新的对象上改动，赋值回去，解锁

BlockingQueue

传统的Queue和PriorityQueue都需要用户去自己控制并发同步
在Concurrent包中添加了BlockingQueue来提供更好的并发queue, 参考BlockingQueue

BlockingQueue很容易的就可以实现, produer/consumer模式

ArrayBlockingQueue: 基于数组的阻塞队列实现, input和output指针公用一把锁, 无法实现真正的并发, 但在插入和删除数据是不需要产生和销毁node对象

LinkedBlockingQueue: 基于链表的阻塞队列, 最常用到的BlockingQueue, input和output指针使用独立的锁, 所以较好的并发性. 需要注意的是, 默认容量是Integer.MAX_VALUE, 所以最好指定大小, 否则会爆内存

DelayQueue: 当元素被放入到queue中后, 必须过了延迟时间后, 才能被取出. 应用场景, 管理一个超时未响应的连接队列

PriorityBlockingQueue: 基于优先级的阻塞队列(优先级的判断通过构造函数传入的Compator对象来决定), 但需要注意的是PriorityBlockingQueue并不会阻塞数据生产者, 所以当心爆内存

SynchronousQueue: 无缓冲的等待队列, producer直接把数据给consumer, 而不用真正使用作为中介的queue

BlockingDeque

 

具体参考, Java多线程（六）之Deque与LinkedBlockingDeque深入分析

关键是Deque可以实现work stealing模式
典型场景, 多个队列, 多个consumer, 为了避免竞争的耗费, 可以每个queue分配一个consumer, 但这样会有分配不均的情况, 所以可以让空闲的consumer从队尾取数据, 以避免在队头产生竞争

ConcurrentLinkedQueue

这个应该是最高效的并发Queue, 因为不是使用悲观锁的blocking, 号称是"lock free”
其实是使用volatile和cas等更底层的机制, 来实现并发互斥

聊聊并发（六）——ConcurrentLinkedQueue的实现原理分析

Synchronizer

用于并发同步的对象都可以称为synchronizer, 所以前面提到的并发容器也都是synchronizer
还有其他一些synchronizer

CountDownLatch

CountDownLatch是一种闭锁, 它通过内部一个计数器count来标示状态, 当count>0时, 所有调用其await方法的线程都需等待, 当通过其countDown方法将count降为0时所有等待的线程将会被唤起执行
应用场景, 倒计时, 当count减为0时, 所有线程一起执行

Semaphore

类实际上就是操作系统中谈到的信号量的一种实现, 提供acquire和release方法来实现资源管理

CyclicBarrier

和latch有点象, 只是解锁条件不同
Barrier用于一组线程互相等待，直到到达某个公共屏障点 (common barrier point)时, 一起被执行, 因为该barrier在释放等待线程后可以重用，所以称它为循环的barrier
场景, 旅游时集合, 必须等所有人都到齐, 才能去下个景点

Future and FutureTask

Future接口

A Future represents the result of an asynchronous computation.

• cancel(boolean mayInterruptIfRunning): 取消任务的执行, 参数指定是否立即中断任务执行，或者等等任务结束
• isCancelled: 任务是否已经取消, 任务正常完成前将其取消, 则返回 true
• isDone: 任务是否已经完成, 需要注意的是如果任务正常终止、异常或取消，都将返回true
• V get(): 等待任务执行结束, 然后获得V类型的结果
  异常类型, InterruptedException 线程被中断异常, ExecutionException任务执行异常, CancellationException任务被取消
• V get (long timeout, TimeUnit unit) timeout 超时时间，uint 时间单位
  除了上面的异常, 还会抛出超时异常TimeoutException

FutureTask

The FutureTask class is an implementation of Future that implements Runnable, and so may be executed by an Executor.

Callable和Runnable
实现Callable接口的类和实现Runnable的类都是可被其它线程执行
- Callable和Runnable有几点不同：  
- Callable规定的方法是call()，而Runnable规定的方法是run().
- Callable的任务执行后可返回值，而Runnable的任务是不能返回值的。  
- call()方法可抛出异常，而run()方法是不能抛出异常的。
- 运行Callable任务可拿到一个Future对象

FutureTask是对Future接口的实现, 并且还实现了Runable所以可以被Executor直接执行

FutureTask<String> future =
new FutureTask<String>(new Callable<String>() {
  public String call() {
    return searcher.search(target);
}});
executor.execute(future);

try {   
    result = future.get(5000, TimeUnit.MILLISECONDS); //取得结果，同时设置超时执行时间为5秒。同样可以用future.get()，不设置执行超时时间取得结果   
} catch (InterruptedException e) {   
    futureTask.cancel(true);   
} catch (ExecutionException e) {   
    futureTask.cancel(true);   
} catch (TimeoutException e) {   
    futureTask.cancel(true);   
} finally {   
    executor.shutdown();   
}  

FutureTask利用Callable来实现, 把异步逻辑封装成callable, callable会返回future对象
可以通过future接口来操作这个future对象
并且FutureTask还提供Done接口来提供callback, 当isDone为true被自动调用, 默认是空逻辑, 需要override自己的逻辑

实现并发的Memoizer

作为上面内容的很好的总结例子, 实现对高耗费计算的结果的buffer, 并提高并发支持, 直接看代码

public class Memoizer<A, V> implements Computable<A, V> {
    //1.使用ConcurrentHashMap来支持并发
    //2.使用Future, 避免在一个计算没有结束时, 其他并发的重复计算
    //查到Future就知道已经在计算, 只需要等待
    private final ConcurrentMap<A, Future<V>> cache
        = new ConcurrentHashMap<A, Future<V>>(); 
    
    private final Computable<A, V> c;  //计算逻辑对象

    public Memoizer(Computable<A, V> c) { this.c = c; }//构造函数

    public V compute(final A arg) throws InterruptedException {
        while (true) {
            Future<V> f = cache.get(arg); //查询cache
            if (f == null) {
                Callable<V> eval = new Callable<V>() {  //将逻辑c.compute封装成callable
                    public V call() throws InterruptedException {
                        return c.compute(arg);
                    }
                };
                FutureTask<V> ft = new FutureTask<V>(eval); //使用FutureTask
                f = cache.putIfAbsent(arg, ft); //3.使用putIfAbsent原子化复合操作, 防止两个线程都试图put future对象
                if (f == null) { f = ft; ft.run(); }
            }
            try {
                return f.get(); //调用FutureTask.get阻塞的等待结果
            } catch (CancellationException e) {
                cache.remove(arg, f); //防止污染缓存, 删除失败的future对象
            } catch (ExecutionException e) {
                throw launderThrowable(e.getCause());
            }
        }
    }
}

Executor框架

直接使用线程, 你需要自己管理线程的生命周期, 比较麻烦
Java SE5的java.util.concurrent包中的Executor将为你管理Thread对象, 从而简化了并发编程
包括ThreadPool，Executor，Executors，ExecutorService，CompletionService，Future，Callable等类

Executor Interface

Executor接口本身很简单, 表明只能执行Runnable对象

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

简单的接口, 却是强大Framework的基础, 和producer/consumer模式结合, 就可以简单的分离方法的调用和执行
producer传入Runable, consumer就是Executor负责execute

ExecutorService interface

扩展了executor接口, 提供了更多的生命周期管理的接口
ExecutorService interface extends Executor, adding a number of methods for lifecycle management.

public interface ExecutorService extends Executor {
    void shutdown();
    List<Runnable> shutdownNow();
    boolean isShutdown();
    boolean isTerminated();
    boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException;
    //  ... additional convenience methods for task submission
}

Executors Class
Executors是工厂和工具类, 最主要是可以用于创建threadpool
You can create a thread pool by calling one of the static factory methods in Executors:

newFixedThreadPool.
A fixed-size thread pool creates threads as tasks are submitted, up to the maximum pool size, and then attempts to keep the pool size constant (adding new threads if a thread dies due to an unexpected Exception).

newCachedThreadPool.
A cached thread pool has more flexibility to reap idle threads when the current size of the pool exceeds the demand for processing, and to add new threads when demand increases, but places no bounds on the size of the pool.

newSingleThreadExecutor.
A single-threaded executor creates a single worker thread to process tasks, replacing it if it dies unexpectedly. Tasks are guaranteed to be processed sequentially according to the order imposed by the task queue (FIFO, LIFO, priority order).

newScheduledThreadPool.
A fixed-size thread pool that supports delayed and periodic task execution, similar to Timer.

使用Executor来构造web server

class TaskExecutionWebServer {
    private static final int NTHREADS = 100;
    private static final Executor exec   
        = Executors.newFixedThreadPool(NTHREADS); //创建线程池, 返回应该是ExecutorService

    public static void main(String[] args) throws IOException {
        ServerSocket socket = new ServerSocket(80);
        while (true) {
            final Socket connection = socket.accept();
            Runnable task = new Runnable() {  //定义Runnable对象
                public void run() {
                    handleRequest(connection);
                }
            };
            exec.execute(task); //放到线程池中执行
        }
    }
}

CompletionService Interface

其实这个概念很象select, 提交多个future给Executor, future完成的先后不一定, 需要轮询check
CompletionService提供一种封装, 对一组提交给CompletionService的future对象, 只会把已经完成的future对象放到内部的BlockingQueue里面, 所以你就不用轮询, 直接用take取出来的一定是完成的

Future<V> submit(Callable<V> task)
Future<V> take() throws InterruptedException //阻塞等待完成的future
Future<V> poll() //非阻塞,可以加timeout

Class ExecutorCompletionService 接口的实现

void solve(Executor e, Collection<Callable<Result>> solvers)
 throws InterruptedException, ExecutionException {
 CompletionService<Result> ecs = new ExecutorCompletionService<Result>(e); //用Executor初始化
 for (Callable<Result> s : solvers)
     ecs.submit(s); //提交多个Callable对象
 int n = solvers.size();
 for (int i = 0; i < n; ++i) {
     Result r = ecs.take().get(); //take取出最先完成的future对象
     if (r != null)
         use(r);
 }
}

Lock-free and CAS (compare and swap)

Nonblocking algorithms, which use low-level atomic machine instructions such as compare-and-swap instead of locks to ensure data integrity under concurrent access.

Nonblocking algorithms are considerably more complicated to design and implement than lock-based alternatives, but they can offer significant scalability and liveness advantages.

基于底层的乐观锁和volatile的机制

在激烈竞争的情况下, 锁会优于CAS, 但是在正常竞争的情况下, CAS会优于锁

Java中如果使用CAS, (AtomicXxx in java.util.concurrent. atomic)

In Java 5.0, low-level support was added to expose CAS operations on int, long, and object references, and the JVM compiles these into the most efficient means provided by the underlying hardware.

There are twelve atomic variable classes, divided into four groups: scalars, field updaters, arrays, and compound variables. The most commonly used atomic variables are the scalars: AtomicInteger, AtomicLong, AtomicBoolean, and AtomicReference.

并发编程Annotation

主要包括三类：
1、类 Annotation（注解）
就像名字一样，这些注解是针对类的。主有要以下三个：
@Immutable
@ThreadSafe
@NotThreadSafe

@ThreadSafe 是表示这个类是线程安全的。具体是否真安全，那要看实现者怎么实现的了，反正打上这个标签只是表示一下。不线程安全的类打上这个注解也没事儿。
@Immutable 表示，类是不可变的，包含了@ThreadSafe的意思。
      @NotThreadSafe 表示这个类不是线程安全的。如果是线程安全的非要打上这个注解，那也不会报错。

这三个注解，对用户和维护者是有益的，用户可以立即看出来这个类是否是线程安全的，维护者则是可以根据这个注解，重点检查线程安全方面。另外，代码分析工具可能会利用这个注解。

2、域 Annotation（注解）
域注解是对类里面成员变量加的注解。
3、方法 Annotation（注解）
方法注解是对类里面方法加的注解。

域注解和方法注解都是用@GuardedBy( lock )来标识。里面的Lock是告诉维护者：这个状态变量，这个方法被哪个锁保护着。这样可以强烈的提示类的维护者注意这里。

@GuardedBy( lock )有以下几种使用形式：

1、@GuardedBy( "this" ) 受对象内部锁保护
2、@GuardedBy( "fieldName" ) 受 与fieldName引用相关联的锁 保护。
3、@GuardedBy( "ClassName.fieldName" ) 受 一个类的静态field的锁 保存。
4、@GuardedBy( "methodName()" ) 锁对象是 methodName() 方法的返值，受这个锁保护。
5、@GuardedBy( "ClassName.class" ) 受 ClassName类的直接锁对象保护。而不是这个类的某个实例的锁对象。