三、Java线程同步增强包

　　相信你已经了解了Java用于同步的3板斧：synchronized/wait/notify，它们的确简单而有效。但是在某些情况下，我们需要更加复杂的同步工具。有些简单的同步工具类，诸如ThreadBarrier，Semaphore，ReadWriteLock等，可以自己编程实现。现在要介绍的是牛人Doug Lea的Concurrent包。这个包专门为实现Java高级并行程序所开发，可以满足我们绝大部分的要求。更令人兴奋的是，这个包公开源代码，可自由下载。且在JDK1.5中该包将作为SDK一部分提供给Java开发人员。

　　Concurrent Package提供了一系列基本的操作接口，包括sync，channel，executor,barrier,callable等。这里将对前三种接口及其部分派生类进行简单的介绍。

　　sync接口：专门负责同步操作，用于替代Java提供的synchronized关键字，以实现更加灵活的代码同步。其类关系图如下：

　　Semaphore：和前面介绍的代码类似，可用于pool类实现资源管理限制。提供了acquire()方法允许在设定时间内尝试锁定信号量，若超时则返回false。

　　Mutex：和Java的synchronized类似，与之不同的是，synchronized的同步段只能限制在一个方法内，而Mutex对象可以作为参数在方法间传递，所以可以把同步代码范围扩大到跨方法甚至跨对象。

　　NullSync：一个比较奇怪的东西，其方法的内部实现都是空的，可能是作者认为如果你在实际中发现某段代码根本可以不用同步，但是又不想过多改动这段代码，那么就可以用NullSync来替代原来的Sync实例。此外，由于NullSync的方法都是synchronized，所以还是保留了“内存壁垒”的特性。

　　ObservableSync：把sync和observer模式结合起来，当sync的方法被调用时，把消息通知给订阅者，可用于同步性能调试。

　　TimeoutSync：可以认为是一个adaptor，其构造函数如下：

public TimeoutSync(Sync sync, long timeout){…}

　　具体上锁的代码靠构造函数传入的sync实例来完成，其自身只负责监测上锁操作是否超时，可与SyncSet合用。

　　Channel接口：代表一种具备同步控制能力的容器，你可以从中存放/读取对象。不同于JDK中的Collection接口，可以把Channel看作是连接对象构造者(Producer)和对象使用者(Consumer)之间的一根管道。如图所示：

　　通过和Sync接口配合，Channel提供了阻塞式的对象存取方法（put/take）以及可设置阻塞等待时间的offer/poll方法。实现Channel接口的类有LinkedQueue，BoundedLinkedQueue，BoundedBuffer，BoundedPriorityQueue，SynchronousChannel，Slot等。

　　使用Channel我们可以很容易的编写具备消息队列功能的代码，示例如下：

Package org.javaresearch.j2seimproved.thread;

Import EDU.oswego.cs.dl.util.concurrent.*;

public class TestChannel ...{
    final Channel msgQ = new LinkedQueue(); //log信息队列 

    public static void main(String[] args) ...{
        TestChannel tc = new TestChannel();
        For(int i = 0;i < 10;i ++)...{
            Try...{
                tc.serve();
                Thread.sleep(1000);
            }catch(InterruptedException ie)...{
            }
        }
    }

    public void serve() throws InterruptedException ...{
        String status = doService();//把doService()返回状态放入Channel，后台logger线程自动读取之 
        msgQ.put(status);
    }

    private String doService() ...{
        // Do service here return "service completed OK! "; 
    }

    public TestChannel() ...{
        // start background thread 
        Runnable logger = new Runnable() ...{
            public void run() ...{
                try ...{
                    for (; ; )
                        System.out.println("Logger: " + msgQ.take());
                } catch (InterruptedException ie) ...{
                }
            }
        };

        new Thread(logger).start();
    }
}

　　Excutor/ThreadFactory接口: 把相关的线程创建/回收/维护/调度等工作封装起来，而让调用者只专心于具体任务的编码工作（即实现Runnable接口），不必显式创建Thread类实例就能异步执行任务。

　　使用Executor还有一个好处，就是实现线程的“轻量级”使用。前面章节曾提到，即使我们实现了Runnable接口，要真正的创建线程，还是得通过new Thread()来完成，在这种情况下，Runnable对象(任务)和Thread对象(线程)是1对1的关系。如果任务多而简单，完全可以给每条线程配备一个任务队列，让Runnable对象(任务)和Executor对象变成n:1的关系。使用了Executor，我们可以把上面两种线程策略都封装到具体的Executor实现中，方便代码的实现和维护。

　　具体的实现有: PooledExecutor，ThreadedExecutor，QueuedExecutor，FJTaskRunnerGroup等。类关系图如下：

　　下面给出一段代码，使用PooledExecutor实现一个简单的多线程服务器：

package org.javaresearch.j2seimproved.thread;

import java.net.*;
import EDU.oswego.cs.dl.util.concurrent.*;

public class TestExecutor ...{

    public static void main(String[] args) ...{
        PooledExecutor pool = new PooledExecutor(new BoundedBuffer(10), 20);
        pool.createThreads(4);
        try ...{
            ServerSocket socket = new ServerSocket(9999);
            for (; ; ) ...{
                final Socket connection = socket.accept();
                pool.execute(new Runnable() ...{
                    public void run() ...{
                        new Handler().process(connection);
                    }
                });
            }
        } catch (Exception e) ...{
        }
        // die 
    }

    static class Handler ...{
        void process(Socket s)...{

        }
    }
}