励精图治---Concurrency---ThreadPoolExecutor最详

               心情不好，不适合写博客。今天开篇。周末完成它。唉我的妈呀。这个世界怎么了？

               结果周末太慵懒，也太忙。还是没写成。今天写。

学习了java的几个类。把其中的东西记录梳理了一下。

类组织结构流程大致是这样的

           


这也没什么看不懂的。就这样子。接着一个类一个类的讲

Executor详解

这是一个接口类，代码很简单。需要其他地方去做具体的实现。

public interface Executor {

    void execute(Runnable command);

}

那么，这个函数是做什么的，这个类的目的是什么？

这个类提供了一个提交执行任务的函数，这个interface旨在解耦任务传递。达到多线程运行的目的。Executor是用来替代new Thread的创建方式。

 范例：线程创建方式是这样的

 new Thread(new RunnableTask())

 Executor创建方式是这样的

 Executor executor = new somekindExecutor();//#这里要看ThreadPoolExecutor的具体类型

 executor.execute(new RunnableTask());

 

Executor的特点

 1. Executor类不要求被执行的runnable是异步的。

 范例：runnable被直接传入

  class DirectExecutor implements Executor {

    public void execute(Runnable r) {

      r.run();

    }

  }}

 

 这里的runnable是可以直接被执行的。你根本不知道runnable是同步还是异步

 
 2. execute中的runnable还能被thread再次封装使用。

  class ThreadPerTaskExecutor implements Executor {

    public void execute(Runnable r) {

      new Thread(r).start();

    }

  }}

 线程是可以不断被分派出来一个新的线程的。

 3. execute可以直接执行一个runnable，但是如果线程不断被执行，也是会引发资源耗尽的。所以可能就有限制。

 什么时候执行，用什么样子的方式执行，状态的存储

  class SerialExecutor implements Executor {

    final Queue<Runnable> tasks = new ArrayDeque<Runnable>();

    final Executor executor;

    Runnable active;

 

    SerialExecutor(Executor executor) {

      this.executor = executor;

    }

 

    public synchronized void execute(final Runnable r) {

      tasks.offer(new Runnable() {

        public void run() {

          try {

            r.run();

          } finally {

            scheduleNext();

          }

        }

      });

      if (active == null) {

        scheduleNext();

      }

    }

    protected synchronized void scheduleNext() {

      if ((active = tasks.poll()) != null) {

        executor.execute(active);

      }

    }

  }}

 这里增加了同步函数scheduleNext，execute也改成了同步的。这里不至于阻塞，应该都是立即返回的。tasks可以用来储存runnable。是队列存储tasks。

 

 4. ExecutorService类extends自Executor类，增加了一些函数。ThreadPoolExecutor提供了可扩展线程池的实现，这个类实现了execute函数。

 从设计模式的角度看，Executor.java提供了工厂方法。

 从内存一致性看，一个runnable对象提交给executor之后，才会执行。那么就意味着runnable会先被读进主存，这就满足了内存可见性。所以，execute这个步骤我认为是线程安全的. 

特别注释happen-before是一种内存可见性的关系。指的是某个操作一定要在另一个操作之前才能保证其内存可见性。。。

ExecutionService详解

先吃饭。今天进展太慢了

日子还得过, 今天继续写。

简述

ExecutorService继承自Executor，扩展出了task关闭以及通过Future可以追踪多个异步task状态的函数

Future是一个可以追踪多个异步task状态的类。

特征1

ExecutorService是可以被关闭的，在其被关闭之间会拒绝接受新task。

ExecutorService的关闭通过两个函数实现：shutdown shutdownNow

shutdown:关闭之前提交的task可以被excute。该函数不会将已提交未执行的task放弃掉。

shutdownNow:会阻止已提交未执行的task，同时停止当前正在执行的task。

在ExecutorService被终止后，就不会有执行着的，等待着的task，也不会有新task被提交。在被终止之后，ExecutorService就可以被jvm回收。

特征2

submit函数扩展自execute函数，区别在于submit函数使用Future包装了task，然后转手给execute。

invokeAny和invokeAll提供一种执行一组tasks的方法

invokeAny，会在任意task完成后就返回结果，

invokeAll，会等待所有的任务完成然后返回一个list

值得注意的是ExecutorCompletionService类，它是BlockingQueue和Executor的组合，已阻塞的方式实现了invokeAny的功能

 

范例1：网络服务的简易框架。

class NetworkService implements Runnable { //#对网络不熟悉的人，注意一下这里的implements

    private final ServerSocket serverSocket;

    private final ExecutorService pool;

 

    public NetworkService(int port, int poolSize)

        throws IOException {

      serverSocket = new ServerSocket(port);

      pool = Executors.newFixedThreadPool(poolSize);

    }

 

    public void run() { // run the service

      try {

        for (;;) {

          pool.execute(new Handler(serverSocket.accept()));

        }

      } catch (IOException ex) {

        pool.shutdown();

      }

    }

  }

 

  class Handler implements Runnable {

    private final Socket socket;

    Handler(Socket socket) { this.socket = socket; }

    public void run() {

      // read and service request on socket

    }

  }

范例2：两步关闭ExecutorService。是相对优化的一段代码

void shutdownAndAwaitTermination(ExecutorService pool) {

    pool.shutdown(); // Disable new tasks from being submitted

    try {

      // Wait a while for existing tasks to terminate

      if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {

        pool.shutdownNow(); // Cancel currently executing tasks

        // Wait a while for tasks to respond to being cancelled

        if (!pool.awaitTermination(60, TimeUnit.SECONDS))

            System.err.println("Pool did not terminate");

      }

    } catch (InterruptedException ie) {

      // (Re-)Cancel if current thread also interrupted

      pool.shutdownNow();

      // Preserve interrupt status

      Thread.currentThread().interrupt();

    }

}}

  
为了保证内存一致性，线程中Runnable对象或者Callable对象提交到ExecutorService之前的操作，一定要在其执行开始之前完成(这可能在另一个线程中)，这是一种happen-before关系。

每个动作执行状态可以通过Future.get依次获取。

AbstractExecutorService详解

简述

AbstractExecutorService是一个抽象类，提供了ExecutorService的默认实现。

此类利用newTaskFor返回的RunnableFuture实现了submit，invokeAny，invokeAll。

值得注意的是newTaskFor 的权限，见源码

protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {

    return new FutureTask<T>(runnable, value);

}

这是说newTaskFor可以被重写。

同时，这里的RunnableFuture用的是java.util.concurrent包中的FutureTask，任何一个实现了RunnableFuture都可以被return。

类似这样

  public class CustomThreadPoolExecutor extends ThreadPoolExecutor {

 

    static class CustomTask<V> implements RunnableFuture<V> {...}

 

    protected <V> RunnableFuture<V> newTaskFor(Callable<V> c) {

        return new CustomTask<V>(c);

    }

    protected <V> RunnableFuture<V> newTaskFor(Runnable r, V v) {

        return new CustomTask<V>(r, v);

    }

    // ... add constructors, etc.

  }}

ThreadPoolExecutor详解

明天再写

简介

ThreadPoolExecutor继承自AbstractExecutorService，一般用Execetors.java中的某一个工厂方法执行任务

线程池解决了两个问题

1. 提高了在执行大数量异步任务时的性能，减少了每个任务的调用开销，提供绑定和管理资源。

2. 每个ThreadPoolExecutor还保存一些基本数据，比如完成任务数

 

线程池的普通使用方式：利用Executors.java中的工厂方法

1. Executors.newCachedThreadPool 

2. Executors.newFixedThreadPool 

3. Executors.newSingleThreadExecutor 

4. Executors.newScheduledThreadPool

5. Executors.newWorkStealingPool

 

对ThreadPoolExecutor的参数解析

corePoolSize和maximumPoolSize

1. ThreadPoolExecutor 将根据 corePoolSize和 maximumPoolSize设置的边界自动调整线程池大小。

2. 当新任务在方法 execute 中提交时，如果运行的线程少于 corePoolSize，则创建新线程来处理请求，即使其他辅助线程是空闲的。

3. 如果运行的线程多于 corePoolSize 而少于 maximumPoolSize，则仅当队列满时才创建新线程。意思就是说如果多于corePoolSize少于maximumPoolSize，那么这个提交到任务就不会被执行。一直到列队满。

注意------这也就存在一定时间任务提交了但是不被执行的情况。这个corePoolSize跟maximumPoolSize要慎重考虑

4. 如果设置的 corePoolSize 和 maximumPoolSize 相同，则创建了固定大小的线程池。

5. 如果将 maximumPoolSize 设置为基本的无界值（如 Integer.MAX_VALUE），则允许池适应任意数量的并发任务。

在大多数情况下，corePoolSize和maximumPoolSize一般是在构造函数中设置，不过也可以使用 setCorePoolSize(int) 和 setMaximumPoolSize(int) 进行动态更改。

启动方式

默认情况下，即使coreThread最初只是在新任务到达时才创建和启动的，

也可以使用方法 prestartCoreThread() 或 prestartAllCoreThreads() 对其进行动态重写。

如果构造带有非空队列的池，则可能希望预先启动线程。

创建新线程

使用 ThreadFactory 创建新线程。如果没有另外说明，则在同一个 ThreadGroup 中一律使用 Executors.defaultThreadFactory() 创建线程，

并且这些线程具有相同的 NORM_PRIORITY 优先级和非守护进程状态。

通过提供不同的 ThreadFactory，可以改变线程的名称、线程组、优先级、守护进程状态，等等。

如果从 newThread 返回 null 时 ThreadFactory 未能创建线程，则执行程序将继续运行，但不能执行任何任务。

keepAliveTime

如果池中当前有多于 corePoolSize 的线程，则这些多出的线程在空闲时间超过 keepAliveTime 时将会终止

这提供了当池处于非活动状态时减少资源消耗的方法。如果池后来变得更为活动，则可以创建新的线程。

也可以使用方法 setKeepAliveTime动态地更改此参数。

使用 Long.MAX_VALUE TimeUnit.NANOSECONDS 的值在关闭前有效地从以前的终止状态禁用空闲线程。

默认情况下，保持活动策略只在有多于 corePoolSizeThreads 的线程时应用。

但是只要 keepAliveTime 值非 0，allowCoreThreadTimeOut(boolean) 方法也可将此超时策略应用于核心线程。

队列

所有 BlockingQueue 都可用于传输和保持提交的任务。可以使用此队列与池大小进行交互：

如果运行的线程少于 corePoolSize，则 Executor 始终首选添加新的线程，而不进行排队。

如果运行的线程等于或多于 corePoolSize，则 Executor 始终首选将请求加入队列，而不添加新的线程。

如果无法将请求加入队列，则创建新的线程，除非创建此线程超出 maximumPoolSize，在这种情况下，任务将被拒绝。

排队有三种通用策略：

无队列(直接转交)。工作队列的默认选项是 SynchronousQueue，它将任务直接提交给线程而不保持它们。

在此，如果不存在可用于立即运行任务的线程，则试图把任务加入队列将失败，因此会构造一个新的线程。

此策略可以避免在处理可能具有内部依赖性的请求集时出现锁。

直接提交通常要求无界 maximumPoolSizes 以避免拒绝新提交的任务。这里必须是无界线程池才能适用。不然就会出现任务提交不上去的情况。

同时这里写着是SynchronousQueue一般情况下都是0，只有在提交接受到传递执行中间，碰巧读取才会有 1.

无界队列

使用无界队列（例如，不具有预定义容量的 LinkedBlockingQueue）将导致在所有 corePoolSize 线程都忙时新任务在队列中等待。

这样，创建的线程就不会超过 corePoolSize。（因此，maximumPoolSize 的值也就无效了。）

当每个任务完全独立于其他任务，即任务执行互不影响时，适合于使用无界队列；例如，在 Web 页服务器中。

这种排队可用于处理瞬态突发请求，当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。

但是这种无界队列只有在处理能力够强的情况下才行，如果被无限制的增加，系统崩溃是迟早的事情。

有界队列

当使用有限的 maximumPoolSizes 时，有界队列（如 ArrayBlockingQueue）有助于防止资源耗尽，但是可能较难调整和控制。

队列大小和最大池大小可能需要相互折衷：使用大型队列和小型池可以最大限度地降低 CPU 使用率、操作系统资源和上下文切换开销，

但是可能导致人工降低吞吐量。如果任务频繁阻塞（例如，如果它们是 I/O 边界），则系统可能为超过您许可的更多线程安排时间。

使用小型队列通常要求较大的池大小，CPU 使用率较高，但是可能遇到不可接受的调度开销，这样也会降低吞吐量。

有界队列能很好的防止资源耗尽的情况发生。配合序列化等手段，可以将超出的任务保存到文件中。又降低了消耗。是比较可取的一种方法。

被拒绝的任务

当 Executor 已经关闭，并且 Executor 将有限边界用于最大线程和工作队列容量，且已经饱和时，

在方法 execute中提交的新任务将被拒绝。在以上两种情况下，

execute 方法都将调用其 RejectedExecutionHandler 的 RejectedExecutionHandler.rejectedExecution 方法。

下面提供了四种预定义的处理程序策略：

ThreadPoolExecutor.AbortPolicy:默认方法-------处理程序遭到拒绝将抛出运行时 RejectedExecutionException。

ThreadPoolExecutor.CallerRunsPolicy-------线程调用运行该任务的 execute 本身。此策略提供简单的反馈控制机制，能够减缓新任务的提交速度。

ThreadPoolExecutor.DiscardPolicy-------直接抛弃被拒绝的任务。

ThreadPoolExecutor.DiscardOldestPolicy-------按照规则抛弃任务，但要主要是否是priority排序，FIFO排序，前者会抛弃优先级最高的，后者抛弃最先提交的。慎用！

定义和使用其他种类的 RejectedExecutionHandler 类也是可能的，但这样做需要非常小心，尤其是当策略仅用于特定容量或排队策略时。

钩子 (hook) 方法

此类提供 protected 可重写的 beforeExecute 和 afterExecute 方法，这两种方法分别在执行每个任务之前和之后调用。

它们可用于操纵执行环境；例如，重新初始化 ThreadLocal、搜集统计信息或添加日志条目。此外，还可以重写方法 terminated() 来执行 Executor 完全终止后需要完成的所有特殊处理。

如果钩子 (hook) 或回调方法抛出异常，则内部辅助线程将依次失败并突然终止。

队列维护

方法 getQueue() 仅仅允许出于监控和调试目的而访问工作队列。其他任何目的都不要用。而且这个应该是线程非安全的。看看可以，修改不行。

remove 和 purge() 这两种方法可用于在取消大量已排队任务时帮助进行存储回收。

终止

程序 AND 不再引用的池没有剩余线程会自动 shutdown。

如果希望确保回收取消引用的池（即使用户忘记调用 shutdown()），则必须安排未使用的线程最终终止：设置适当保持活动时间，

设置corePoolSize成 0 或 设置 allowCoreThreadTimeOut(boolean)。

范例

此类的大多数扩展可以重写一个或多个受保护的钩子 (hook) 方法。例如，下面是一个添加了简单的暂停/恢复功能的子类：

 class PausableThreadPoolExecutor extends ThreadPoolExecutor {

   private boolean isPaused;

   private ReentrantLock pauseLock = new ReentrantLock();

   private Condition unpaused = pauseLock.newCondition();

   public PausableThreadPoolExecutor(...) { super(...); }

 

   protected void beforeExecute(Thread t, Runnable r) {

     super.beforeExecute(t, r);

     pauseLock.lock();

     try {

       while (isPaused) unpaused.await();

     } catch(InterruptedException ie) {

       t.interrupt();

     } finally {

       pauseLock.unlock();

     }

   }

 

   public void pause() {

     pauseLock.lock();

     try {

       isPaused = true;

     } finally {

       pauseLock.unlock();

     }

   }

 

   public void resume() {

     pauseLock.lock();

     try {

       isPaused = false;

       unpaused.signalAll();

     } finally {

       pauseLock.unlock();

     }

   }
 }

----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------