Java并发编程实战-6结构化并发应用程序_执行任务

Publish: 2018/7/23   

• Java 并发编程实战

在线程中执行任务

当围绕”任务执行”来设计应用程序结构时，第一步就是要找出清晰的任务边界。

大多数服务器应用程序都提供了一种自然的任务边界选择方式：以独立的客户请求为边界。

串行地执行任务

在单个线程中串行地执行各项任务。大多数GUI框架都是通过单一的线程来串行地处理任务。

显示地为任务创建线程

通过为每个请求创建一个新的线程来提供服务，从而实现更高的响应性。

• 任务处理过程从主线程中分离出来，使得主循环能够更快地重新等待下一个到来的连接。
  
  这使得程序在完成前面的请求之前可以接受新的请求，从而提高响应性。
• 任务可以并行处理，从而能同时服务多个请求。程序的吞吐量将得到提高。
• 任务处理代码必须是线程安全的。

无限制创建线程的不足

• 线程生命周期的开销非常高。
• 资源消耗。创建更多的线程反而降低性能。
• 稳定性。在可创建线程的数量上存在一个限制。破坏这些限制，可能抛出OutOfMemoryError异常。

Exector框架

Executor基于生产者-消费者模式。它提供了一种标准的方法将任务的提交过程与执行过程解耦开来，并用Runnable来表示任务。

Executor的实现还提供了对生命周期的支持，以及统计信息收集，应用程序管理机制和性能监视等机制。

示例：基于Executor的Web服务器

    //基于线程池的web服务器
    class TaskExecutionWebServer{
        private static final int NTHREADS = 100;
        private static final Executor exec = Executors.newFixedThreadPool(NTHREADS);

        public static void main(String[] main){
            ServerSocket socket=new ServerSocket(80);
            while (true){
                final Socket connection =socket.accept();
                Runnable task=new Runnable() {
                    @Override
                    public void run() {
                        handleRequest(connection);
                    }
                };
                exec.execute(task);
            }
        }
    }

    //为每个请求启动一个新线程的Executor
    class TaskPerExecutionWebServer implements Executor{

        @Override
        public void execute(@NonNull Runnable command) {
            new Thread(command).start();
        }
    }

    //在调用线程中以同步方式执行所有任务的Executor
    class WithTaskExecutionWebServer implements Executor{

        @Override
        public void execute(@NonNull Runnable command) {
            command.run();    
        }
    }

执行策略

在执行策略中定义了任务执行的”What、Where、When、How”等方面，包括：

• 在什么(What)线程中执行任务？
• 任务按照什么(What)顺序执行(FIFO,LIFO,优先级)?
• 有多少个(How Many)任务能并发执行？
• 在队列中有多少个(How Many)任务在等待执行？
• 如果系统由过载而需要拒绝一个任务，那么应该选择哪一个(Which)任务？另外，如何(How)通知应用程序有任务被拒绝？

每当看到下面这种形式的代码时：

new Thread(runnable).start();

并且你希望获得一种更灵活的执行策略时，请考虑使用Executor来代替Thread。

线程池

指管理一组同构工作线程的资源池。可以通过Executors中的静态工厂方法之一来创建线程池：

• newFixedThreadPool。
  
  长度固定的线程池，每当提交一个任务时就创建一个线程，直到达到线程池的最大数量(如果某个线程由于Exception结束，线程池会补充一个新的线程)
• newCachedThreadPool。
  
  可缓存的线程池，有空闲的线程会使用空闲的线程，没有就增加。
• newSingleThreadExecutor。
  
  一个单线程的Executor。确保依照任务在队列中的顺序来串行执行(例如FIFO、LIFO、优先级)。
• newScheduledThreadPool。
  
  创建一个固定长度的线程池，以延时或定时的方式执行任务。

Executor的生命周期

Executor扩展了ExecutorService接口，添加了用于生命周期管理的方法。

• 运行。
  
• 在初始创建时处于运行状态
• 关闭。
  
  • shutdown方法将执行平缓的关闭过程：不再接受新的任务，同时等待已经提交的任务执行完成(包括还未开始的任务)
  • shutdownNow方法将执行粗暴的关闭过程：尝试取消所有运行中的任务，并且不在启动队列中尚未开始执行的任务。
  • 关闭后提交任务会抛弃任务或抛出RejectedExecutionException。
• 已终止。
  
  • awaitTermination。等待到达终止状态。
  • isTerminated。轮询是否终止。
  • 通常在调用awqitTermination之后会立即调用shutdown,从而产生同步地关闭ExecutorService的效果。

延迟任务与周期任务

使用ScheduledThreadPool代替Timer

• Timer基于绝对时间而ScheduledThreadPool基于相对时间。
• Timer在执行所有定时任务时只会创建一个线程，如果某个任务执行时间过长，将破坏其他TimeTask的定时精确性。
• Timer不捕获异常，抛出异常时，将终止定时线程，产生”线程泄漏”(被调度但尚未执行的TimerTask将不会再执行，新的任务也不能被调度。)。

使用DelayQueue构建自定义调度服务，它实现了BlockingQueue，并为ScheduledThreadPool提供调度功能。

DelayQueue管理着一组Delayed对象。每一个Delayed对象都有一个相应的延迟时间：只有某个元素逾期后，才能从DelayQueue中执行take操作。返回的对象根据它们的延时时间进行排序。

找出可利用的并行性

不明显的任务边界(如服务器应用程序中，在单个客户请求中仍可能纯在可发掘的并行性，例如数据库服务)。

示例：串行的页面渲染器

先绘制文本元素，同时为图像预留出矩形的占位空间，在处理完第一遍文本后，再开始下载图像，并将它们绘制到相应的占位空间中。

携带结果的任务Callable和Future

Runnable不能返回一个值或抛出一个受检查的异常。

• Runnable和Callable描述的都是抽象的计算任务。这些任务通常是有范围的，即都有一个明确的起始点，并且最终会结束。Executor将Runnable或Callable从提交线程安全的发布到最终执行任务的线程。
• Future表示一个任务的生命周期，并提供了相应的方法来判断是否已经完成或取消，以及获取任务的结果和取消任务等。任务的周期只能前进，不能后退。将结果从计算它的线程安全的发布到任何通过get获得它的线程。

在异构任务并行化中存在的局限

洗碗工作,分配工人清洗和烘干：

1. 分配不同类型的任务并不容易。
2. 任务大小(耗时)可能不同。
3. 分解任务时，需要的任务协调开销：不能高于并行性实现的提升。

只有当大量相互独立且同构的任务可以并发进行处理时，才能体现出将程序的工作负载分配到多个任务中带来的真正性能提升。

CompletionService:Executor与BlockingQueue

CompletionService将Executor和BlockingQueue的功能融合再一起。将Callable任务提交给它来执行，然后使用类似于队列操作的take和poll等方法来获得已完成的结果，这些结果完成时将被封装为Future。

ExecutorCompletionService实现了CompletionService，并将计算部分委托给一个Executor。在构造函数中创建一个BlockingQueue来保存计算完成的结果。当计算完成时，调用Future-Task中的done方法。当提交某个任务时，该任务将首先包装为一个QueueingFuture，这是FutureTask的一个子类，然后再改写子类的done方法，并将结果放入BlockingQueue中。

多个ExecutorCompletionService可以共享一个Executor。

示例：使用CompletionService实现页面渲染器

    public class Renderer{
        private final ExecutorService executor;

        Renderer(ExecutorService executor){this.executor=executor}

        void renderPage(CharSequence source){
            List<ImageInfo> info=scanForImageInfo(source);
            CompletionService<ImageData> completionService
                    =new ExecutorCompletionService<ImageData>(executor);
            for (final ImageInfo imageInfo : info) {
                //为每幅图的下载创建一个独立任务，并在线程池中执行。减少下载所有图像的总时间。
                completionService.submit(new Callable<ImageData>() {
                    @Override
                    public ImageData call() throws Exception {
                        return imageInfo.downloadImage();
                    }
                });
            }

            renderText(source);

            try {
                for (int i = 0; i < info.size(); i++) {
                    //通过从CompletionService中获取结果以及使图片下载完成后立即显示出来，提高响应。
                    Future<ImageData> f=completionService.take();
                    ImageData imageData=f.get();
                    renderImage(imageData);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
    }

为任务设置时限

如果某个任务无法在指定时间内完成，那么将不在需要它的结果。

在支持时间限制的Future.get中支持这种需求：当结果可用时，它将立即返回，如果在指定时限内没有计算出结果，那么将抛出TimeoutExeception，并通过Future来取消任务释放资源。