Java并发之线程池Executor框架的深入理解

public class ScheduledThreadPoolTest {
  public static void main(String[] args) {
    ScheduledExecutorService ses = Executors.newScheduledThreadPool(10);
    ses.scheduleWithFixedDelay(new Runnable() {
      
      @Override
      public void run() {
        try {
          Thread.sleep(1000);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
        System.out.println(new Date());
      }
    }, 1000, 2000, TimeUnit.MILLISECONDS);
  }
  
}

public interface Executor{
  void executor(Runnable command);
}

Future<?> submit(Runnable task)

<T> Future<T> submit(Callable<T> task)

protected void beforeExecute(Thread t, Runnable r) { }

protected void afterExecute(Runnable r, Throwable t) { }

protected void terminated() { }

public class ExtThreadPoolTest {
  
  static class MyTask implements Runnable {
    public String name;
    
    public MyTask(String name) {
      super();
      this.name = name;
    }
    
    @Override
    public void run() {
      try {
        Thread.sleep(500);
        System.out.println("执行中:"+this.name);
        Thread.sleep(500);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
  }
  
  public static void main(String[] args) throws InterruptedException {
    
    ExecutorService es = new ThreadPoolExecutor(5,5,0,TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()){

      @Override
      protected void beforeExecute(Thread t, Runnable r) {
        System.out.println("准备执行:" + ((MyTask)r).name);
      }

      @Override
      protected void afterExecute(Runnable r, Throwable t) {
        System.out.println("执行完成:" + ((MyTask)r).name);
      }

      @Override
      protected void terminated() {
        System.out.println("执行退出");
      }
      
    };
    
    
    for(int i=0;i<5;i++){
      MyTask task = new MyTask("Task-"+i);
      es.execute(task);
    }
    
    Thread.sleep(10);  // 等待terminated()执行
    
    es.shutdown();  // 若无该方法，主线程不会结束。
  }
  
}

public ThreadPoolExecutor(int corePoolSize,
               int maximumPoolSize,
               long keepAliveTime,
               TimeUnit unit,
               BlockingQueue<Runnable> workQueue,
               ThreadFactory threadFactory,
               RejectedExecutionHandler handler) {
  
  .....
  
  }

public static class AbortPolicy implements RejectedExecutionHandler {

    /**
     * Creates an {@code AbortPolicy}.
     */
    public AbortPolicy() { }

    /**
     * Always throws RejectedExecutionException.
     *
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     * @throws RejectedExecutionException always.
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
      throw new RejectedExecutionException("Task " + r.toString() +
                         " rejected from " +
                         e.toString());
    }
  }

public static class DiscardPolicy implements RejectedExecutionHandler {

    /**
     * Creates a {@code DiscardPolicy}.
     */
    public DiscardPolicy() { }


    /**
     * Does nothing, which has the effect of discarding task r.
     *
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    }

  }

public static class CallerRunsPolicy implements RejectedExecutionHandler {

    /**
     * Creates a {@code CallerRunsPolicy}.
     */
    public CallerRunsPolicy() { }

 
    /**
     * Executes task r in the caller's thread, unless the executor
     * has been shut down, in which case the task is discarded.
     *
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
      if (!e.isShutdown()) {
        r.run();
      }
    }
  }

public static class DiscardOldestPolicy implements RejectedExecutionHandler {

    /**
     * Creates a {@code DiscardOldestPolicy} for the given executor.
     */
    public DiscardOldestPolicy() { }

    /**
     * Obtains and ignores the next task that the executor
     * would otherwise execute, if one is immediately available,
     * and then retries execution of task r, unless the executor
     * is shut down, in which case task r is instead discarded.
     *
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
      if (!e.isShutdown()) {
        e.getQueue().poll();
        e.execute(r);
      }
    }
  }

public class RejectedPolicyHandleTest {
  public static void main(String[] args) throws InterruptedException {
    ExecutorService es = new ThreadPoolExecutor(5,5,0,TimeUnit.MILLISECONDS,new SynchronousQueue<Runnable>(),Executors.defaultThreadFactory(),new RejectedExecutionHandler() {
      
      @Override
      public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        // 打印并丢弃。
        System.out.println(r.toString()+" is discard");
      }
    });
    
    for(int i=0;i<Integer.MAX_VALUE;i++){
      MyTask task = new MyTask("Task-"+i);
      es.execute(task);
      Thread.sleep(10);
    }
    
    es.shutdown();  // 若无该方法，主线程不会结束。
    
  }
}

public interface ThreadFactory {
  Thread newThread(Runnable r);
}

static class DefaultThreadFactory implements ThreadFactory {
    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;

    DefaultThreadFactory() {
      SecurityManager s = System.getSecurityManager();
      group = (s != null) ? s.getThreadGroup() :
                 Thread.currentThread().getThreadGroup();
      namePrefix = "pool-" +
             poolNumber.getAndIncrement() +
             "-thread-";
    }

    public Thread newThread(Runnable r) {
      Thread t = new Thread(group, r,
                 namePrefix + threadNumber.getAndIncrement(),
                 0);
      if (t.isDaemon())
        t.setDaemon(false);
      if (t.getPriority() != Thread.NORM_PRIORITY)
        t.setPriority(Thread.NORM_PRIORITY);
      return t;
    }
  }