第6章 任務執行

在並發應用中，避免為每個任務都分配一個線程

• 線程生命周期的開銷很高，在請求到達率很高的情況下將耗費大量計算資源影響性能
• 資源消耗大，可運行的線程數超過CPU數量後，必定會有線程被閑置等待CPU時間片，但是其仍然占用內存保存其狀態，給GC帶來壓力。而且大量線程競爭CPU的時候額外的性能開銷也不可忽視
• 穩定性，無限制的創建線程將難以避免服務器在高負載或遭到惡意攻擊時崩潰，因此需要對應用程序可創建的線程數量進行限制，並進行全面的測試

Executor框架

/**
 * An object that executes submitted {@link Runnable} tasks. This
 * interface provides a way of decoupling task submission from the
 * mechanics of how each task will be run, including details of thread
 * use, scheduling, etc.  An {
 
@code Executor} is normally used
 * instead of explicitly creating threads. For example, rather than
 * invoking {@code new Thread(new(RunnableTask())).start()} for each
 * of a set of tasks, you might use:
 *
 * <pre>
 * Executor executor = <em>anExecutor</em>;
 * executor.execute(new RunnableTask1());
 * executor.execute(new RunnableTask2());
 * ...
 * </pre>
 *
 * However, the {
 
@code Executor} interface does not strictly
 * require that execution be asynchronous. In the simplest case, an
 * executor can run the submitted task immediately in the caller‘s
 * thread:
 *
 *  <pre> {@code
 * class DirectExecutor implements Executor {
 *   public void execute(Runnable r) {
 *     r.run();
 *   }
 * }}</pre>
 *
 * More typically, tasks are executed in some thread other
 * than the caller‘s thread.  The executor below spawns a new thread
 * for each task.
 *
 *  <pre> {
 
@code
 * class ThreadPerTaskExecutor implements Executor {
 *   public void execute(Runnable r) {
 *     new Thread(r).start();
 *   }
 * }}</pre>
 *
 * Many {@code Executor} implementations impose some sort of
 * limitation on how and when tasks are scheduled.  The executor below
 * serializes the submission of tasks to a second executor,
 * illustrating a composite executor.
 *
 *  <pre> {@code
 * 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);
 *     }
 *   }
 * }}</pre>
 *
 * The {@code Executor} implementations provided in this package
 * implement {@link ExecutorService}, which is a more extensive
 * interface.  The {@link ThreadPoolExecutor} class provides an
 * extensible thread pool implementation. The {@link Executors} class
 * provides convenient factory methods for these Executors.
 *
 * <p>Memory consistency effects: Actions in a thread prior to
 * submitting a {@code Runnable} object to an {@code Executor}
 * <a href="package-summary.html#MemoryVisibility"><i>happen-before</i></a>
 * its execution begins, perhaps in another thread.
 *
 * @since 1.5
 * @author Doug Lea
 */
public interface Executor {

    /**
     * Executes the given command at some time in the future.  The command
     * may execute in a new thread, in a pooled thread, or in the calling
     * thread, at the discretion of the {@code Executor} implementation.
     *
     * @param command the runnable task
     * @throws RejectedExecutionException if this task cannot be
     * accepted for execution
     * @throws NullPointerException if command is null
     */
    void execute(Runnable command);
}

上面展示了Executor接口的Java源代碼，它提供了一種標準的方法將任務的提交過程與執行過程解耦合，並用Runable表示一個可執行的任務。

當你需要改變任務執行的方式，你只需要改變executor的實現，而無需影響本身提交+執行的骨幹代碼。

線程池

線程在創建和銷毀過程中是有巨大開銷的，線程池的設計思路就是重用已有線程，提高響應速度，並保證適當的線程數量使得處理器可以保持在忙碌狀態，並且控制住線程競爭造成的內存消耗。

Executor提供了工廠方法來創建線程池：

newFixedThreadPool：固定長度的線程池

newCachedThreadPool：根據處理需求回收和增加線程數，沒有規模限制

newSingleThreadPool：單一線程，可以保證任務的順序（按優先級）執行

newScheduledThreadPool：固定長度的線程池，並且以定時或者延遲的方式來執行任務

ExecutorService

package java.util.concurrent;
import java.util.List;
import java.util.Collection;

/**
 * An {@link Executor} that provides methods to manage termination and
 * methods that can produce a {@link Future} for tracking progress of
 * one or more asynchronous tasks.
 *
 * <p>An {@code ExecutorService} can be shut down, which will cause
 * it to reject new tasks.  Two different methods are provided for
 * shutting down an {@code ExecutorService}. The {@link #shutdown}
 * method will allow previously submitted tasks to execute before
 * terminating, while the {@link #shutdownNow} method prevents waiting
 * tasks from starting and attempts to stop currently executing tasks.
 * Upon termination, an executor has no tasks actively executing, no
 * tasks awaiting execution, and no new tasks can be submitted.  An
 * unused {@code ExecutorService} should be shut down to allow
 * reclamation of its resources.
 *
 * <p>Method {@code submit} extends base method {@link
 * Executor#execute(Runnable)} by creating and returning a {@link Future}
 * that can be used to cancel execution and/or wait for completion.
 * Methods {@code invokeAny} and {@code invokeAll} perform the most
 * commonly useful forms of bulk execution, executing a collection of
 * tasks and then waiting for at least one, or all, to
 * complete. (Class {@link ExecutorCompletionService} can be used to
 * write customized variants of these methods.)
 *
 * <p>The {@link Executors} class provides factory methods for the
 * executor services provided in this package.
 *
 * <h3>Usage Examples</h3>
 *
 * Here is a sketch of a network service in which threads in a thread
 * pool service incoming requests. It uses the preconfigured {@link
 * Executors#newFixedThreadPool} factory method:
 *
 *  <pre> {@code
 * class NetworkService implements Runnable {
 *   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
 *   }
 * }}</pre>
 *
 * The following method shuts down an {@code ExecutorService} in two phases,
 * first by calling {@code shutdown} to reject incoming tasks, and then
 * calling {@code shutdownNow}, if necessary, to cancel any lingering tasks:
 *
 *  <pre> {@code
 * 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();
 *   }
 * }}</pre>
 *
 * <p>Memory consistency effects: Actions in a thread prior to the
 * submission of a {@code Runnable} or {@code Callable} task to an
 * {@code ExecutorService}
 * <a href="package-summary.html#MemoryVisibility"><i>happen-before</i></a>
 * any actions taken by that task, which in turn <i>happen-before</i> the
 * result is retrieved via {@code Future.get()}.
 *
 * @since 1.5
 * @author Doug Lea
 */
public interface ExecutorService extends Executor {

    /**
     * Initiates an orderly shutdown in which previously submitted
     * tasks are executed, but no new tasks will be accepted.
     * Invocation has no additional effect if already shut down.
     *
     * <p>This method does not wait for previously submitted tasks to
     * complete execution.  Use {@link #awaitTermination awaitTermination}
     * to do that.
     *
     * @throws SecurityException if a security manager exists and
     *         shutting down this ExecutorService may manipulate
     *         threads that the caller is not permitted to modify
     *         because it does not hold {@link
     *         java.lang.RuntimePermission}{@code ("modifyThread")},
     *         or the security manager‘s {@code checkAccess} method
     *         denies access.
     */
    void shutdown();

    /**
     * Attempts to stop all actively executing tasks, halts the
     * processing of waiting tasks, and returns a list of the tasks
     * that were awaiting execution.
     *
     * <p>This method does not wait for actively executing tasks to
     * terminate.  Use {@link #awaitTermination awaitTermination} to
     * do that.
     *
     * <p>There are no guarantees beyond best-effort attempts to stop
     * processing actively executing tasks.  For example, typical
     * implementations will cancel via {@link Thread#interrupt}, so any
     * task that fails to respond to interrupts may never terminate.
     *
     * @return list of tasks that never commenced execution
     * @throws SecurityException if a security manager exists and
     *         shutting down this ExecutorService may manipulate
     *         threads that the caller is not permitted to modify
     *         because it does not hold {@link
     *         java.lang.RuntimePermission}{@code ("modifyThread")},
     *         or the security manager‘s {@code checkAccess} method
     *         denies access.
     */
    List<Runnable> shutdownNow();

    /**
     * Returns {@code true} if this executor has been shut down.
     *
     * @return {@code true} if this executor has been shut down
     */
    boolean isShutdown();

    /**
     * Returns {@code true} if all tasks have completed following shut down.
     * Note that {@code isTerminated} is never {@code true} unless
     * either {@code shutdown} or {@code shutdownNow} was called first.
     *
     * @return {@code true} if all tasks have completed following shut down
     */
    boolean isTerminated();

    /**
     * Blocks until all tasks have completed execution after a shutdown
     * request, or the timeout occurs, or the current thread is
     * interrupted, whichever happens first.
     *
     * @param timeout the maximum time to wait
     * @param unit the time unit of the timeout argument
     * @return {@code true} if this executor terminated and
     *         {@code false} if the timeout elapsed before termination
     * @throws InterruptedException if interrupted while waiting
     */
    boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException;

    /**
     * Submits a value-returning task for execution and returns a
     * Future representing the pending results of the task. The
     * Future‘s {@code get} method will return the task‘s result upon
     * successful completion.
     *
     * <p>
     * If you would like to immediately block waiting
     * for a task, you can use constructions of the form
     * {@code result = exec.submit(aCallable).get();}
     *
     * <p>Note: The {@link Executors} class includes a set of methods
     * that can convert some other common closure-like objects,
     * for example, {@link java.security.PrivilegedAction} to
     * {@link Callable} form so they can be submitted.
     *
     * @param task the task to submit
     * @param <T> the type of the task‘s result
     * @return a Future representing pending completion of the task
     * @throws RejectedExecutionException if the task cannot be
     *         scheduled for execution
     * @throws NullPointerException if the task is null
     */
    <T> Future<T> submit(Callable<T> task);

    /**
     * Submits a Runnable task for execution and returns a Future
     * representing that task. The Future‘s {@code get} method will
     * return the given result upon successful completion.
     *
     * @param task the task to submit
     * @param result the result to return
     * @param <T> the type of the result
     * @return a Future representing pending completion of the task
     * @throws RejectedExecutionException if the task cannot be
     *         scheduled for execution
     * @throws NullPointerException if the task is null
     */
    <T> Future<T> submit(Runnable task, T result);

    /**
     * Submits a Runnable task for execution and returns a Future
     * representing that task. The Future‘s {@code get} method will
     * return {@code null} upon <em>successful</em> completion.
     *
     * @param task the task to submit
     * @return a Future representing pending completion of the task
     * @throws RejectedExecutionException if the task cannot be
     *         scheduled for execution
     * @throws NullPointerException if the task is null
     */
    Future<?> submit(Runnable task);

    /**
     * Executes the given tasks, returning a list of Futures holding
     * their status and results when all complete.
     * {@link Future#isDone} is {@code true} for each
     * element of the returned list.
     * Note that a <em>completed</em> task could have
     * terminated either normally or by throwing an exception.
     * The results of this method are undefined if the given
     * collection is modified while this operation is in progress.
     *
     * @param tasks the collection of tasks
     * @param <T> the type of the values returned from the tasks
     * @return a list of Futures representing the tasks, in the same
     *         sequential order as produced by the iterator for the
     *         given task list, each of which has completed
     * @throws InterruptedException if interrupted while waiting, in
     *         which case unfinished tasks are cancelled
     * @throws NullPointerException if tasks or any of its elements are {@code null}
     * @throws RejectedExecutionException if any task cannot be
     *         scheduled for execution
     */
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException;

    /**
     * Executes the given tasks, returning a list of Futures holding
     * their status and results
     * when all complete or the timeout expires, whichever happens first.
     * {@link Future#isDone} is {@code true} for each
     * element of the returned list.
     * Upon return, tasks that have not completed are cancelled.
     * Note that a <em>completed</em> task could have
     * terminated either normally or by throwing an exception.
     * The results of this method are undefined if the given
     * collection is modified while this operation is in progress.
     *
     * @param tasks the collection of tasks
     * @param timeout the maximum time to wait
     * @param unit the time unit of the timeout argument
     * @param <T> the type of the values returned from the tasks
     * @return a list of Futures representing the tasks, in the same
     *         sequential order as produced by the iterator for the
     *         given task list. If the operation did not time out,
     *         each task will have completed. If it did time out, some
     *         of these tasks will not have completed.
     * @throws InterruptedException if interrupted while waiting, in
     *         which case unfinished tasks are cancelled
     * @throws NullPointerException if tasks, any of its elements, or
     *         unit are {@code null}
     * @throws RejectedExecutionException if any task cannot be scheduled
     *         for execution
     */
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                  long timeout, TimeUnit unit)
        throws InterruptedException;

    /**
     * Executes the given tasks, returning the result
     * of one that has completed successfully (i.e., without throwing
     * an exception), if any do. Upon normal or exceptional return,
     * tasks that have not completed are cancelled.
     * The results of this method are undefined if the given
     * collection is modified while this operation is in progress.
     *
     * @param tasks the collection of tasks
     * @param <T> the type of the values returned from the tasks
     * @return the result returned by one of the tasks
     * @throws InterruptedException if interrupted while waiting
     * @throws NullPointerException if tasks or any element task
     *         subject to execution is {@code null}
     * @throws IllegalArgumentException if tasks is empty
     * @throws ExecutionException if no task successfully completes
     * @throws RejectedExecutionException if tasks cannot be scheduled
     *         for execution
     */
    <T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException;

    /**
     * Executes the given tasks, returning the result
     * of one that has completed successfully (i.e., without throwing
     * an exception), if any do before the given timeout elapses.
     * Upon normal or exceptional return, tasks that have not
     * completed are cancelled.
     * The results of this method are undefined if the given
     * collection is modified while this operation is in progress.
     *
     * @param tasks the collection of tasks
     * @param timeout the maximum time to wait
     * @param unit the time unit of the timeout argument
     * @param <T> the type of the values returned from the tasks
     * @return the result returned by one of the tasks
     * @throws InterruptedException if interrupted while waiting
     * @throws NullPointerException if tasks, or unit, or any element
     *         task subject to execution is {@code null}
     * @throws TimeoutException if the given timeout elapses before
     *         any task successfully completes
     * @throws ExecutionException if no task successfully completes
     * @throws RejectedExecutionException if tasks cannot be scheduled
     *         for execution
     */
    <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                    long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

繼承了Executor接口，提供了對於任務執行生命周期的管理方法的接口。

共有3種狀態：運行、關閉和已終止

shutdown方法：

• 啟動平緩關閉過程
• 不再接受新任務
• 等待正在執行的任務完成
• 等在已提交但是未執行的任務完成
• 關閉Executor
• 該方法並不會阻塞等待shutdown過程完全結束，如果需要阻塞，需要額外調用awaitTermination方法

showdownNow方法：強制取消正在運行的任務並且不再開啟新任務

不要使用Timer類，該類早已過期，Try ScheduledExecutorService和DelayQueue的組合

Callable與Future

想到一道面試題，如何用java加載一個大頁面，包括文字圖像，音頻視頻等。

Executor的submit方法返回的都是future對象，對於長時間任務，或者I/O開銷大的任務，可以提交一個實現Callable接口的task，並在需要時才調用future對象的get方法查看執行結果。

這裏書中提到：異構任務的並發，還是取決於慢的那個異構任務有多慢，真正能發揮並發特性的是，大量同構且相互獨立的任務並發處理。（搶票啊，秒殺啊，巴拉巴拉）

CompletionServiceExecutor:

Executor+BlockingQueue, 其實現就是將callable對象在提交時包裝在一個擴展的futuretask對象內，該對象復寫了futuretask的done方法，在任務完成的時候會把future對象加入到一個blockingqueue中。使用方可以通過poll和take等隊列操作獲取隊列中已經完成的future任務結果。

簡單說一下poll和take，看源碼註釋，獲取隊頭對象，並刪除，poll不會阻塞，沒有就返回null，take會，poll還有一個帶timeout參數的方法。

TODO：ExecutorService實現類的源碼解析

第7章 取消與關閉

本章內容較為底層，可能是個人對於中斷的理解有所欠缺，通讀一遍下來不太有收獲，之後準備再去復習一下“中斷”的概念再來重看一遍。

任何代碼都可能拋出一個RuntimeException。每當調用另一個方法時，都要對它的行為保持懷疑，不要盲目地認為它一定會正常返回，或者一定會拋出在方法聲明中聲明的某個已檢查異常。對調用的代碼越不熟悉，就越應該對其代碼行為保持懷疑。

守護線程

在JVM啟動時創建的所有線程中，除了主線程以外，其他都是守護線程，包括GC等。當創建一個新線程時，會繼承創建者的守護狀態，因此，主線程創建的線程都是普通線程。

“當一個線程退出時，JVM會檢查其他正在運行的線程，如果這些線程都是守護線程，那麽JVM會正常退出操作。” 這句話完全沒有讀懂，我感覺這一整章都翻譯的不太好，mark一下之後去看一下英文原版是怎麽描述的。

JVM停止時，所有守護線程將被直接拋棄，所以不要在守護線程裏持有一些需要釋放清理的資源。

本章最後說到避免使用finalize方法，這一點在JVM虛擬機那本書裏也提到了，事實上現在大家都是這麽做的--在finally中調用各種close，release，teardown方法進行收尾清理，關閉連接，歸還文件句柄/套接字句柄等。

第8章 線程池的使用

線程池的大小

想要正確地設置線程池的大小，必須分析計算環境、資源預算和任務的特性。

• 多少個CPU？
• 多大內存？
• 計算密集型 還是 I/O密集型 還是兩者都有？
• 是否需要數據庫連接？

對於計算密集型的任務，在擁有N_cpu個處理器的系統上，當線程池大小為N_cpu+1時，通常能實現最優的利用率。這個額外的線程可以確保，其他線程由於某些原因暫停時，可以利用到空出來的CPU時鐘周期。

對於包含I/O操作或者其他阻塞操作的任務，由於線程不會一直執行，因此線程池的規模應該大一些。要正確的設置線程池的大小，需要估算出任務的等待時間與計算時間的比值。

一種比較簡單的調節線程池大小的方式是：在某個基準負載下，分別設置不同大小的線程池來運行應用程序，觀察CPU的利用率水平。

書上給出的公式：要使處理器達到期望的使用率，線程池的最優大小為：

　　N_threads=N_cpu*U_cpu*（1+W/C）

其中，U_cpu是期望的CPU利用率，W是等待時間，C是計算時間。

而對於內存等其他資源，可以通過以下方式計算線程池的大小約束：

　　先計算每個任務對該資源的需求量，然後用該資源的可用總量除以每個任務的需求量，得出的就是線程池大小的上限。

隊列任務

當請求的到達速率超過了服務器的處理速率，請求就會堆積起來，ThreadPoolExecutor的應對措施是允許通過一個BlockingQueue來保存等待執行的任務。

其中newFixedThreadPool在默認情況下使用的一個無上界隊列，LinkedBlockingQueue，如果所有工作者線程都處於忙碌狀態，那麽任務將在隊列中等候，並且如果處理速度一直跟不上，隊列將無限制的增加。

當然，跟穩妥的方式是使用有界隊列，例如有界的LinkedBlockingQueue，PriorityBlockingQueue等，加上適當的飽和策略（隊列滿了之後如何處理後續來的請求）。

飽和策略

ThreadPoolExecutor的飽和策略可以通過調用setRejectedHandler來修改

• AbortPolicy：拋出異常，讓調用者自己處理
• DiscardPolicy：悄悄的直接拋棄任務
• DiscardOldest：拋棄隊列頭部的任務（或者優先級最高的），嘗試提交新的任務。
• Caller-runs：是一種調節機制，該策略不拋棄任務也不扔出異常，而是將任務退回給調用者的主線程執行，從而使得主線程不能accpet新任務，新任務將會堆積在TCP層面，如果TCP層滿了，會自行拋棄請求，該策略實現了一種平緩的降低性能機制。

可以通過擴展ThreadPoolExecutor（繼承），實現beforeExecute，afterExecute（如果是計算任務時間的話，before和after可以通過threadlocal變量來通信）和terminate等方法，來對任務的執行添加日誌監控和統計信息收集。

第9章 圖形用戶界面應用程序

由於GUI界面比較過氣了，這章大家可以快速通讀一下，看看有沒有什麽值得借鑒的設計思想和方法論，然後進入第三部分：活躍性、性能與測試

《Java並發編程實戰》學習筆記 - 第二部分