摘要：進程線程與協程它們都是并行機制的解決方案。選擇是任意性的，并在對實現做出決定時發生。線程池的大小一旦達到最大值就會保持不變，如果某個線程因為執行異常而結束，那么線程池會補充一個新線程。此線程池支持定時以及周期性執行任務的需求。

并發(Concurrency): 問題域中的概念—— 程序需要被設計成能夠處理多個同時（或者幾乎同時）發生的事件

并行(Parallelism): 方法域中的概念——通過將問題中的多個部分 并行執行，來加速解決問題。

它們都是并行機制的解決方案。

進程: 進程是什么呢？直白地講，進程就是應用程序的啟動實例。比如我們運行一個游戲，打開一個軟件，就是開啟了一個進程。進程擁有代碼和打開的文件資源、數據資源、獨立的內存空間。啟動一個進程非常消耗資源，一般一臺機器最多啟動數百個進程。

線程: 線程從屬于進程，是程序的實際執行者。一個進程至少包含一個主線程，也可以有更多的子線程。線程擁有自己的棧空間。在進程內啟動線程也要消耗一定的資源，一般一個進程最多啟動數千個線程。操作系統能夠調度的最小單位就是線程了。

協程: 協程又從屬于線程，它不屬于操作系統管轄，完全由程序控制，一個線程內可以啟動數萬甚至數百萬協程。但也正是因為它由程序控制，它對編寫代碼的風格改變也最多。

主線程: 獨立生命周期的線程

守護線程: 被主線程創建,隨著創建線程結束而結束

要注意的是,線程不是調用start之后馬上進入運行中的狀態,而是在"可運行"狀態,由操作系統來決定調度哪個線程來運行。

Web服務器都有自己管理的線程池, 比如輕量級的Jetty, 就有以下三種類型的線程:

Acceptor

Selector

Worker

繼承Thread類

實現Runnable接口

實際使用中顯然實現接口更好, 避免了單繼承限制。

Runnable：實現run方法，無法拋出受檢查的異常，運行時異常會中斷主線程，但主線程無法捕獲，所以子線程應該自己處理所有異常

Callable：實現call方法，可以拋出受檢查的異常，可以被主線程捕獲，但主線程無法捕獲運行時異常，也不會被打斷。

需要返回值的話，就用Callable接口
一個實現了Callable接口的對象,需要被包裝為RunnableFuture對象, 然后才能被新線程執行, 而RunnableFuture其實還是實現了Runnable接口。

Future, Runnable 和FutureTask的關系如下:

可以看出FutureTask其實是RunnableFuture接口的實現類，下面是使用Future的示例代碼

public class Callee implements Callable {
    AtomicInteger counter = new AtomicInteger(0);

    private Integer seq=null;

    public Callee()
    {
        super();
    }

    public  Callee(int seq)
    {
        this.seq = seq;
    }

    /**
     * call接口可以拋出受檢查的異常
     * @return
     * @throws InterruptedException
     */
    @Override
    public Person call() throws InterruptedException {
        Person p = new Person("person"+ counter.incrementAndGet(), RandomUtil.random(0,150));
        System.out.println("In thread("+seq+"), create a Person: "+p.toString());
        Thread.sleep(1000);
        return  p;
    }
}

Callee callee1 = new Callee();
FutureTask ft= new FutureTask(callee1);
Thread thread = new Thread(ft);
thread.start();

try {
    thread.join();
} catch (InterruptedException e) {
    e.printStackTrace();
    return;
}

System.out.println("ft.isDone: "+ft.isDone());

Person result1;
try {
    result1 = ((Future) ft).get();
} catch (InterruptedException e) {
    e.printStackTrace();
    result1 = null;
} catch (ExecutionException e) {
    e.printStackTrace();
    result1 = null;
}
Person result = result1;
System.out.println("main thread get result: "+result.toString());

Thread.yield() 方法：調用這個方法，會讓當前線程退回到可運行狀態，而不是阻塞狀態，這樣就留給其他同級線程一些運行機會

Thread.sleep(long millis)：調用這個方法,真的會讓當前線程進入阻塞狀態,直到時間結束

線程對象的join()：這個方法讓當前線程進入阻塞狀態，直到要等待的線程結束。

線程對象的interrupt():不要以為它是中斷某個線程！它只是線線程發送一個中斷信號，讓線程在無限等待時（如死鎖時）能拋出異常，從而結束線程，但是如果你吃掉了這個異常，那么這個線程還是不會中斷的！

Object類中的wait()：線程進入等待狀態，直到其他線程調用此對象的 notify() 方法或 notifyAll() 喚醒方法。這個狀態跟加鎖有關，所以是Object的方法。

Object類中的notify()：喚醒在此對象監視器上等待的單個線程。如果所有線程都在此對象上等待，則會選擇喚醒其中一個線程。選擇是任意性的，并在對實現做出決定時發生。線程通過調用其中一個 wait 方法，在對象的監視器上等待。 直到當前的線程放棄此對象上的鎖定，才能繼續執行被喚醒的線程。被喚醒的線程將以常規方式與在該對象上主動同步的其他所有線程進行競爭；類似的方法還有一個notifyAll()，喚醒在此對象監視器上等待的所有線程。

由于線程在并行時,可能會"同時"訪問一個變量, 所以共享變量的時候,會出現值處于一個不確定的狀況, 例如下面的代碼, c是一個實例變量, 多個線程同時訪問increment或decrement方法時,就可能出現一致性錯誤,最終讓c變成"奇怪"的值。

public class Counter {
    private int c = 0;

    public void increment() {
        c++;
    }

    public void decrement() {
        c--;
    }

    public int value() {
        return c;
    }
}

public class Foo {
    private int x = -1;
    private volatile boolean v = false;
    public void setX(int x) {
        this.x = x;
        v = true;
    }
    public int getX() {
        if (v == true) {
            return x;
        }
        return 0;
    }
}

volatile關鍵字實際上指定了變量不使用寄存器, 并且對變量的訪問不會亂序執行，從而避免了并行訪問的不一致問題。但這個方案僅僅對原始類型變量本身生效，如果是++或者--這種“非原子”操作，則不能保證多線程操作的正確性了

JDK提供了一系列對基本類型的封裝,形成原子類型(Atomic Variables)，特別適合用來做計數器

import java.util.concurrent.atomic.AtomicInteger;

class AtomicCounter {
    private AtomicInteger c = new AtomicInteger(0);

    public void increment() {
        c.incrementAndGet();
    }

    public void decrement() {
        c.decrementAndGet();
    }

    public int value() {
        return c.get();
    }
}

原子操作的實現原理，在Java8之前和之后不同

Java7

public final int getAndIncrement() {
    for (;;) {
        int current = get();
        int next = current + 1;
        if (compareAndSet(current, next))
            return current;
    }
}

Java8

public final int getAndIncrement() {
    return unsafe.getAndAddInt(this, valueOffset, 1);
}

至于Compare-and-Swap，以及Fetch-and-Add兩種算法，是依賴機器底層機制實現的。

BlockingQueue: 定義了一個先進先出的數據結構，當你嘗試往滿隊列中添加元素，或者從空隊列中獲取元素時，將會阻塞或者超時

ConcurrentMap: 是 java.util.Map 的子接口，定義了一些有用的原子操作。移除或者替換鍵值對的操作只有當 key 存在時才能進行，而新增操作只有當 key 不存在時。使這些操作原子化，可以避免同步。ConcurrentMap 的標準實現是 ConcurrentHashMap，它是 HashMap 的并發模式。

ConcurrentNavigableMap: 是 ConcurrentMap 的子接口，支持近似匹配。ConcurrentNavigableMap 的標準實現是 ConcurrentSkipListMap，它是 TreeMap 的并發模式。

ThreadLoal 變量，它的基本原理是，同一個 ThreadLocal 所包含的對象（對ThreadLocal< String >而言即為 String 類型變量），在不同的 Thread 中有不同的副本（實際是不同的實例，后文會詳細闡述）。這里有幾點需要注意

因為每個 Thread 內有自己的實例副本，且該副本只能由當前 Thread 使用。這是也是 ThreadLocal 命名的由來
既然每個 Thread 有自己的實例副本，且其它 Thread 不可訪問，那就不存在多線程間共享的問題。

它與普通變量的區別在于，每個使用該變量的線程都會初始化一個完全獨立的實例副本。ThreadLocal 變量通常被private static修飾。當一個線程結束時，它所使用的所有 ThreadLocal 相對的實例副本都可被回收。

總的來說，ThreadLocal 適用于每個線程需要自己獨立的實例且該實例需要在多個方法中被使用，也即變量在線程間隔離而在方法或類間共享的場景。后文會通過實例詳細闡述該觀點。另外，該場景下，并非必須使用 ThreadLocal ，其它方式完全可以實現同樣的效果，只是 ThreadLocal 使得實現更簡潔。

方法加鎖：其實不是加在指定的方法上，而是在指定的對象上，只不過在方法開始前會檢查這個鎖

靜態方法鎖：加在類上，它和加在對象上的鎖互補干擾

代碼區塊鎖：其實不是加在指定的代碼塊上，而是加在指定的對象上，只不過在代碼塊開始前會檢查這個鎖。一個對象只會有一個鎖，所以代碼塊鎖和實例方法鎖是會互相影響的

需要注意的是：無論synchronized關鍵字加在方法上還是對象上，它取得的鎖都是對象，而不是把一段代碼或函數當作鎖――而且同步方法很可能還會被其他線程的對象訪問,每個對象只有一個鎖（lock）與之相關聯

加鎖不慎可能會造成死鎖

真正的多線程使用，是從線程池開始的，Callable接口，基本上也是被線程池調用的。

        ExecutorService pool = Executors.newFixedThreadPool(3);

        Callable worker1 = new Callee();
        Future ft1 = pool.submit(worker1);

        Callable worker2 = new Callee();
        Future ft2 = pool.submit(worker2);

        Callable worker3 = new Callee();
        Future ft3 = pool.submit(worker3);

        System.out.println("準備通知線程池shutdown...");
        pool.shutdown();
        System.out.println("已通知線程池shutdown");
        try {
            pool.awaitTermination(2L, TimeUnit.SECONDS);
            System.out.println("線程池完全結束");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

任務排隊：當前能并發執行的線程數總是有限的，但任務數可以很大

線程調度：線程的創建是比較消耗資源的，需要一個池來維持活躍線程

結果收集：每個任務完成以后，其結果需要統一采集

newSingleThreadExecutor：創建一個單線程的線程池。這個線程池只有一個線程在工作，也就是相當于單線程串行執行所有任務。如果這個唯一的線程因為異常結束，那么會有一個新的線程來替代它。此線程池保證所有任務的執行順序按照任務的提交順序執行。

newFixedThreadPool：創建固定大小的線程池。每次提交一個任務就創建一個線程，直到線程達到線程池的最大大小。線程池的大小一旦達到最大值就會保持不變，如果某個線程因為執行異常而結束，那么線程池會補充一個新線程。

newCachedThreadPool：創建一個可緩存的線程池。如果線程池的大小超過了處理任務所需要的線程，那么就會回收部分空閑（60秒不執行任務）的線程，當任務數增加時，此線程池又可以智能的添加新線程來處理任務。此線程池不會對線程池大小做限制，線程池大小完全依賴于操作系統（或者說JVM）能夠創建的最大線程大小。

newScheduledThreadPool：創建一個大小無限的線程池。此線程池支持定時以及周期性執行任務的需求。

newSingleThreadScheduledExecutor：創建一個單線程的線程池。此線程池支持定時以及周期性執行任務的需求。

線程池在構造前（new操作）是初始狀態，一旦構造完成線程池就進入了執行狀態RUNNING。嚴格意義上講線程池構造完成后并沒有線程被立即啟動，只有進行“預啟動”或者接收到任務的時候才會啟動線程。這個會后面線程池的原理會詳細分析。但是線程池是出于運行狀態，隨時準備接受任務來執行。

線程池運行中可以通過shutdown()和shutdownNow()來改變運行狀態。shutdown()是一個平緩的關閉過程，線程池停止接受新的任務，同時等待已經提交的任務執行完畢，包括那些進入隊列還沒有開始的任務，這時候線程池處于SHUTDOWN狀態；shutdownNow()是一個立即關閉過程，線程池停止接受新的任務，同時線程池取消所有執行的任務和已經進入隊列但是還沒有執行的任務，這時候線程池處于STOP狀態。

一旦shutdown()或者shutdownNow()執行完畢，線程池就進入TERMINATED狀態，此時線程池就結束了。

isTerminating()描述的是SHUTDOWN和STOP兩種狀態。

isShutdown()描述的是非RUNNING狀態，也就是SHUTDOWN/STOP/TERMINATED三種狀態。

計算密集型的任務，最好很少有IO等待，也沒有Sleep之類的，最好是本身就適合遞歸處理的算法

在給定的線程數內，盡可能地最大化利用CPU資源，但又不會導致其他資源過載（比如內存），或者大量空線程等待。

ForkJoinPool主要用來使用分治法(Divide-and-Conquer Algorithm)來解決問題。典型的應用比如快速排序算法。

這里的要點在于，ForkJoinPool需要使用相對少的線程來處理大量的任務。

比如要對1000萬個數據進行排序，那么會將這個任務分割成兩個500萬的排序任務和一個針對這兩組500萬數據的合并任務。以此類推，對于500萬的數據也會做出同樣的分割處理，到最后會設置一個閾值來規定當數據規模到多少時，停止這樣的分割處理。比如，當元素的數量小于10時，會停止分割，轉而使用插入排序對它們進行排序。

那么到最后，所有的任務加起來會有大概2000000+個。問題的關鍵在于，對于一個任務而言，只有當它所有的子任務完成之后，它才能夠被執行。

所以當使用ThreadPoolExecutor時，使用分治法會存在問題，因為ThreadPoolExecutor中的線程無法像任務隊列中再添加一個任務并且在等待該任務完成之后再繼續執行。而使用ForkJoinPool時，就能夠讓其中的線程創建新的任務，并掛起當前的任務，此時線程就能夠從隊列中選擇子任務執行。

以上程序的關鍵是fork()和join()方法。在ForkJoinPool使用的線程中，會使用一個內部隊列來對需要執行的任務以及子任務進行操作來保證它們的執行順序。

那么使用ThreadPoolExecutor或者ForkJoinPool，會有什么性能的差異呢？

首先，使用ForkJoinPool能夠使用數量有限的線程來完成非常多的具有父子關系的任務，比如使用4個線程來完成超過200萬個任務。但是，使用ThreadPoolExecutor時，是不可能完成的，因為ThreadPoolExecutor中的Thread無法選擇優先執行子任務，需要完成200萬個具有父子關系的任務時，也需要200萬個線程，顯然這是不可行的。

ps：ForkJoinPool在執行過程中，會創建大量的子任務，導致GC進行垃圾回收，這些是需要注意的。

ForkJoinPool首先是ExecutorService的實現類，因此是特殊的線程池。

創建了ForkJoinPool實例之后，就可以調用ForkJoinPool的submit(ForkJoinTask task) 或invoke(ForkJoinTask task)方法來執行指定任務了。

其中ForkJoinTask代表一個可以并行、合并的任務。ForkJoinTask是一個抽象類，它還有兩個抽象子類：RecusiveAction和RecusiveTask。其中RecusiveTask代表有返回值的任務，而RecusiveAction代表沒有返回值的任務。

個人認為ForkJoinPool設計不太好的地方在于，ForkJoinTask不是個接口，而是抽象類，實際使用時基本上不是繼承RecursiveAction就是繼承RecursiveTask，對業務類有限制。

典型的一個例子，就是一串數組求和

public interface Calculator {
    long sumUp(long[] numbers);
}

public class ForkJoinCalculator implements Calculator {
    private ForkJoinPool pool;

    private static class SumTask extends RecursiveTask {
        private long[] numbers;
        private int from;
        private int to;

        public SumTask(long[] numbers, int from, int to) {
            this.numbers = numbers;
            this.from = from;
            this.to = to;
        }

        @Override
        protected Long compute() {
            // 當需要計算的數字小于6時，直接計算結果
            if (to - from < 6) {
                long total = 0;
                for (int i = from; i <= to; i++) {
                    total += numbers[i];
                }
                return total;
            // 否則，把任務一分為二，遞歸計算
            } else {
                int middle = (from + to) / 2;
                SumTask taskLeft = new SumTask(numbers, from, middle);
                SumTask taskRight = new SumTask(numbers, middle+1, to);
                taskLeft.fork();
                taskRight.fork();
                return taskLeft.join() + taskRight.join();
            }
        }
    }

    public ForkJoinCalculator() {
        // 也可以使用公用的 ForkJoinPool：
        // pool = ForkJoinPool.commonPool()
        pool = new ForkJoinPool();
    }

    @Override
    public long sumUp(long[] numbers) {
        return pool.invoke(new SumTask(numbers, 0, numbers.length-1));
    }
}

這個例子展示了當數組被拆分得足夠小（<6）之后，就不需要并行處理了，而更大的數組就拆為兩半，分別處理。

別搞混了，跟IO的Stream完全不是一回事，可以把它看做是集合處理的聲明式語法，類似數據庫操作語言SQL。當然也有跟IO類似的地方，就是Stream只能消費一次，不能重復使用。

看個例子：

int sum = widgets.stream()
.filter(w -> w.getColor() == RED)
 .mapToInt(w -> w.getWeight())
 .sum();

流提供了一個能力，任何一個流，只要獲取一次并行流，后面的操作就都可以并行了。
例如：

Stream stream = Stream.of("a", "b", "c","d","e","f","g");
String str = stream.parallel().reduce((a, b) -> a + "," + b).get();
System.out.println(str);

Collection.stream()

Collection.parallelStream()

Arrays.stream(T array) or Stream.of()

java.io.BufferedReader.lines()

java.util.stream.IntStream.range()

java.nio.file.Files.walk()

java.util.Spliterator

Random.ints()

BitSet.stream()

Pattern.splitAsStream(java.lang.CharSequence)

JarFile.stream()

示例

// 1. Individual values
Stream stream = Stream.of("a", "b", "c");
// 2. Arrays
String [] strArray = new String[] {"a", "b", "c"};
stream = Stream.of(strArray);
stream = Arrays.stream(strArray);
// 3. Collections
List list = Arrays.asList(strArray);
stream = list.stream();

需要注意的是，對于基本數值型，目前有三種對應的包裝類型 Stream：

IntStream、LongStream、DoubleStream。當然我們也可以用 Stream、Stream >、Stream，但是 boxing 和 unboxing 會很耗時，所以特別為這三種基本數值型提供了對應的 Stream。

一個流可以后面跟隨零個或多個 intermediate 操作。其目的主要是打開流，做出某種程度的數據映射/過濾，然后返回一個新的流，交給下一個操作使用。這類操作都是惰性化的（lazy），就是說，僅僅調用到這類方法，并沒有真正開始流的遍歷。

已知的Intermediate操作包括：map (mapToInt, flatMap 等)、 filter、 distinct、 sorted、 peek、 limit、 skip、 parallel、 sequential、 unordered。

一個流只能有一個 terminal操作，當這個操作執行后，流就被使用“光”了，無法再被操作。所以這必定是流的最后一個操作。Terminal 操作的執行，才會真正開始流的遍歷，并且會生成一個結果，或者一個 side effect。

已知的Terminal操作包括：forEach、 forEachOrdered、 toArray、 reduce、 collect、 min、 max、 count、 anyMatch、 allMatch、 noneMatch、 findFirst、 findAny、 iterator

reduce解析: reduce本質上是個聚合方法，它的作用是用流里面的元素生成一個結果，所以用來做累加，字符串拼接之類的都非常合適。它有三個參數

初始值：最終結果的初始化值，可以是一個空的對象

聚合函數：一個二元函數（有兩個參數），第一個參數是上一次聚合的結果，第二個參數是某個元素

多個部分結果的合并函數:如果流并發了,那么聚合操作會分為多段進行,這里顯示了多段之間如何配合

collect: collect比reduce更強大:reduce最終只能得到一個跟流里數據類型相同的值, 但collect的結果可以是任何對象。簡單的collect也有三個參數：

最終要返回的數據容器

把元素并入返回值的方法

多個部分結果的合并

兩個collect示例

//和reduce相同的合并字符操作
String concat = stringStream.collect(StringBuilder::new, StringBuilder::append,StringBuilder::append).toString();
//等價于上面,這樣看起來應該更加清晰
String concat = stringStream.collect(() -> new StringBuilder(),(l, x) -> l.append(x), (r1, r2) -> r1.append(r2)).toString();

//把stream轉成map
Stream stream = Stream.of(1, 2, 3, 4).filter(p -> p > 2);

List result = stream.collect(() -> new ArrayList<>(), (list, item) -> list.add(item), (one, two) -> one.addAll(two));
/* 或者使用方法引用 */
result = stream.collect(ArrayList::new, List::add, List::addAll);

協程，英文Coroutines，也叫纖程（Fiber）是一種比線程更加輕量級的存在。正如一個進程可以擁有多個線程一樣，一個線程也可以擁有多個協程。

協程實際上是在語言底層(或者框架)對需要等待的程序進行調度,從而充分利用CPU的方法, 其實這完全可以通過回調來實現, 但是深層回調的代碼太{{BANNED}}了,所以發明了協程的寫法。理論上多個協程不會真的"同時"執行，也就不會引起共享變量操作的不確定性,不需要加鎖(待確認)。

pythone協程示例

Pythone， Golang和C#都內置了協程的語法，但Java沒有，只能通過框架實現，常見的框架包括：Quasar，kilim和ea-async。

Java ea-async 協程示例

import static com.ea.async.Async.await;
import static java.util.concurrent.CompletableFuture.completedFuture;

public class Store
{
    //購物操作, 傳一個商品id和一個價格
    public CompletableFuture buyItem(String itemTypeId, int cost)
    {
        //銀行扣款(長時間操作)
        if(!await(bank.decrement(cost))) {
            return completedFuture(false);
        }
        try {
            //商品出庫(長時間操作)
            await(inventory.giveItem(itemTypeId));
            return completedFuture(true);
        } catch (Exception ex) {
            await(bank.refund(cost));
            throw new AppException(ex);
        }
    }
}

《七周七并發模型》電子書

深入淺出Java Concurrency——線程池

Java多線程學習（吐血超詳細總結）

Jetty基礎之線程模型

Jetty-server高性能,多線程特性的源碼分析

Java 編程要點之并發（Concurrency）詳解

Java Concurrency in Depth (Part 1)

Java進階（七）正確理解Thread Local的原理與適用場景

Java 并發編程筆記：如何使用 ForkJoinPool 以及原理

ForkJoinPool簡介

多線程 ForkJoinPool

Java 8 中的 Streams API 詳解

Java中的協程實現

漫畫:什么是協程

學習源碼