A:
This paper analyzes the source code of concurrentHashMap based on JDK1.8, and analyzes the code of expansion mechanism and size calculation method of concurrentHashMap with put() method as the entrance
2. ConcurrentHashMap member variable
// Maximum capacity
private static final int MAXIMUM_CAPACITY = 1 << 30;
// The default capacity
private static final int DEFAULT_CAPACITY = 16;
// Load factor
private static final float LOAD_FACTOR = 0.75 f;
// List length threshold for converting lists to red-black trees
static final int TREEIFY_THRESHOLD = 8;
// Convert a red-black tree to a list length threshold
static final int UNTREEIFY_THRESHOLD = 6;
// The node array length threshold for converting a list to a red-black tree
static final int MIN_TREEIFY_CAPACITY = 64;
// The default thread migration data range
private static final int MIN_TRANSFER_STRIDE = 16;
// Number of cpus on the current server
static final int NCPU = Runtime.getRuntime().availableProcessors();
// A real container for storing data
transient volatile Node<K,V>[] table;
// A new array for expansion
private transient volatile Node<K,V>[] nextTable;
// Used with counterCells to calculate the size of concurrentHashMap
private transient volatile long baseCount;
// The value -1 indicates that the node array is being initialized. After initialization, the value is assigned to the expansion threshold
private transient volatile int sizeCtl;
// Index of data migration
private transient volatile int transferIndex;
// The cas lock is required to calculate the size of concurrentHashMap
private transient volatile int cellsBusy;
// The size used to calculate concurrentHashMap the default length is 2
private transient volatile CounterCell[] counterCells;
Copy the codeConcurrentHashMap source code analysis
Put method as the entrance to the source code analysis
1. Flowchart of PUT method
2. Source code analysis
Put () calls putVal(). If the current array is not initialized, initTable() is first called to initialize the array, and then check whether the current subscript is NULL according to the calculated array subscript. If it is null, cas is used to ensure thread safety. If it is not null, hash conflicts need to be resolved. Split lists and red-black trees are handled separately.
A. Linked list: Traverses the linked list, if there is the same key to overwrite the operation, otherwise add to the end of the list (tail interpolation method).
B. Red-black tree: Traverse the red-black tree. If there is the same key, overwrite it. If there is not, add the nodes that build the red-black tree to the red-black tree, and ensure the balance of the red-black tree by left-rotation or right-rotation.
After the element is added, determine if the list length is greater than or equal to 8. If so, call treeifyBin().
Finally, addCount () is called to count the total number of elements in the array.
final V putVal(K key, V value, boolean onlyIfAbsent) {
// Neither key nor value can be null or a null pointer exception will be thrown
if (key == null || value == null) throw new NullPointerException();
// Computes the hash value of the key
int hash = spread(key.hashCode());
int binCount = 0;
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh; K fk; V fv;
// If TAB is empty, initialize the array first
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//(n-1) &hash the array index that the current value should hold
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
// If the node of the computed array position is null, replace it with cas
if (casTabAt(tab, i, null.new Node<K,V>(hash, key, value)))
break; // no lock when adding to empty bin
}
//MOVED Indicates that data is being migrated to the current node
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
//onlyIfAbsent True Indicates that the original value cannot be overwritten. The default value is False
else if (onlyIfAbsent // check first node without acquiring lock&& fh == hash && ((fk = f.key) == key || (fk ! =null&& key.equals(fk))) && (fv = f.val) ! =null)
// If the same key exists and value is not empty, return the existing value
return fv;
else {
// This code is the logic to handle hash conflicts
V oldVal = null;
// Lock the node
synchronized (f) {
if (tabAt(tab, i) == f) {
// For linked list processing
if (fh >= 0) {
binCount = 1;
for (Node<K,V> e = f;; ++binCount) {
K ek;
// If the list has the same key and can be overwritten, then overwrite the original value directly
// No CAS is required to ensure thread safety because locks are already in place
if(e.hash == hash && ((ek = e.key) == key || (ek ! =null && key.equals(ek)))) {
oldVal = e.val;
if(! onlyIfAbsent) e.val = value;break;
}
Node<K,V> pred = e;
// Traversing to the end of the list proves that the end method adds the list node
if ((e = e.next) == null) {
pred.next = new Node<K,V>(hash, key, value);
break; }}}// For red-black tree processing
else if (f instanceof TreeBin) {
Node<K,V> p;
binCount = 2;
//
// If there is a value with the same key that is allowed to be overwritten, the old value is overwritten or it is added to the tree according to the red-black tree rules
PutTreeVal () resolves hash collisions using the same logic as a list but involving left and right rotation of the tree
if((p = ((TreeBin<K,V>)f).putTreeVal(hash, key, value)) ! =null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
else if (f instanceof ReservationNode)
throw new IllegalStateException("Recursive update"); }}if(binCount ! =0) {
// Check whether the list length reaches the tree threshold (default: 8)
// Note that this does not mean that the threshold is reached and the tree will be treed
// If the array length is greater than 64 and the list length is greater than or equal to the threshold, the tree will be expanded first to reduce the list length
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if(oldVal ! =null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
}
Copy the codeThe initTable() method initializes the array. Cas is used to ensure that only one thread initializes the array.
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
if ((sc = sizeCtl) < 0)
// If sizeCtl is less than 0 (another thread has successfully modified sizeCtl, so -1 means we are initializing the array)
// Show that a thread is already performing capacity expansion
Thread.yield(); // lost initialization race; just spin
// If sizeCtl is set to -1 and the sizeCtl is set to -1, it can be initialized
else if (U.compareAndSetInt(this, SIZECTL, sc, -1)) {
try {
if ((tab = table) == null || tab.length == 0) {
int n = (sc > 0)? sc : DEFAULT_CAPACITY;@SuppressWarnings("unchecked")
// Initialize the array and assign to the table member variable
Node<K,V>[] nt = (Node<K,V>[])newNode<? ,? >[n]; table = tab = nt;//sc sets the threshold for array expansion to original capacity - original capacity /4;
sc = n - (n >>> 2); }}finally {
// sizeCtl will also become the array size threshold
sizeCtl = sc;
}
break; }}return tab;
}
Copy the codeThe treeifyBin() and addCount () methods are analyzed next.
TreeifyBin () method
The treeifyBin method is called when the list is larger than 8. First, the length of the current array is judged. If the array length is less than 64, the expansion operation is performed; if the list is larger than 64, the list is converted to a red-black tree.
Therefore, two conditions should be met to transform a linked list into a red-black tree:
1. The length of the linked list is greater than 8
2. The node array length is greater than 64
private final void treeifyBin(Node<K,V>[] tab, int index) {
Node<K,V> b; int n;
if(tab ! =null) {
// If the array length is less than 64, expand the array
if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
// By default, capacity expansion is twice the original capacity
tryPresize(n << 1);
else if((b = tabAt(tab, index)) ! =null && b.hash >= 0) {
If the array length is greater than 64, tree it
// Only nodes are locked and the granularity of the lock is strictly controlled
synchronized (b) {
// Convert the linked list to a red-black tree
if (tabAt(tab, index) == b) {
TreeNode<K,V> hd = null, tl = null;
for(Node<K,V> e = b; e ! =null; e = e.next) {
TreeNode<K,V> p =
new TreeNode<K,V>(e.hash, e.key, e.val,
null.null);
if ((p.prev = tl) == null)
hd = p;
else
tl.next = p;
tl = p;
}
// Set the array subscript value to the converted red-black tree reference
setTabAt(tab, index, new TreeBin<K,V>(hd));
}
}
}
}
}
Copy the codeIf the array length is less than 64, the tryPresize() method is called to expand the array
Capacity Expansion Flowchart
Source code analysis: First, calculate the capacity after expansion and determine whether the node array is initialized. If there is no initialization, initialize the node array first and then call transfer() method for expansion. ConcurrentHashMap is cleverly designed during expansion. If other linear call apis discover that concurrentHashMap is expanding capacity, helpTransfer is called to assist data migration and improve capacity expansion efficiency.
private final void tryPresize(int size) {
// Check whether the capacity is greater than half of the maximum capacity. If yes, the capacity is directly expanded to the maximum capacity
//tableSizeFor Used to ensure that an array is 2 to the NTH power
int c = (size >= (MAXIMUM_CAPACITY >>> 1))? MAXIMUM_CAPACITY : tableSizeFor(size + (size >>>1) + 1);
int sc;
while ((sc = sizeCtl) >= 0) {
Node<K,V>[] tab = table; int n;
// Initialize the array if it has not already been initialized
if (tab == null || (n = tab.length) == 0) {
n = (sc > c) ? sc : c;
// The lock replacement will be executed only when the cas is successful
// This code is basically the same as initTable()
if (U.compareAndSetInt(this, SIZECTL, sc, -1)) {
try {
if (table == tab) {
// Create a new array and assign it to the table member variable
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])newNode<? ,? >[n]; table = nt; sc = n - (n >>>2); }}finally{ sizeCtl = sc; }}}// If the capacity after expansion is greater than the maximum capacity, it is returned
else if (c <= sc || n >= MAXIMUM_CAPACITY)
break;
else if (tab == table) {
// Calculate the expansion stamp to ensure that each expansion is unique
int rs = resizeStamp(n);
// Use CAS to lock
if (U.compareAndSetInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
// True expansion method
transfer(tab, null); }}}Copy the codeAnalysis of transfer() method
The transfer() method is a true expansion method. It creates a new node array with twice the capacity of the original array, migrates the elements of the original array to the new array, and assigns the new array to the member variable table.
ConcurrentHashMap supports multiple threads to participate in data migration. Firstly, the scope of data migration for each thread is calculated. During data migration, the node currently migrated will be changed to ForwardingNode. Avoid repeated migrations.) Then, data migration is carried out for linked list and red-black tree respectively.
1. The list:
Traversing the list to calculate the subscript position of each node in the new array yields only two cases A. B. Calculate the index of the new array as (the index of the original array + the capacity of the original array). In case A, store the data in the low list; in case B, store the data in the high list and assign the value to the corresponding index of the new array.
2. Red and black tree: traverse the red-black tree, calculation of each node in the new array subscript position, calculation is the same as the results of the two kinds of results If it is a case, the data is low tree, if is b situation exists in the high tree, if high or low tree length less than 6, so will be high or low trees for the linked list, and then assign values to the new array respectively
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
The default range for a thread migration based on the number of cpus is 16
For an old array of 32, the first thread's first migration should be in the range 16-31
if ((stride = (NCPU > 1)? (n >>>3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
if (nextTab == null) { // initiating
// If the new array is empty, create a new array with twice the size of the original one
try {
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])newNode<? ,? >[n <<1];
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;
//transferIndex Indicates the start index of data migration. By default, the migration starts from the end of the original array
// So for an old array of 32, the first thread's first migration should be in the range 16-31
transferIndex = n;
}
int nextn = nextTab.length;
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
boolean advance = true;
// Indicates whether data migration is complete
boolean finishing = false; // to ensure sweep before committing nextTab
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
while (advance) {
int nextIndex, nextBound;
// The first three judgments are to determine whether the current thread has completed its scope to migrate data
// Advance = false breaks out of the while loop
if (--i >= bound || finishing)
advance = false;
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
else if (U.compareAndSetInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
bound = nextBound;
i = nextIndex - 1;
advance = false; }}if (i < 0 || i >= n || i + n >= nextn) {
int sc;
if (finishing) {
// Assign the new array to the table if the capacity expansion is complete
nextTable = null;
table = nextTab;
sizeCtl = (n << 1) - (n >>> 1);
return;
}
// Number of threads currently involved in data migration -1 Use CAS to ensure data security
if (U.compareAndSetInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
// If (sc-2)! = resizeStamp(n) << RESIZE_STAMP_SHIFT if the condition is true, it means that the migration has been completed
SizeCtl (this, sc, rs << RESIZE_STAMP_SHIFT) + 2)
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
return;
finishing = advance = true;
i = n; // recheck before commit}}else if ((f = tabAt(tab, i)) == null)
// If the node is empty, cas is used to replace it with FWD. If the node is FWD, it is a MOVE node
advance = casTabAt(tab, i, null, fwd);
else if ((fh = f.hash) == MOVED)
// If it is MOVE, it will skip because there is already a thread migrating and it does not need the current thread to migrate
advance = true; // already processed
else {
// Lock the node
synchronized (f) {
if (tabAt(tab, i) == f) {
// Save the high and low bits of the node separately
Node<K,V> ln, hn;
// Migrate data for linked lists
if (fh >= 0) {
int runBit = fh & n;
Node<K,V> lastRun = f;
for(Node<K,V> p = f.next; p ! =null; p = p.next) {
// Recalculate the array index of each node in the list. If the index is the same as the original, then put it in the lower chain. Otherwise, put it in the higher chain
int b = p.hash & n;
if (b != runBit) {
runBit = b;
lastRun = p;
}
}
if (runBit == 0) {
ln = lastRun;
hn = null;
}
else {
hn = lastRun;
ln = null;
}
for(Node<K,V> p = f; p ! = lastRun; p = p.next) {int ph = p.hash; K pk = p.key; V pv = p.val;
if ((ph & n) == 0)
// Build the low level linked list
ln = new Node<K,V>(ph, pk, pv, ln);
else
// Build a high-order list
hn = new Node<K,V>(ph, pk, pv, hn);
}
// Place the low list at the original index of the new array
setTabAt(nextTab, i, ln);
// Place the high-order list in the new array with subscript (original subscript position + original capacity size)
setTabAt(nextTab, i + n, hn);
// Setting the location of the old array to the FWD node indicates that the node has migrated data
setTabAt(tab, i, fwd);
advance = true;
}
// For red-black tree processing
else if (f instanceof TreeBin) {
// Do the same for the high and low linked list of data
TreeBin<K,V> t = (TreeBin<K,V>)f;
TreeNode<K,V> lo = null, loTail = null;
TreeNode<K,V> hi = null, hiTail = null;
int lc = 0, hc = 0;
for(Node<K,V> e = t.first; e ! =null; e = e.next) {
int h = e.hash;
TreeNode<K,V> p = new TreeNode<K,V>
(h, e.key, e.val, null.null);
if ((h & n) == 0) {
if ((p.prev = loTail) == null)
lo = p;
else
loTail.next = p;
loTail = p;
++lc;
}
else {
if ((p.prev = hiTail) == null)
hi = p;
elsehiTail.next = p; hiTail = p; ++hc; }}// If the length of the migrated high and low list is less than 6 (UNTREEIFY_THRESHOLD) then the original number structure will be changed to the list structureln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) : (hc ! =0)?newTreeBin<K,V>(lo) : t; hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) : (lc ! =0)?new TreeBin<K,V>(hi) : t;
// Place the low list at the original index of the new array
setTabAt(nextTab, i, ln);
// Place the high-order list in the new array with subscript (original subscript position + original capacity size)
setTabAt(nextTab, i + n, hn);
// Setting the location of the old array to the FWD node indicates that the node has migrated data
setTabAt(tab, i, fwd);
advance = true;
}
}
}
}
}
}
Copy the codehelpTransfer()
HelpTransfer is called to help expand concurrentHashMap if it is found to be expanding when other threads operate on concurrentHashMap
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
Node<K,V>[] nextTab; int sc;
// Further determine whether data migration is in progress and if so, assist data
if(tab ! =null && (f instanceofForwardingNode) && (nextTab = ((ForwardingNode<K,V>)f).nextTable) ! =null) {
int rs = resizeStamp(tab.length);
while (nextTab == nextTable && table == tab &&
(sc = sizeCtl) < 0) {
if((sc >>> RESIZE_STAMP_SHIFT) ! = rs || sc == rs +1 ||
sc == rs + MAX_RESIZERS || transferIndex <= 0)
// Other threads have migrated data and return directly
break;
// The number of threads responsible for the migration using CAS + 1
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
/ / call transfer
transfer(tab, nextTab);
break; }}return nextTab;
}
return table;
}
Copy the codeAddCount () method
1. The flow chart
2. Source code analysis
When calculating the current number of elements in the array, CounterCell array and baseCount are used for calculation. First, cas is used to replace the value of baseCount. If the value fails, the competition is fierce and CounterCell array is used for calculation. Each thread uses one element of the CounterCell array to sum up the total number, which is the sum of each element of the CounterCell array plus baseCount.
private final void addCount(long x, int check) {
CounterCell[] cs; long b, s;
// We use the CounterCell array to relieve cas of counting count
BaseCount = baseCount; baseCount = baseCount; baseCount = baseCount
if((cs = counterCells) ! =null| |! U.compareAndSetLong(this, BASECOUNT, b = baseCount, s = b + x)) {
CounterCell c; long v; int m;
boolean uncontended = true;
/ / use ThreadLocalRandom. GetProbe () randomly generated CounterCell array subscript position
// The array is not initialized or the array value is null or cas counts the number of nodes. On failure, call fullAddCount() to initialize and calculate the value
if (cs == null || (m = cs.length - 1) < 0 ||
(c = cs[ThreadLocalRandom.getProbe() & m]) == null| |! (uncontended = U.compareAndSetLong(c, CELLVALUE, v = c.value, v + x))) { fullAddCount(x, uncontended);return;
}
if (check <= 1)
return;
s = sumCount();
}
// This section is to determine whether the element needs to be expanded after adding elements
if (check >= 0) {
Node<K,V>[] tab, nt; int n, sc;
SizeCtl specifies the size threshold after the array is initialized. S >= sizeCtl specifies the size threshold
// The following code is the same as the expansion code
while (s >= (long)(sc = sizeCtl) && (tab = table) ! =null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
int rs = resizeStamp(n);
if (sc < 0) {
if((sc >>> RESIZE_STAMP_SHIFT) ! = rs || sc == rs +1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0)
break;
if (U.compareAndSetInt(this, SIZECTL, sc, sc + 1))
transfer(tab, nt);
}
else if (U.compareAndSetInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);
// Count the number of nodes in the arrays = sumCount(); }}}Copy the codefullAddCount()
First cas initialization CounterCell array, using ThreadLocalRandom. GetProbe () the generated random number and length calculation of CounterCell array subscript, count, using cas to compute if failure will CounterCell array capacity, And then compute it again.
private final void fullAddCount(long x, boolean wasUncontended) {
int h;
if ((h = ThreadLocalRandom.getProbe()) == 0) {
ThreadLocalRandom.localInit(); // force initialization
h = ThreadLocalRandom.getProbe();
wasUncontended = true;
}
boolean collide = false; // True if last slot nonempty
for (;;) {
CounterCell[] cs; CounterCell c; int n; long v;
// The first entry to the method counterCells must be null, so the if logic will not be followed
if((cs = counterCells) ! =null && (n = cs.length) > 0) {
// The subscript of the current thread CounterCell array is null
if ((c = cs[(n - 1) & h]) == null) {
if (cellsBusy == 0) { // Try to attach new Cell
// Initialize
CounterCell r = new CounterCell(x); // Optimistic create
//cas is locked for operation
if (cellsBusy == 0 &&
U.compareAndSetInt(this, CELLSBUSY, 0.1)) {
boolean created = false;
try { // Recheck under lock
CounterCell[] rs; int m, j;
if((rs = counterCells) ! =null &&
(m = rs.length) > 0 &&
rs[j = (m - 1) & h] == null) {
rs[j] = r;
created = true; }}finally {
cellsBusy = 0;
}
if (created)
break;
continue; // Slot is now non-empty
}
}
collide = false;
}
else if(! wasUncontended)// CAS already known to fail
wasUncontended = true; // Continue after rehash
// cas calculates count
else if (U.compareAndSetLong(c, CELLVALUE, v = c.value, v + x))
break;
//
else if(counterCells ! = cs || n >= NCPU) collide =false; // At max size or stale
else if(! collide) collide =true;
// The previous cas count failed, indicating that the competition is fierce and the counterCells array needs to be expanded
// the cellsBusy field is used for cas locking
else if (cellsBusy == 0 &&
U.compareAndSetInt(this, CELLSBUSY, 0.1)) {
try {
// Expand counterCells twice as large as before
if (counterCells == cs) // Expand table unless stale
counterCells = Arrays.copyOf(cs, n << 1);
} finally {
cellsBusy = 0;
}
collide = false;
// The capacity expansion succeeds and the next loop is displayed
continue; // Retry with expanded table
}
h = ThreadLocalRandom.advanceProbe(h);
}
// The first entry method must be 0 and counterCells should not be initialized with null
// Cas locks ensure that only one thread is available for initialization
else if (cellsBusy == 0 && counterCells == cs &&
U.compareAndSetInt(this, CELLSBUSY, 0.1)) {
boolean init = false;
try { // Initialize table
if (counterCells == cs) {
// An array of size 2 is created by default for the current thread
CounterCell[] rs = new CounterCell[2];
// Initialize the array subscript CounterCell required by the current thread
rs[h & 1] = new CounterCell(x);
counterCells = rs;
init = true; }}finally {
cellsBusy = 0;
}
if (init)
break;
}
// Add cas to baseCount if the calculation fails
else if (U.compareAndSetLong(this, BASECOUNT, v = baseCount, v + x))
break; // Fall back on using base}}Copy the codesumCount()
The sum of each element of the CounterCell array + baseCount is the size of the current concurrentHashMap
final long sumCount(a) {
CounterCell[] as = counterCells; CounterCell a;
long sum = baseCount;
if(as ! =null) {
for (int i = 0; i < as.length; ++i) {
if((a = as[i]) ! =null) sum += a.value; }}return sum;
}
Copy the codeFour: Concluding remarks
When reading long methods, it is recommended to view them in sections, such as transfer() method and fullAddCount() method. There are many if and else methods, which will be easier to understand if we read them in sections.