Java 并发编程框架（一）

Java 并发编程框架（一）

在Java1.5之前，编写多线程并非易事，那么编写多线程为啥不想想象的那么简单，为什么需要线程池？先来回答这个问题。

Why Thread Pool

在Java中，如果每当一个请求到达就创建一个新线程，开销是相当大的。在实际使用中，每个请求创建新线程的服务器在创建和销毁线程上花费的时间和消耗的系统资源，甚至可能要比花在实际处理实际的用户请求的时间和资源要多的多。除了创建和销毁线程的开销之外，活动的线程也需要消耗系统资源。如果在一个JVM中创建太多的线程，可能会导致系统由于过度消耗内存或者“切换过度”而导致系统资源不足。为了防止资源不足，服务器应用程序需要一些办法来限制任何给定时刻处理的请求数目，尽可能减少创建和销毁线程的次数，特别是一些资源耗费比较大的线程的创建和销毁，尽量利用已有对象来进行服务，这就是“池化资源”技术产生的原因。

线程池主要用来解决线程生命周期开销问题和资源不足问题，通过对多个任务重用线程，线程创建的开销被分摊到多个任务上了，而且由于在请求到达时线程已经存在，所以消除了创建所带来的延迟。这样，就可以立即请求服务，使应用程序响应更快。另外，通过适当的调整线程池中的线程数据可以防止出现资源不足的情况。

Java1.5推出线程池Executor框架以后，编写多线程变得简单易行，既然好用，那么很值得对Executor框架做个简单的了解，在了解之前，需要先了解一下Thread基本概念

Thread

我们知道生成一个子线程有两种方法，一种是直接生成一个匿名内部类，重写run方法，另外一种是新建一个runable类实现runable接口，然后作为参数生成Thread实例

方式1：

[code]new Thread()
    {
        @Override
        public void run()
        {
            super.run();

            //TODO
        }
    }.start();

方式2：

[code]private class Worker implements  Runnable
{
    @Override
    public void run()
    {
        //TODO
    }
}

new Thread(new Worker()).start();

Tips： 方式2 灵活性更高，因为Worker类还可以继承其他父类或者实现其他接口，而方式1除了可以继承Thread外不可以再继承别的类，虽然可以实现其他接口，灵活性毕竟不如方式2

Note： 即使方式2比较灵活，但是如果我们想重用Thread,很容易想到Java1.5为我们提供的Executor框架，所以接下来一起来看看这个Executor框架长什么样？

Executor框架

在了解这个框架之前，首先看看这个框架有哪些相关的主要类

Executor

ExecutorService

Executors

Future

RunnableFuture

FutureTask

Callable

CompletionService

ExecutorCompletionService

然后分别对上面列出来的主要类做个简单的代码了解

Executor

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

这是个顶层的接口，只有一个execute方法，接收Runnable实例作为参数。

ExecutorService

[code]public interface ExecutorService extends Executor {
    void shutdown();
    boolean isShutdown();
    <T> Future<T> submit(Callable<T> task);
    Future<?> submit(Runnable task);
    ...
}

这是个顶层Executor接口的子接口，这个子接口扩张了顶层Executor接口，这些接口明显增强了开发者对Thread的控制能力。

Executors

[code]public class Executors {
    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());
    }

    public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
        return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>(),threadFactory);
    }

    public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()));
    }

    public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>());
    }
    ...
}

这个类定义成ExecutorUtil更合适，因为它是个彻头彻尾的工具类，通过Executors类中的静态方法可以获得十分丰富的线程池，对于这些线程池具体有什么区别，后面会逐一介绍

Future

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

这是个顶层接口，这个接口明显的弥补了Thread的缺陷，因为Thread并不能对自身进行控制，更不能获得线程执行的结果（显示的结果，当然如果想进一步将这个Result返回给其他线程，只需要通过线程间通信将Result传递过去亦可，但是过程稍显冗余）。

RunnableFuture

[code]public interface RunnableFuture<V> extends Runnable, Future<V> {
    void run();
}

这是个Future和Runnable接口的子接口，这个接口只是将Runnable接口的唯一run方法缩小的可见范围，为啥要缩小可见范围？尚不清楚，那么如果这么改写这个类，可以行得通否？。

[code]public interface RunnableFuture<V> extends Runnable, Future<V> {
}

FutureTask

[code]public class FutureTask<V> implements RunnableFuture<V> {
    public FutureTask(Callable<V> callable) {
            if (callable == null)
                throw new NullPointerException();
            this.callable = callable;
            this.state = NEW;       // ensure visibility of callable
        }

    public FutureTask(Runnable runnable, V result) {
        this.callable = Executors.callable(runnable, result);
        this.state = NEW;       // ensure visibility of callable
    }

    public V get() throws InterruptedException, ExecutionException {
        int s = state;
        if (s <= COMPLETING)
            s = awaitDone(false, 0L);
        return report(s);
    }

    public boolean isCancelled() {
        return state >= CANCELLED;
    }

    public boolean isDone() {
        return state != NEW;
    }
    ...
}

这个类是RunnableFuture接口实现类，从构造器可以看出，接收Callable实例或者Runnable并带有V泛型的Result实例。

Callable

[code]public interface Executor {
    V call() throws Exception;
}

这是个顶层的接口，只有一个call方法，并且返回一个*V泛型作为Result，这个是所有可控制，可取得线程执行结果的顶层类，一般作为参数生成线程实例或者传给线程池。

CompletionService

[code]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;
}

这是个顶层的接口，那么这个接口有神马用呢？这里简单的给一点提示。

Tips: 当向Executor提交批处理任务时，并且希望在它们完成后获得结果，如果用FutureTask，你可以循环获取task，并用future.get()去获取结果，但是如果这个task没有完成，你就得阻塞在这里，这个实效性不高，其实在很多场合，其实你拿第一个任务结果时，此时结果并没有生成并阻塞，其实在阻塞在第一个任务时，第二个task的任务已经早就完成了，显然这种情况用future task不合适的，效率也不高。

Note: 那么自己维护list和CompletionService的区别？

1: 从list中遍历的每个Future对象并不一定处于完成状态，这时调用get()方法就会被阻塞住，如果系统是设计成每个线程完成后就能根据其结果继续做后面的事，这样对于处于list后面的但是先完成的线程就会增加了额外的等待时间。

2: 而CompletionService的实现是维护一个保存Future对象的BlockingQueue。只有当这个Future对象状态是结束的时候，才会加入到这个Queue中，take()方法其实就是Producer-Consumer中的Consumer。它会从Queue中取出Future对象，如果Queue是空的，就会阻塞在那里，直到有完成的Future对象加入到Queue中。

ExecutorCompletionService

[code]public class ExecutorCompletionService<V> implements CompletionService<V> {
    private class QueueingFuture extends FutureTask<Void> {
        QueueingFuture(RunnableFuture<V> task) {
            super(task, null);
            this.task = task;
        }
        protected void done() { completionQueue.add(task); }
        private final Future<V> task;
    }

    public Future<V> submit(Callable<V> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<V> f = newTaskFor(task);
        executor.execute(new QueueingFuture(f));
        return f;
    }
    ...
}

这是类是CompletionService实现类，这里需要特别注意的是completionQueue这个队列的维护。

Example1

Case.Java 测试用例

[code]public class Case
{
    public static void main(String[] args)
    {
        CommonExecutor commonExecutor = new CommonExecutor();

        commonExecutor.submit(FutureCase.FUTURETASK_THREAD);
    }

    public static enum FutureCase
    {
        //分别是三种方式，通过Future和Executor线程池
        //FutureTask和Executor线程池
        //FutureTask和Thread
        FUTURE_EXECUTOR, FUTURETASK_EXECUTOR,  FUTURETASK_THREAD;
    }

}

CommonExecutor.java 业务逻辑类

[code]public class CommonExecutor
{
    public static class Worker implements Callable<Integer>
    {
        private int mNumber;

        Worker(int number)
        {
            mNumber = number;
        }

        @Override
        public Integer call() throws Exception
        {
            try
            {
                Thread.sleep(new Random().nextInt(2000));
                System.out.println("Current Thread is: " + Thread.currentThread().getName() + " " + mNumber);
            }
            catch (Exception e)
            {
                e.printStackTrace();
            }
            return mNumber;
        }
    }

    public void submit(FutureCase futureCase)
    {
        ExecutorService executor = Executors.newCachedThreadPool();

        Worker worker = new Worker(1);

        switch (futureCase)
        {
        case FUTURE_EXECUTOR:

            Future<Integer> future = executor.submit(worker);

            executor.shutdown();

            System.out.println("Wait......");

            try
            {
                System.out.println("Result： " + future.get());
            }
            catch (InterruptedException e)
            {
                e.printStackTrace();
            }
            catch (ExecutionException e)
            {
                e.printStackTrace();
            }

            break;

        case FUTURETASK_THREAD:

            FutureTask<Integer> futureTask2 = new FutureTask<Integer>(worker);

            new Thread(futureTask2).start();

            System.out.println("Wait......");

            try
            {
                System.out.println("Result： " + futureTask2.get());
            }
            catch (InterruptedException e)
            {
                e.printStackTrace();
            }
            catch (ExecutionException e)
            {
                e.printStackTrace();
            }

            break;

        case FUTURETASK_EXECUTOR:

            FutureTask<Integer> futureTask = new FutureTask<Integer>(worker);

            executor.submit(futureTask);

            executor.shutdown();

            System.out.println("Wait......");

            try
            {
                System.out.println("Result： " + futureTask.get());
            }
            catch (InterruptedException e)
            {
                e.printStackTrace();
            }
            catch (ExecutionException e)
            {
                e.printStackTrace();
            }
            break;

        default:
            break;
        }
    }
}

运行结果

[code]Wait......
Current Thread is: Thread-0 1
Result： 1

Example2

我们将上面的FutureCase再增加一种情况，专门测试CompletionService

Case.java 测试用例类

[code]public static enum FutureCase
    {
        FUTURETASK_EXECUTOR, FUTURE_EXECUTOR, FUTURETASK_THREAD, COMPLETIONSERVICE 
    }

CommonExecutor.java 业务逻辑类

[code]case COMPLETIONSERVICE:

            ExecutorService pool = Executors.newFixedThreadPool(5);

            CompletionService<Integer> completionService = new ExecutorCompletionService<Integer>(pool);

            for (int i = 0; i < 5; i++)
            {
                completionService.submit(new Worker(i));
            }

            for (int i = 0; i < 5; i++)
            {
                try
                {
                    System.out.println("Result： " + completionService.take().get());
                }
                catch (InterruptedException e)
                {
                    e.printStackTrace();
                }
                catch (ExecutionException e)
                {
                    e.printStackTrace();
                }
            }

            break;

运行结果

[code]Current Thread is: pool-2-thread-2 1
Result： 1
Current Thread is: pool-2-thread-4 3
Result： 3
Current Thread is: pool-2-thread-5 4
Result： 4
Current Thread is: pool-2-thread-1 0
Result： 0
Current Thread is: pool-2-thread-3 2
Result： 2