轉載自??Java8 Striped64 和 LongAdder

數據 STRIPING

根據維基百科的這段說明：

In computer data storage, data striping is the technique of segmenting logically sequential data, such as a file, so that consecutive segments are stored on different physical storage devices.

Striping is useful when a processing device requests data more quickly than a single storage device can provide it. By spreading segments across multiple devices which can be accessed concurrently, total data throughput is increased. It is also a useful method for balancing I/O load across an array of disks. Striping is used across disk drives in redundant array of independent disks (RAID) storage, network interface controllers, different computers in clustered file systems and grid-oriented storage, and RAM in some systems.

數據 striping 就是把邏輯上連續的數據分為多個段，使這一序列的段存儲在不同的物理設備上。通過把段分散到多個設備上可以增加訪問并發性，從而提升總體的吞吐量。

Striped64

JDK 8 的?java.util.concurrent.atomic?下有一個包本地的類?Striped64?，它持有常見表示和機制用于類支持動態 striping 到 64bit 值上。

設計思路

這個類維護一個延遲初始的、原子地更新值的表，加上額外的 “base” 字段。表的大小是 2 的冪。索引使用每線程的哈希碼來masked。這個的幾乎所有聲明都是包私有的，通過子類直接訪問。

表的條目是 Cell 類，一個填充過（通過?sun.misc.Contended?）的 AtomicLong 的變體，用于減少緩存競爭。填充對于多數 Atomics 是過度殺傷的，因為它們一般不規則地分布在內存里，因此彼此間不會有太多沖突。但存在于數組的原子對象將傾向于彼此相鄰地放置，因此將通常共享緩存行（對性能有巨大的副作用），在沒有這個防備下。

部分地，因為Cell相對比較大，我們避免創建它們直到需要時。當沒有競爭時，所有的更新都作用到 base 字段。根據第一次競爭（更新 base 的 CAS 失敗），表被初始化為大小 2。表的大小根據更多的競爭加倍，直到大于或等于CPU數量的最小的 2 的冪。表的槽在它們需要之前保持空。

一個單獨的自旋鎖（“cellsBusy”）用于初始化和resize表，還有用新的Cell填充槽。不需要阻塞鎖，當鎖不可得，線程嘗試其他槽（或 base）。在這些重試中，會增加競爭和減少本地性，這仍然好于其他選擇。

通過 ThreadLocalRandom 維護線程探針字段，作為每線程的哈希碼。我們讓它們為 0 來保持未初始化直到它們在槽 0 競爭。然后初始化它們為通常不會互相沖突的值。當執行更新操作時，競爭和/或表沖突通過失敗了的 CAS 來指示。根據沖突，如果表的大小小于容量，它的大小加倍，除非有些線程持有了鎖。如果一個哈希后的槽是空的，且鎖可得，創建新的Cell。否則，如果槽存在，重試CAS。重試通過 “重散列，double hashing” 來繼續，使用一個次要的哈希算法（Marsaglia XorShift）來嘗試找到一個自由槽位。

表的大小是有上限的，因為，當線程數多于CPU數時，假如每個線程綁定到一個CPU上，存在一個完美的哈希函數映射線程到槽上，消除了沖突。當我們到達容量，我們隨機改變碰撞線程的哈希碼搜索這個映射。因為搜索是隨機的，沖突只能通過CAS失敗來知道，收斂convergence 是慢的，因為線程通常不會一直綁定到CPU上，可能根本不會發生。然而，盡管有這些限制，在這些案例下觀察到的競爭頻率顯著地低。

當哈希到特定 Cell 的線程終止后，Cell 可能變為空閑的，表加倍后導致沒有線程哈希到擴展的 Cell 也會出現這種情況。我們不嘗試去檢測或移除這些 Cell，在實例長期運行的假設下，觀察到的競爭水平將重現，所以 Cell 將最終被再次需要。對于短期存活的實例，這沒關系。

設計思路小結

• striping和緩存行填充：通過把類數據 striping 為 64bit 的片段，使數據成為緩存行友好的，減少CAS競爭。
• 分解表示：對于一個數字 5，可以分解為一序列數的和：2 + 3，這個數字加 1 也等價于它的分解序列中的任一 數字加 1：5 + 1 = 2 + (3 + 1)。
• 通過把分解序列存放在表里面，表的條目都是填充后的 Cell；限制表的大小為 2 的冪，則可以用掩碼來實現索引；同時把表的大小限制為大于等于CPU數量的最小的 2 的冪。
• 當表的條目上出現競爭時，在到達容量前表擴容一倍，通過增加條目來減少競爭。

CELL 類

Cell?類是?Striped64?的靜態內部類。通過注解?@sun.misc.Contended?來自動實現緩存行填充，讓Java編譯器和JRE運行時來決定如何填充。本質上是一個填充了的、提供了CAS更新的volatile變量。

@sun.misc.Contended static final class Cell {volatile long value;Cell(long x) { value = x; }final boolean cas(long cmp, long val) {return UNSAFE.compareAndSwapLong(this, valueOffset, cmp, val);}// Unsafe mechanicsprivate static final sun.misc.Unsafe UNSAFE;private static final long valueOffset;static {try {UNSAFE = sun.misc.Unsafe.getUnsafe();Class<?> ak = Cell.class;valueOffset = UNSAFE.objectFieldOffset(ak.getDeclaredField("value"));} catch (Exception e) {throw new Error(e);}} }

STRIPED64

Striped64 通過一個 Cell 數組維持了一序列分解數的表示，通過 base 字段維持數的初始值，通過 cellsBusy 字段來控制 resing 和/或 創建Cell。它還提供了對數進行累加的機制。

abstract class Striped64 extends Number {static final int NCPU = Runtime.getRuntime().availableProcessors();// 存放 Cell 的表。當不為空時大小是 2 的冪。transient volatile Cell[] cells;// base 值，在沒有競爭時使用，也作為表初始化競爭時的一個后備。transient volatile long base;// 自旋鎖，在 resizing 和/或 創建Cell時使用。transient volatile int cellsBusy; }

累加機制 longAccumulate

設計思路里針對機制的實現，核心邏輯。該方法處理涉及初始化、resing、創建新cell、和/或競爭的更新。

邏輯如下：

• if 表已初始化

  • if 映射到的槽是空的，加鎖后再次判斷，如果仍然是空的，初始化cell并關聯到槽。
  • else if （槽不為空）在槽上之前的CAS已經失敗，重試。
  • else if （槽不為空、且之前的CAS沒失敗，）在此槽的cell上嘗試更新
  • else if 表已達到容量上限或被擴容了，重試。
  • else if 如果不存在沖突，則設置為存在沖突，重試。
  • else if 如果成功獲取到鎖，則擴容。
  • else 重散列，嘗試其他槽。

• else if 鎖空閑且獲取鎖成功，初始化表

• else if 回退 base 上更新且成功則退出
• else 繼續

final void longAccumulate(long x, LongBinaryOperator fn,boolean wasUncontended) {int h;if ((h = getProbe()) == 0) {// 未初始化的ThreadLocalRandom.current(); // 強制初始化h = getProbe();wasUncontended = true;}// 最后的槽不為空則 true，也用于控制擴容，false重試。boolean collide = false;for (;;) {Cell[] as; Cell a; int n; long v;if ((as = cells) != null && (n = as.length) > 0) {// 表已經初始化if ((a = as[(n - 1) & h]) == null) {// 線程所映射到的槽是空的。if (cellsBusy == 0) { // 嘗試關聯新的Cell// 鎖未被使用，樂觀地創建并初始化cell。Cell r = new Cell(x);if (cellsBusy == 0 && casCellsBusy()) {// 鎖仍然是空閑的、且成功獲取到鎖boolean created = false;try { // 在持有鎖時再次檢查槽是否空閑。Cell[] rs; int m, j;if ((rs = cells) != null &&(m = rs.length) > 0 &&rs[j = (m - 1) & h] == null) {// 所映射的槽仍為空rs[j] = r; // 關聯 cell 到槽created = true;}} finally {cellsBusy = 0; // 釋放鎖}if (created)break; // 成功創建cell并關聯到槽，退出continue; // 槽現在不為空了}}// 鎖被占用了，重試collide = false;}// 槽被占用了else if (!wasUncontended) // 已經知道 CAS 失敗wasUncontended = true; // 在重散列后繼續// 在當前槽的cell上嘗試更新else if (a.cas(v = a.value, ((fn == null) ? v + x :fn.applyAsLong(v, x))))break;// 表大小達到上限或擴容了；// 表達到上限后就不會再嘗試下面if的擴容了，只會重散列，嘗試其他槽else if (n >= NCPU || cells != as)collide = false; // At max size or stale// 如果不存在沖突，則設置為存在沖突else if (!collide)collide = true;// 有競爭，需要擴容else if (cellsBusy == 0 && casCellsBusy()) {// 鎖空閑且成功獲取到鎖try {if (cells == as) { // 距上一次檢查后表沒有改變，擴容：加倍Cell[] rs = new Cell[n << 1];for (int i = 0; i < n; ++i)rs[i] = as[i];cells = rs;}} finally {cellsBusy = 0; // 釋放鎖}collide = false;continue; // 在擴容后的表上重試}// 沒法獲取鎖，重散列，嘗試其他槽h = advanceProbe(h);}else if (cellsBusy == 0 && cells == as && casCellsBusy()) {// 加鎖的情況下初始化表boolean init = false;try { // Initialize tableif (cells == as) {Cell[] rs = new Cell[2];rs[h & 1] = new Cell(x);cells = rs;init = true;}} finally {cellsBusy = 0; // 釋放鎖}if (init)break; // 成功初始化，已更新，跳出循環}else if (casBase(v = base, ((fn == null) ? v + x :fn.applyAsLong(v, x))))// 表未被初始化，可能正在初始化，回退使用 base。break; // 回退到使用 base} }

LongAdder

LongAdder 繼承自 Striped64，它的方法只針對簡單的情況：cell存在且更新無競爭，其余情況都通過 Striped64 的longAccumulate方法來完成。

public void add(long x) {Cell[] as; long b, v; int m; Cell a;if ((as = cells) != null || !casBase(b = base, b + x)) {// cells 不為空 或在 base 上cas失敗。也即出現了競爭。boolean uncontended = true;//if (as == null || (m = as.length - 1) < 0 ||(a = as[getProbe() & m]) == null ||!(uncontended = a.cas(v = a.value, v + x)))// 如果所映射的槽不為空，且成功更新則返回，否則進入復雜處理流程。longAccumulate(x, null, uncontended);} }// 獲取當前的和。base值加上每個cell的值。 public long sum() {Cell[] as = cells; Cell a;long sum = base;if (as != null) {for (int i = 0; i < as.length; ++i) {if ((a = as[i]) != null)sum += a.value;}}return sum; }

總結

以上是生活随笔為你收集整理的Java8 Striped64 和 LongAdder的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。