摘要：并發(fā)教程原子變量和原文譯者飛龍協(xié)議歡迎閱讀我的多線程編程系列教程的第三部分。如果你能夠在多線程中同時且安全地執(zhí)行某個操作，而不需要關(guān)鍵字或上一章中的鎖，那么這個操作就是原子的。當(dāng)多線程的更新比讀取更頻繁時，這個類通常比原子數(shù)值類性能更好。
Java 8 并發(fā)教程：原子變量和 ConcurrentMap
原文：Java 8 Concurrency Tutorial: Synchronization and Locks
譯者：飛龍
協(xié)議：CC BY-NC-SA 4.0
歡迎閱讀我的Java8多線程編程系列教程的第三部分。這個教程包含并發(fā)API的兩個重要部分：原子變量和ConcurrentMap。由于最近發(fā)布的Java8中的lambda表達式和函數(shù)式編程，二者都有了極大的改進。所有這些新特性會以一些簡單易懂的代碼示例來描述。希望你能喜歡。
第一部分：線程和執(zhí)行器
第二部分：同步和鎖
第三部分：原子變量和 ConcurrentMap
出于簡單的因素，這個教程的代碼示例使用了定義在這里的兩個輔助函數(shù)sleep(seconds) 和 stop(executor)。
AtomicInteger
java.concurrent.atomic包包含了許多實用的類，用于執(zhí)行原子操作。如果你能夠在多線程中同時且安全地執(zhí)行某個操作，而不需要synchronized關(guān)鍵字或上一章中的鎖，那么這個操作就是原子的。
本質(zhì)上，原子操作嚴(yán)重依賴于比較與交換（CAS），它是由多數(shù)現(xiàn)代CPU直接支持的原子指令。這些指令通常比同步塊要快。所以在只需要并發(fā)修改單個可變變量的情況下，我建議你優(yōu)先使用原子類，而不是上一章展示的鎖。
譯者注：對于其它語言，一些語言的原子操作用鎖實現(xiàn)，而不是原子指令。
現(xiàn)在讓我們選取一個原子類，例如AtomicInteger：
AtomicInteger atomicInt = new AtomicInteger(0);
ExecutorService executor = Executors.newFixedThreadPool(2);
IntStream.range(0, 1000)
.forEach(i -> executor.submit(atomicInt::incrementAndGet));
stop(executor);
System.out.println(atomicInt.get()); // => 1000
通過使用AtomicInteger代替Integer，我們就能線程安全地并發(fā)增加數(shù)值，而不需要同步訪問變量。incrementAndGet()方法是原子操作，所以我們可以在多個線程中安全調(diào)用它。
AtomicInteger支持多種原子操作。updateAndGet()接受lambda表達式，以便在整數(shù)上執(zhí)行任意操作：
AtomicInteger atomicInt = new AtomicInteger(0);
ExecutorService executor = Executors.newFixedThreadPool(2);
IntStream.range(0, 1000)
.forEach(i -> {
Runnable task = () ->
atomicInt.updateAndGet(n -> n + 2);
executor.submit(task);
});
stop(executor);
System.out.println(atomicInt.get()); // => 2000
accumulateAndGet()方法接受另一種類型IntBinaryOperator的lambda表達式。我們在下個例子中，使用這個方法并發(fā)計算0~1000所有值的和：
AtomicInteger atomicInt = new AtomicInteger(0);
ExecutorService executor = Executors.newFixedThreadPool(2);
IntStream.range(0, 1000)
.forEach(i -> {
Runnable task = () ->
atomicInt.accumulateAndGet(i, (n, m) -> n + m);
executor.submit(task);
});
stop(executor);
System.out.println(atomicInt.get()); // => 499500
其它實用的原子類有AtomicBoolean、AtomicLong 和 AtomicReference。
LongAdder
LongAdder是AtomicLong的替代，用于向某個數(shù)值連續(xù)添加值。
ExecutorService executor = Executors.newFixedThreadPool(2);
IntStream.range(0, 1000)
.forEach(i -> executor.submit(adder::increment));
stop(executor);
System.out.println(adder.sumThenReset()); // => 1000
LongAdder提供了add()和increment()方法，就像原子數(shù)值類一樣，同樣是線程安全的。但是這個類在內(nèi)部維護一系列變量來減少線程之間的爭用，而不是求和計算單一結(jié)果。實際的結(jié)果可以通過調(diào)用sum()或sumThenReset()來獲取。
當(dāng)多線程的更新比讀取更頻繁時，這個類通常比原子數(shù)值類性能更好。這種情況在抓取統(tǒng)計數(shù)據(jù)時經(jīng)常出現(xiàn)，例如，你希望統(tǒng)計Web服務(wù)器上請求的數(shù)量。LongAdder缺點是較高的內(nèi)存開銷，因為它在內(nèi)存中儲存了一系列變量。
LongAccumulator
LongAccumulator是LongAdder的更通用的版本。LongAccumulator以類型為LongBinaryOperatorlambda表達式構(gòu)建，而不是僅僅執(zhí)行加法操作，像這段代碼展示的那樣：
LongBinaryOperator op = (x, y) -> 2 * x + y;
LongAccumulator accumulator = new LongAccumulator(op, 1L);
ExecutorService executor = Executors.newFixedThreadPool(2);
IntStream.range(0, 10)
.forEach(i -> executor.submit(() -> accumulator.accumulate(i)));
stop(executor);
System.out.println(accumulator.getThenReset()); // => 2539
我們使用函數(shù)2 * x + y創(chuàng)建了LongAccumulator，初始值為1。每次調(diào)用accumulate(i)的時候，當(dāng)前結(jié)果和值i都會作為參數(shù)傳入lambda表達式。
LongAccumulator就像LongAdder那樣，在內(nèi)部維護一系列變量來減少線程之間的爭用。
ConcurrentMap
ConcurrentMap接口繼承自Map接口，并定義了最實用的并發(fā)集合類型之一。Java8通過將新的方法添加到這個接口，引入了函數(shù)式編程。
在下面的代碼中，我們使用這個映射示例來展示那些新的方法：
ConcurrentMap map = new ConcurrentHashMap<>();
map.put("foo", "bar");
map.put("han", "solo");
map.put("r2", "d2");
map.put("c3", "p0");
forEach()方法接受類型為BiConsumer的lambda表達式，以映射的鍵和值作為參數(shù)傳遞。它可以作為for-each循環(huán)的替代，來遍歷并發(fā)映射中的元素。迭代在當(dāng)前線程上串行執(zhí)行。
map.forEach((key, value) -> System.out.printf("%s = %s
", key, value));
新方法putIfAbsent()只在提供的鍵不存在時，將新的值添加到映射中。至少在ConcurrentHashMap的實現(xiàn)中，這一方法像put()一樣是線程安全的，所以你在不同線程中并發(fā)訪問映射時，不需要任何同步機制。
String value = map.putIfAbsent("c3", "p1");
System.out.println(value); // p0
getOrDefault()方法返回指定鍵的值。在傳入的鍵不存在時，會返回默認(rèn)值：
String value = map.getOrDefault("hi", "there");
System.out.println(value); // there
replaceAll()接受類型為BiFunction的lambda表達式。BiFunction接受兩個參數(shù)并返回一個值。函數(shù)在這里以每個元素的鍵和值調(diào)用，并返回要映射到當(dāng)前鍵的新值。
map.replaceAll((key, value) -> "r2".equals(key) ? "d3" : value);
System.out.println(map.get("r2")); // d3
compute()允許我們轉(zhuǎn)換單個元素，而不是替換映射中的所有值。這個方法接受需要處理的鍵，和用于指定值的轉(zhuǎn)換的BiFunction。
map.compute("foo", (key, value) -> value + value);
System.out.println(map.get("foo")); // barbar
除了compute()之外還有兩個變體：computeIfAbsent() 和 computeIfPresent()。這些方法的函數(shù)式參數(shù)只在鍵不存在或存在時被調(diào)用。
最后，merge()方法可以用于以映射中的現(xiàn)有值來統(tǒng)一新的值。這個方法接受鍵、需要并入現(xiàn)有元素的新值，以及指定兩個值的合并行為的BiFunction。
map.merge("foo", "boo", (oldVal, newVal) -> newVal + " was " + oldVal);
System.out.println(map.get("foo")); // boo was foo
ConcurrentHashMap
所有這些方法都是ConcurrentMap接口的一部分，因此可在所有該接口的實現(xiàn)上調(diào)用。此外，最重要的實現(xiàn)ConcurrentHashMap使用了一些新的方法來改進，便于在映射上執(zhí)行并行操作。
就像并行流那樣，這些方法使用特定的ForkJoinPool，由Java8中的ForkJoinPool.commonPool()提供。該池使用了取決于可用核心數(shù)量的預(yù)置并行機制。我的電腦有四個核心可用，這會使并行性的結(jié)果為3：
System.out.println(ForkJoinPool.getCommonPoolParallelism()); // 3
這個值可以通過設(shè)置下列JVM參數(shù)來增減：
-Djava.util.concurrent.ForkJoinPool.common.parallelism=5
我們使用相同的映射示例來展示，但是這次我們使用具體的ConcurrentHashMap實現(xiàn)而不是ConcurrentMap接口，所以我們可以訪問這個類的所有公共方法：
ConcurrentHashMap map = new ConcurrentHashMap<>();
map.put("foo", "bar");
map.put("han", "solo");
map.put("r2", "d2");
map.put("c3", "p0");
Java8引入了三種類型的并行操作：forEach、search 和 reduce。這些操作中每個都以四種形式提供，接受以鍵、值、元素或鍵值對為參數(shù)的函數(shù)。
所有這些方法的第一個參數(shù)是通用的parallelismThreshold。這一閾值表示操作并行執(zhí)行時的最小集合大小。例如，如果你傳入閾值500，而映射的實際大小是499，那么操作就會在單線程上串行執(zhí)行。在下一個例子中，我們使用閾值1，始終強制并行執(zhí)行來展示。
forEach
forEach()方法可以并行迭代映射中的鍵值對。BiConsumer以當(dāng)前迭代元素的鍵和值調(diào)用。為了將并行執(zhí)行可視化，我們向控制臺打印了當(dāng)前線程的名稱。要注意在我這里底層的ForkJoinPool最多使用三個線程。
map.forEach(1, (key, value) ->
System.out.printf("key: %s; value: %s; thread: %s
",
key, value, Thread.currentThread().getName()));
// key: r2; value: d2; thread: main
// key: foo; value: bar; thread: ForkJoinPool.commonPool-worker-1
// key: han; value: solo; thread: ForkJoinPool.commonPool-worker-2
// key: c3; value: p0; thread: main
search
search()方法接受BiFunction并為當(dāng)前的鍵值對返回一個非空的搜索結(jié)果，或者在當(dāng)前迭代不匹配任何搜索條件時返回null。只要返回了非空的結(jié)果，就不會往下搜索了。要記住ConcurrentHashMap是無序的。搜索函數(shù)應(yīng)該不依賴于映射實際的處理順序。如果映射的多個元素都滿足指定搜索函數(shù)，結(jié)果是非確定的。
String result = map.search(1, (key, value) -> {
System.out.println(Thread.currentThread().getName());
if ("foo".equals(key)) {
return value;
}
return null;
});
System.out.println("Result: " + result);
// ForkJoinPool.commonPool-worker-2
// main
// ForkJoinPool.commonPool-worker-3
// Result: bar
下面是另一個例子，僅僅搜索映射中的值：
String result = map.searchValues(1, value -> {
System.out.println(Thread.currentThread().getName());
if (value.length() > 3) {
return value;
}
return null;
});
System.out.println("Result: " + result);
// ForkJoinPool.commonPool-worker-2
// main
// main
// ForkJoinPool.commonPool-worker-1
// Result: solo
reduce
reduce()方法已經(jīng)在Java 8 的數(shù)據(jù)流之中用過了，它接受兩個BiFunction類型的lambda表達式。第一個函數(shù)將每個鍵值對轉(zhuǎn)換為任意類型的單一值。第二個函數(shù)將所有這些轉(zhuǎn)換后的值組合為單一結(jié)果，并忽略所有可能的null值。
String result = map.reduce(1,
(key, value) -> {
System.out.println("Transform: " + Thread.currentThread().getName());
return key + "=" + value;
},
(s1, s2) -> {
System.out.println("Reduce: " + Thread.currentThread().getName());
return s1 + ", " + s2;
});
System.out.println("Result: " + result);
// Transform: ForkJoinPool.commonPool-worker-2
// Transform: main
// Transform: ForkJoinPool.commonPool-worker-3
// Reduce: ForkJoinPool.commonPool-worker-3
// Transform: main
// Reduce: main
// Reduce: main
// Result: r2=d2, c3=p0, han=solo, foo=bar
我希望你能喜歡我的Java8并發(fā)系列教程的第三部分。這個教程的代碼示例托管在Github上，還有許多其它的Java8代碼片段。歡迎fork我的倉庫并自己嘗試。
如果你想要支持我的工作，請向你的朋友分享這篇教程。你也可以在Twiiter上關(guān)注我，因為我會不斷推送一些Java或編程相關(guān)的東西。
第一部分：線程和執(zhí)行器
第二部分：同步和鎖
第三部分：原子變量和 ConcurrentMap
