多线程摘录 003

同步器
A synchronizer is any object that coordinates the control flow of threads based on its state. Blocking queues can act as synchronizers; other types of synchronizers include semaphores, barriers, and latches

Latch
Latch是门闩的意思, 一旦关闭(打开), 就保持这个状态, 一直不变.
用脚指头也能想到用来干什么了: 保证某些动作只执行一次! 在应用上就广泛一些, 比如初始化某个对象, 在有依赖关系的任务流里面控制执行顺序等

CoutDownLatch是一种Latch实现, 允许多个线程等待一组事件的发生.
下面这一小段代码是为了统计一个任务N次运行的平均时间. 不能采用new Thread().start()的方式, 那样的话, 早建立的线程就先运行完毕, 统计不精确.

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.countDown();
                        }
                    } catch (InterruptedException ignored) { }
                }
            };
            t.start();
        }

        long start = System.nanoTime();
        startGate.countDown();
        endGate.await();
        long end = System.nanoTime();
        return end-start;
    }
}

FutureTask
FutureTask的行为也像是Latch, 一旦任务完成返回了Future的结果, 就不再改变.
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 {
            return future.get();
        } catch (ExecutionException e) {
            ...
        }
    }
}
当其他方法调用Preloader.get()的时候, 如果任务未完成, 会阻塞, 否则直接返回结果, 但是结果不再改变

Semaphore
简而言之, Semaphore控制许可证的数量. 当可用许可证的数量为0, acquire()就阻塞. release()可以把一个许可还给Semaphore. 如果许可证的数量为1, 其实就成了一个互斥体(mutex), 和synchronized的作用相同.

Barrier
The key difference is that with a barrier, all the threads must come together at a barrier point at the same time in order to proceed. Latches are for waiting for events; barriers are for waiting for other threads.
其实Barrier和Latch是很相似的...   (TODO: 这里说明的差别很模糊, 怎么具体来比较呢?)

例子可以参考JDK5的API手册. CyclicBarrier是Barrier的一个实现类, 和Latch不同的是, 它可以被reset, 从而重新使用, Latch则是到达最终状态后不再改变

小技巧:
如果想按照系统的CPU数量来分配线程数, 怎么做? 很简单, Runtime类能告诉我们:
Runtime.getRuntime().availableProcessors()

一个并发的Cache
这个Cache的设计很有意思, 考虑了很多并发的因素. Cache用简单的Map方式来实现.
1) 使用ConcurrentHashMap来存放计算结果
2) 最有趣的一点, ConcurrentHashMap里存放的内容是FutureTask. (知道为什么? 这相当于放一个Latch)
3) 使用了FutureTask来调度计算, 这样多个线程可以分别进行针对不同的参数的计算
4) 基于FutureTask的特性, 先把FutureTask放进了cache, 再对FutureTask计算
5) 用来一个 while(true) 无限循环...
    ???? 这是为什么? 因为如果发现已经有一个线程中进行相同的计算的话, 就直接等待计算结果. 但是等待过程中, 如果FutureTask被中断了呢? 抓住异常, 再等...
6) 使用putIfAbsent操作, 避免出现两个线程都进行相同的计算

public class Memoizer<A, V> implements Computable<A, V> {
    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);
            if (f == null) {
                Callable<V> eval = new Callable<V>() {
                    public V call() throws InterruptedException {
                        return c.compute(arg);
                    }
                };
                FutureTask<V> ft = new FutureTask<V>(eval);
                f = cache.putIfAbsent(arg, ft); //先把FutureTask放进去再说
                if (f == null) { f = ft; ft.run(); } //开始计算
            }
            try {
                return f.get();
            } catch (CancellationException e) {
                cache.remove(arg, f);
            } catch (ExecutionException e) {
                throw launderThrowable(e.getCause());
            }
        }
    }
}

想象一下自己以前写的(呵呵, 简单易懂, 绝对能工作, 但并发性能就...)
public synchronized Object get(Object k) {
    Object obj = cache.get(k);
    if (obj==null) {
        obj = loadFromDB(k);
        cache.put(k, obj);
    }
    return obj;
}

转自：http://hi.baidu.com/iwishyou2/blog/item/2df183ece3062135279791e0.html