多线程摘录 008

* !!! synchronized使得在等待锁定的线程无法被中断, 也无法因为超时而返回. 这是JDK5重新引入ReentrantLock的主要原因

* ReentrantLock的典型使用风格
Lock lock = new ReentrantLock();
...
lock.lock();
try {
    // update object state
    // catch exceptions and restore invariants if necessary
} finally {
    lock.unlock();
}

* 下面是一段利用ReentrantLock.tryLock()来尝试锁定多个资源的代码, 只要其中有一个资源没有锁定, 就不断尝试下一次去获取锁定. 虽然获得了更多的控制如超时, 重新尝试甚至加入异常处理, 然而代码却比使用synchronized来的复杂
public boolean transferMoney(Account fromAcct,
                             Account toAcct,
                             DollarAmount amount,
                             long timeout,
                             TimeUnit unit)
        throws InsufficientFundsException, InterruptedException {
    long fixedDelay = getFixedDelayComponentNanos(timeout, unit);
    long randMod = getRandomDelayModulusNanos(timeout, unit);
    long stopTime = System.nanoTime() + unit.toNanos(timeout);

    while (true) {
        if (fromAcct.lock.tryLock()) {
        //尝试锁定, 如果成功就一定进入, 并且会执行finally块解锁
            try {
                if (toAcct.lock.tryLock()) {
                    try {
                        if (fromAcct.getBalance().compareTo(amount)
                                < 0)
                            throw new InsufficientFundsException();
                        else {
                            fromAcct.debit(amount);
                            toAcct.credit(amount);
                            return true;
                        }
                    } finally {
                        toAcct.lock.unlock();
                    }
                 }
             } finally {
                 fromAcct.lock.unlock();
             }
         }
         if (System.nanoTime() < stopTime)
             return false;
         NANOSECONDS.sleep(fixedDelay + rnd.nextLong() % randMod);
     }
}

* 仍然回到一个我们初始的问题:
既然是锁定, 那么我们锁定的对象是什么? 因为使用了ReentrantLock, 没有办法显式去指定被锁定对象, 而是对ReentrantLock内部的某个对象锁定, 不像synchronized(xxx)来指定获取对xxx的锁定, 这个时候, 就要求ReentrantLock是被多个线程共享的, 举一个很简单的例子, 朋友请吃饭, 却只有一个锅汤和一个勺子, 那么这N个线程都需要共享这个勺子:

public class Lunch {
    private ReentrantLock scoop = new ReentrantLock();

    public boolean dip() {
        try {
            scoop.lockInterruptibly();
        } catch (InterruptedException e) {
            Logger.log("someone call me for better food ~~~ ");
            return false;
        }

        Logger.log("hah, I got the scoop");
        try {
            // suppose we need 5s to dip the soup
            try {
                Thread.sleep(5000);
            } catch (InterruptedException i) {
                Logger.log("someone rob my scoop, 55~~~ ");
                return false;
            }
            Logger.log("I got delicious food ");
        } finally {
            scoop.unlock();
        }
        return true;
    }
}

public class Buddy extends Thread {
    private final Lunch lunch;
    public Buddy(Lunch lunch, String name) {
        this.lunch = lunch;
        this.setName(name);
    }
    public void run() {
        while (!lunch.dip()) {
            Logger.log("I will wait for a while to dip");
            try {
                Thread.sleep(100);
            } catch (InterruptedException ignore) {}
        }
    }
   
}


public class Party {
    public static void main(String[] args) throws Exception {
        Lunch lunch = new Lunch();
       
        /*new threads MUST share the same lunch instance*/
        Buddy titi= new Buddy(lunch, "titi");
        Buddy michael = new Buddy(lunch, "michael");
        Buddy ting= new Buddy(lunch, "ting");

        titi.start();
        Thread.sleep(100);
        michael.start();
        ting.start();
       
        Thread.sleep(1000);

        // why still hanging? rob him
        titi.interrupt();

        // ask michael to other food
        michael.interrupt();
    }
}

* 锁的公平性
所谓公平性, 就是按照线程访问共享资源的到达顺序来先后访问.如果任由线程自由竞争, 就是不公平. 然而, 公平性的意义可能不大, 而且浪费了太大的代价来维护这个顺序(线程调度的代价). ReentrantLock默认采用的是unfair策略, 也就是由线程自由竞争. 当然也允许使用设置公平策略.

有一种情况可以解释为什么不公平的锁性能更好. 例如, 线程A锁定了资源X, 线程B在等候, 当A完成自己的工作后, 通知B, 然后B被唤醒, 但是如果在A释放锁定情况下, 线程C刚刚到来, 发现可以锁定, 并且在B被唤醒的期间完成了工作, 释放锁定, 最后B再进行操作. 这是一个双赢的结果, 自然吞吐量就上去了. 如果采用公平策略, 那么C必须要排队在B之后.

   锁定 工作   释放
A: |~~~~~~~~~~~~>
   等待         唤醒            锁定 工作       释放
B: -------------|...............|~~~~~~~~~~~~~~~~>
                到达,锁定    释放
C:              |~~~~~~~~~~~~>


* 悲观锁synchronized和ReentrantLock的比较
1) synchronized更简单, 而且一直被使用, 至少编译器保证不会语法出错
2) ReentrantLock是一个危险的工具, 最常见的是忘记使用try-finally去执行unlock, 这样就为并发处理埋下隐患
3) ReentrantLock提供了更高级的功能: 超时, 响应中断, 公平性, 非阻塞锁定, 但是如果不需要这些特性, 为什么不使用synchronized?
4) 在JDK5的环境下, synchronized可以被JVM检测到死锁, 这些信息能被反映在线程的dump日志里面, 但JVM对ReentrantLock一无所知
5) 应该只在synchronized造成非常大的性能瓶颈的情况下才采用ReentrantLock

* ReadWriteLock
ReentrantLock是一种标准的互斥体(mutex)的实现, 但是对于"read-most"模式的共享资源, 互斥的方式极大的限制了性能, 这也是为什么有ReadWriteLock的原因

ReadWriteLock的机制很简单, 允许多个线程以读到方式访问, 但只允许一个线程以写的方式访问. 当要写入时, 一直等待到所有读的线程退出; 如果线程需要读取而又发现有线程中写入, 则一直等待写线程退出.

public interface ReadWriteLock {
    Lock readLock();
    Lock writeLock();
}

ReadWriteLock是一个接口, 显然, 它允许多种实现. 那么, 实现类可以选择哪些策略呢?
1) 读优先还是写优先? 比如当'写线程'释放了锁定, 如果在此前已经有读/写线程排队等待, 那么下一步应该读线程优先获得锁定呢, 还是写线程?
2) 是否允许临时读线程"闯入"? 比如当线程释放了锁定, 如果在此前已经有写线程排队等待, 那么这个新来的读线程是否可以跳过写线程直接去锁定还是排在写线程后面?
3) 可重入性
4) 升级降级. 一个正在写的线程能否降级为"读"锁定, 或者反过来升级.

ReentrantReadWriteLock:
如果设置了公平性: 对于1), 会倾向于分配锁定给等待时间最长的线程, 不过是读还是写. 对于2), 读线程会排到写线程后; 如果是自由竞争, 任何访问顺序都不保证; 允许降级, 不允许升级

* 代码: 用ReentrantReadWriteLock来保证HashMap的并发访问
public class ReadWriteMap<K,V> {
    private final Map<K,V> map;
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    private final Lock r = lock.readLock();
    private final Lock w = lock.writeLock();

    public ReadWriteMap(Map<K,V> map) {
        this.map = map;
    }

    public V put(K key, V value) {
        w.lock();
        try {
            return map.put(key, value);
        } finally {
            w.unlock();
        }
    }
    // Do the same for remove(), putAll(), clear()

    public V get(Object key) {
        r.lock();
        try {
            return map.get(key);
        } finally {
            r.unlock();
        }
    }
    // Do the same for other read-only Map methods
}
只是演示如何使用ReentrantReadWriteLock. 实际上, ConcurrentHashMap的性能以及足够好.

测试表明在中等并发以以上的情况下, ReentrantReadWriteLock的吞吐量是ReentrantLock的3倍

* 和状态相关的线程设计
很多实际情况中, 要执行某些操作, 必须先满足某些前提条件. 如果前提条件不满足, 那么有两种处理方法, 一是直接返回失败的结果, 一是等待, 知道满足为止. JDk5提供了Condition类, 并且可以和Lock相关.

我们看一下使用"等待前提条件满足"策略的基本思路

获取对前提条件对象的锁定        //当然,是用来检测到
while(前提条件不满足) {
    - 释放锁定                  //如果不释放,就无法被其他线程修改
    - 等待                      //自然是等待其他线程的通知了
                                //可以有多种方式, 比如超时, 定时
    - 某些异常处理              //比如, 当前线程被中断了
    - 重新获取锁定              //
}

在JDK5里面, 可以这样做
Condition c = ReentrantLock.newCondition();
ReentrantLock.lock();
while (!isPreconditionSatisfied) {
    c.await();                  //这一行代码搞定了上面的所有要求
}
try{ ... } finally{ReentrantLock.unlock();}

这种代码行为是否似曾相识? 很像BlockingQueue吧...
再想想Latch, 是不是也有这种判断? 但Latch的不同在于, 一点状态被修改, 就在不能回复

考虑一下如果我们使用直接返回失败结果的方式, 弊端之一, 客户端必须不断尝试去重新测试前提条件是否满足, 这样会很浪费CPU, 但是我们可以让线程休眠啊? 这是弊端之二, 因为线程可能睡过头, 结果就是响应时间长了. 当然这个不算是什么太大的问题, 除非这个客户端被连续使用N次...

任何java对象都能作为一个锁, 也能作为一个"条件队列", 即让其他线程都进入"队列"等待通知, 来判断前提条件是否成立

Object.wait atomically releases the lock and asks the OS to suspend the current thread, allowing other threads to acquire the lock and therefore modify the object state. Upon waking, it reacquires the lock before returning
很简单吧, Object.wait方法就实现了上面的几点前提条件的要求

直接用java Object的API来实现, 当然了, 实际当中可能还需要更实用的方法, 比如包括可以超时的put/take方法
public class BoundedBuffer<V> extends BaseBoundedBuffer<V> {
    ...
    public synchronized void put(V v) throws InterruptedException {
        while (isFull())
            wait();
        doPut(v);
        notifyAll();
    }

    public synchronized V take() throws InterruptedException {
        while (isEmpty())
            wait();
        V v = doTake();
        notifyAll();
        return v;
    }
}

或者, 更进一步的小优化
public static final Object lock = new Object();
void stateDependentMethod() throws InterruptedException {
    // condition predicate must be guarded by lock
    synchronized(lock) {
        while (!conditionPredicate())
            lock.wait();
        // object is now in desired state
    }
}

但是, 如果忘记测试前提条件或者测试有误, 会怎样? 可能会导致某些线程T一直等待下去!!! 因为如果再没有其他线程修改状态而发出通知的话, 这个线程T就挂了, 它需要额外的一个通知. 这种现象被称为"消失的信号"(Missed Signal), 与死锁类似. 当我们在代码里面使用 object.notity 而不是object.notifyAll的时候, 导致这种错误的几率非常大, 因为有多个线程都在等待, 而object.notity只是通知JVM唤醒了某一个线程, 除非是仅有一个线程中等待

当然, notifyAll的开销比notify稍微大一些. 也可以对notifyAll进行一点控制, 比如
public synchronized void put(V v) throws InterruptedException {
    while (isFull())
        wait();
    boolean wasEmpty = isEmpty();
    doPut(v);
    //留意这个判断, 如果本来队列就有元素, 说明没有线程在take, 即没有
    //线程中等待, 也就无需通知了
    if (wasEmpty)  
        notifyAll();
}

一个例子, 用来控制门的开和关, 有一些需要思考的地方
public class ThreadGate {
    // CONDITION-PREDICATE: opened-since(n) (isOpen || generation>n)
    @GuardedBy("this") private boolean isOpen;
    @GuardedBy("this") private int generation;

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

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

    // BLOCKS-UNTIL: opened-since(generation on entry)
    public synchronized void await() throws InterruptedException {
        int arrivalGeneration = generation;
        while (!isOpen
            && arrivalGeneration == generation) {
            wait();
        }
    }
}

这个arrivalGeneration == generation判断很令人意外, 但却是必须的, 说明了并发的复杂性. 考虑下面的情景:
1) 当前门处于关闭状态, isOpen=false, generation=0
2) 当前线程A访问的时候, 获取了锁定, 当然!isOpen
&& arrivalGeneration == generation (为0), 两个条件都是成立的, A睡觉去了, 释放了锁定
3) 线程B来了, 取得锁定, 并且开了门: ThreadGate.open(), 这时generation=1, 然后唤醒一堆线程, 包括A
4) 在A被唤醒还没有获得锁定之前, 线程C快人一步, 并做了下面的事情: 先获取锁定, 然后关门, 那么 isOpen = false
5) 姗姗来迟的A终于获取到锁定, 发现isOpen=false, 但generation居然是1, 而自己手上的arrivalGeneration为0, 才明白不对劲, 不能继续睡了

即使A发现不对, 但isOpen仍旧是false, 是否应该继续wait? 问题出在哪里?
问题出在close(), 它修改了isOpen的值, 却没有通知其他线程

* 如果状态相关的并发类被继承?
A state-dependent class should either fully expose (and document) its waiting and notification protocols to subclasses, or prevent subclasses from participating in them at all

* "强制的"唤醒线程
AbstractQueuedSynchronizer, upon which most of the state-dependent classes in java.util.concurrent are built (see Section 14.4), exploits the concept of exit protocol. Rather than letting synchronizer classes perform their own notification, it instead requires synchronizer methods to return a value indicating whether its action might have unblocked one or more waiting threads. This explicit API requirement makes it harder to "forget" to notify on some state transitions.
大意是java.util.concurrent包的大部分类都是基于AbstractQueuedSynchronizer搭建的, 它要求这些并发类返回一个结果来表明需要唤醒一个或多个线程.

* 悲观条件队列(Intrinsic condition queues )的一个缺点就是, 悲观锁只能有一个条件队列, 这让一些不相关的线程也阻塞了. Lock和Condition 就是为解决这些问题设计的. 一个Lock可以有多个相关的Condition, 不同的线程可以分别在不同的Condition上等待, 知道满足为止. 当然还有其他的高级功能

Condition offers a richer feature set than intrinsic condition queues: multiple wait sets per lock, interruptible and uninterruptible condition waits, deadline-based waiting, and a choice of fair or nonfair queueing

public interface Condition {
    void await() throws InterruptedException;
    boolean await(long time, TimeUnit unit)
            throws InterruptedException;
    long awaitNanos(long nanosTimeout) throws InterruptedException;
    void awaitUninterruptibly();
    boolean awaitUntil(Date deadline) throws InterruptedException;

    void signal();
    void signalAll();
}

注意, Object的wait, notify, notifyAll 在Condition接口的等价方法为await, signal, signalAll, 但因为Condition同时也是Object的子类, 也有wait, notify, notifyAll, 不要用错了

下面的例子网上到处有
public class ConditionBoundedBuffer<T> {
    protected final Lock lock = new ReentrantLock();
    // CONDITION PREDICATE: notFull (count < items.length)
    private final Condition notFull    = lock.newCondition();
    // CONDITION PREDICATE: notEmpty (count > 0)
    private final Condition notEmpty = lock.newCondition();
    @GuardedBy("lock")
    private final T[] items = (T[]) new Object[BUFFER_SIZE];
    @GuardedBy("lock") private int tail, head, count;

    // BLOCKS-UNTIL: notFull
    public void put(T x) throws InterruptedException {
        lock.lock();
        try {
            while (count == items.length)
                notFull.await();
            items[tail] = x;
            if (++tail == items.length)
                tail = 0;
            ++count;
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    // BLOCKS-UNTIL: notEmpty
    public T take() throws InterruptedException {
        lock.lock();
        try {
            while (count == 0)
                notEmpty.await();
            T x = items[head];
            items[head] = null;
            if (++head == items.length)
                head = 0;
            --count;
            notFull.signal();
            return x;
        } finally {
            lock.unlock();
        }
    }
}

当然, Condition也不是Object.wait,notify,notifyAll的替代物, 按需使用嘛

* 剖析同步器
ReentrantLock 与 Semaphore 有许多相同点. Both classes act as a "gate", allowing only a limited number of threads to pass at a time; threads arrive at the gate and are allowed through (lock or acquire returns successfully), are made to wait (lock or acquire blocks), or are turned away (tryLock or tryAcquire returns false, indicating that the lock or permit did not become available in the time allowed). Further, both allow interruptible, uninterruptible, and timed acquisition attempts, and both allow a choice of fair or nonfair queueing of waiting threads.

ReentrantLock,Semaphore,CountDownLatch, ReentrantReadWriteLock, SynchronousQueue,FutureTask等都是基于AbstractQueuedSynchronizer 构建
转自：http://hi.baidu.com/iwishyou2/blog/item/5b6fb3ef88d51fe7ce1b3eff.html