preface
HashMap source code has been read before, implement HashMap source code. However, HashMap is thread-unsafe and ConcurrentHashMap is thread-safe, so I decided to investigate ConcurrentHashMap. Sometimes a method needs to be studied for at least several days, opening up an ocean of knowledge. Last year, I finally saw the complete put method, the ranch, to expand there. It was delayed by job changes and interviews, but now I’m picking it up again, mainly because I’ve just written an interpretation of the source code for HashMap. Well, I’ve been learning about distributed transactions in parallel lately. In other words, it is not so easy to get tired of some. Before reading this article, it is recommended to take a look at the basics of HashMap, because many of the principles of Java8 ConcurrentHashMap are the same as those of HashMap. 👉 Java8 HashMap source code interpretation
Anyway, after studying this, I discovered one thing. It seems that for high-concurrency scenarios, no matter the ConcurrentHashMap or the distributed system, from a macro perspective, the basic idea of solution is the idea of divide and conquer, just like yu taming water, blocking is better than draining.
An overview of the
ConcurrentHashMap is a thread-safe set of maps under the Concurrent package. HashMap is thread-safe, HashTable is thread-safe, but it locks the entire hash table. The lock scope is too large.
The ConcurrentHashMap Java 7
structure
In the case of multiple cores and cpus, multiple threads can run in parallel, but when it comes to Hashtables, things change. Since the entire HashTable is locked, only one thread can hold the lock at a time, while the other threads wait. Instead of doing more than one thing at a time, you can only do one thing at a time. ConcurrentHashMap 1.7 reduces the lock scope. Instead of locking the entire table at a time, the lock scope is divided into multiple blocks, which means that multiple threads can work at the same time.
ConcurrentHashMap is different. When putting an element: 1. Hash the key to get a subscript 2 of Segment[]. Segment [I] lock 2.1 If the lock cannot be obtained, the thread will try to obtain the lock for a certain number of times. After the lock cannot be obtained for a certain number of times, the thread will enter the blocking state and wait. 3. Start to put the hashEntry of segment[I]. 3.1Segment inherited from ReentrantLock, exclusive ReentrantLock. So we don’t need to worry about concurrency when we get the Segment, because the Segment is an exclusive lock.
Segment[] indicates the number of concurrent threads. 16 segments indicate that 16 threads can run in parallel at the same time. As the amount of concurrency increases, even if there is 16 concurrency, it is not enough for the large number of threads, and a large number of threads will still be blocked.
So there is a 1.8 optimization that just ditches segment.
The ConcurrentHashMap java8
As shown in figure 1.8, the structure of ConcurrentHashMap is the same as that of HashMap, which is a simple array of linked lists. The basic calculation methods of hash, addressing, and list splitting during expansion are the same as that of HashMap. There is no Segment[], so its concurrency is the length of the array. In this case, without sere-entry locks, how to ensure thread safety, how to ensure the thread safety of PUT and expansion? The solution lies in the use of CAS and control of threads.
Pre-knowledge point
Some things to know before you start parsing concurrentHashMap
CAS
CAS is to compare and swap. First, the thread reads the value a of X from the memory, and then sets the value a to the value of x in the current memory, which is set as B. Compare a and B. If the expected values of A and B are the same, change the value of x in the memory to C. If they are different, do not change the value. This is an optimistic locking strategy, no locking.
In the case of multi-threading,CAS may cause ABA problem: if X value in memory is 1, thread A gets X value of 1, then thread B preempts the CPU time slice and changes X in memory to 2 and then 1. Thread A comes back and finds the value of X is still 1, so it changes it to 3. The ABA problem may not seem like a big deal, but in a program, the same value does not mean there is no change, such as the memory address may change.
· There are two ways to solve this problem: (1) make the comparison modification part of the operation has atomic, such as in the second kill scenario, using Lua script operation Redis to modify data, because Redis only supports optimistic locks, Lua program execution is atomic. ② Add the version number. For each change, the version number is increased by 1.
Memory model for threads
The Java thread memory model is shown below: ↓
First, the Java memory model specifies that all variables are stored in main memory, and each thread has its own working memory. All operations are performed in the working memory, and you cannot directly manipulate main memory or other threads’ working memory.
A thread modifies a variable by first modifying the value in working memory and then writing it to main memory (when is unknown). For multiple threads to Shared variables, then, if a thread, thread b common operation variable y, thread a modified y variable value of 2, has not yet been written into the memory, then thread b can direct manipulation is himself first working memory, then his work y values in the memory is not 2, so how to make the thread b know the value of the variable y has been turned into 2? · For this problem, Java provides a keyword volatile to make shared variables visible. When thread 1 changes the value of x in working memory, it forces the value of x to be written to main memory. Second, thread 1’s modification operation will cause the cache of X in thread 2’s working memory to be invalid, so when thread 1 wants to read X, it will read it in main memory.
Note that the volatile keyword guarantees visibility, but it does not guarantee atomicity: · For example, a shared variable y has a value of 1. · Thread 1 is the first to obtain the value 1 of the shared variable Y. · At this time, thread 2 preempts the CPU time slice to obtain the shared variable Y, and modifies the value of Y by adding 1 to equal 2, and then refreshes the value of Y in main memory; · Then thread 1 will add the value of Y obtained before by 1 to become 2, and refresh main memory y value to become 2; · Technically,y is a shared variable, and both threads increment it by 1. The value in main memory should be 3. However, the thread’s addition of 1 to the shared variable Y does not have atomicity, leading to the error of the final result. If you think about it, it looks like the isolation level of a transaction.
Concurrency and parallelism of threads
Concurrency concurrency is when multiple threads execute alternately on a single CPU. It looks like multiple threads are working at the same time, but they switch quickly.
Parallelism means doing two things at the same time. In the case of multiple cores and cpus, multiple threads may be distributed among different logical processors that do not need to compete for CPU resources.
Generally open multithreading, the number of threads is also exquisite, such as my computer is 1 CPU, 4 cores, 8 logical processors, a logical processor a thread, indicating that 1 kernel can support 2 super threads. So 8 threads is more appropriate.
Java stack memory and heap memory
The Java language is known for its security. why? The reason is simple: Java doesn’t have Pointers and doesn’t allow us to manipulate memory directly. Even with the Unsafe class, we usually don’t get it directly, we need to get it through reflection.
Source code & principle analysis
The core operation methods and core global variables that you must know before parsing the PUT method
Relying on the class
The Unsafe class
Unsafe takes an important place in classes under the java.util. connCurrent package, where most CAS operations are performed. The Unsafe class gives Java the ability to manipulate memory Spaces as Pointers do in THE C language. Unsafe directly manipulates memory, making it faster and providing greater efficiency in the context of concurrency. But it also means that it is unsafe, not managed by the JVM, can’t be GC, and needs to be released by itself, and a memory leak can happen if you’re not careful. Unsafe most methods in the Unsafe class are modified native, meaning that these methods call the C implementation interface.
/** * Compares whether the address at address offset L relative to o is equal to o1, if so, changes the value to O2, and returns true. * No, return false * params: * O - the object instance of the comparison exchange * L - the offset based on the O address * O1 - Expected value of o in main memory * O2 - Modified value */
public final native boolean compareAndSwapObject(java.lang.Object o, long l, java.lang.Object o1, java.lang.Object o2);
/** * this method differs from the previous one in that it is an int */
public final native boolean compareAndSwapInt(java.lang.Object o, long l, int i, int i1);
/** * The difference between this method and the previous method is that the comparison modification is a long */
public final native boolean compareAndSwapLong(java.lang.Object o, long l, long l1, long l2);
/** * get the value of offset L */ in main memory o object
public native java.lang.Object getObjectVolatile(java.lang.Object o, long l);
/** * the offset L of object o is changed to o1 and immediately visible to other threads */
public native void putObjectVolatile(java.lang.Object o, long l, java.lang.Object o1);
/** * get the address of the first element of the array, i.e. the address of the first element of the array. * /
public native int arrayBaseOffset(java.lang.Class
aClass);
/** * returns the number of bytes used by an array element, such as int[], which is 4 bytes long and returns 4 */
public native int arrayIndexScale(java.lang.Class
aClass);
Copy the codeThe use of Unsafe in concurrentHashMap
Unsafe
// Unsafe mechanics
private static final sun.misc.Unsafe U;
//sizeCtl offset
private static final long SIZECTL;
// The offset of transferIndex
private static final long TRANSFERINDEX;
//baseCount offset
private static final long BASECOUNT;
//cellsBusy offset
private static final long CELLSBUSY;
//cellValue offset
private static final long CELLVALUE;
// The offset of the array header
private static final long ABASE;
1<
private static final int ASHIFT;
static {
try{ U = sun.misc.Unsafe.getUnsafe(); Class<? > k = ConcurrentHashMap.class;// Get the sizeCtl field offset
SIZECTL = U.objectFieldOffset
(k.getDeclaredField("sizeCtl"));
TRANSFERINDEX = U.objectFieldOffset
(k.getDeclaredField("transferIndex"));
BASECOUNT = U.objectFieldOffset
(k.getDeclaredField("baseCount"));
CELLSBUSY = U.objectFieldOffset
(k.getDeclaredField("cellsBusy")); Class<? > ck = CounterCell.class; CELLVALUE = U.objectFieldOffset (ck.getDeclaredField("value")); Class<? > ak = Node[].class; ABASE = U.arrayBaseOffset(ak);int scale = U.arrayIndexScale(ak);
if ((scale & (scale - 1)) != 0)
throw new Error("data type scale not a power of two");
/** * It is well known that the int type is 4 bytes, and one byte is 8 bits binary. 4 bytes is 32 binary * Integer numberOfLeadingZeros (n) * this method returns the 32 binary, count from left to right, to n between the highest how many 0; N =8, the method * returns 29; * 31 - Integer. NumberOfLeadingZeros (scale), can be obtained from right to left to n * between highest how many bits. * 31-Integer.numberOfLeadingZeros(8)=3 */
ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);
} catch (Exception e) {
throw newError(e); }}Copy the codeWith ABASE, ASHIFT is used to get an offset of an array element, and to access and manipulate array elements based on the offset. ABASE is the first address of the array. ASHIFT means the number of bytes between the array elements. Here’s an example: ASHIFT=3, Subscript elements for I = 3 address is 16 + (3-0) 16 + 3 x 8 x 8 = = = ABSE + 40 (I – 0) x (1 + I < < < < ASHIFT) = ABASE ASHIFT
Ps: in the sun’s source code, only the computer binary operations, such as &, |, ^, < <, > > >, because compared with x, the present, the binary operation is much faster. So ASHIFT exists only to facilitate binary arithmetic
Important shared variables
SizeCtl variable
This variable is expanded throughout the ConcurrentHashMap, and I was stuck trying to figure it out.
SizeCtl is a multithreaded shared variable. -1 means there are threads initializing arrays. If the value is greater than 0, the threshold for triggering expansion is recorded. Sizecl-1 means there are (-sizecl-1) threads in the addCount method. If the number of threads is smaller than -1, the addCount method is sizecl-1. Even contradicting that statement, I always thought it was a pit. It’s killing my train of thought to cross that out. If the value is less than -1, some threads are expanding: 👈
/** * Table initialization and resizing control. When negative, the * table is being initialized or resized: -1 for initialization, * else -(1 + the number of active resizing threads). Otherwise, * when table is null, holds the initial table size to use upon * creation, or 0 for default. After initialization, holds the * next element count value upon which to resize the table. */
private transient volatile int sizeCtl;
Copy the codeThe table [] variables
The array object currently in use
/** * The array of bins. Lazily initialized upon first insertion. * Size is always a power of two. Accessed directly by iterators. */
transient volatile Node<K,V>[] table;
Copy the codeNextTab [] variables
The new array object pointed to during expansion is null after expansion
/** * The next table to use; non-null only while resizing. */
private transient volatile Node<K,V>[] nextTable;
Copy the codeBaseCount variable
After the put success by CAS (Unsafe.com pareAndSwapLong () method) on the Unsafe, if the CAS fails, that there are other threads competition, then directly use LongAdder methods to count, this involves another variable counterCells
/** * Base counter value, used mainly when there is no contention, * but also as a fallback during table initialization * races. Updated via CAS. */
private transient volatile long baseCount;
Copy the codeCounterCells variable
Counter array, the idea is longAdder principle. The details of the addCount() method will be explained below.
Anyway, concurrentHashMap
/** * Table of counter cells. When non-null, size is a power of 2. */
private transient volatile CounterCell[] counterCells;
Copy the codetransferIndex
In expansion, the maximum subscript range of the list (array of lists) to be processed +1, or the number of lists to be processed. For example, if the old array length is 16 and nexttab. length=32, then the total range of nodes to be processed is from the first to the 32nd, i.e. transferIndex=32.
/** * The next table index (plus one) to split while resizing. */
private transient volatile int transferIndex;
Copy the codeThe meaning of ConcurrentHashMap. Node. Some hash value
· -1 The linked list of this node is being expanded · -2 is the root node of a red-black tree
/* * Encodings for Node hash fields. See above for explanation. */
static final int MOVED = -1; // hash for forwarding nodes
static final int TREEBIN = -2; // hash for roots of trees
Copy the codeSeveral subclasses of the inner class Node
ConcurrentHashMap. Node. Java classes
Inherited from Map.Entry. Normally the element node of ConcurrentHashMap. Such nodes generally have a hash value greater than 0.
static class Node<K.V> implements Map.Entry<K.V> {
final int hash;
final K key;
volatile V val;
volatile Node<K,V> next;
Node(int hash, K key, V val, Node<K,V> next) {
this.hash = hash;
this.key = key;
this.val = val;
this.next = next;
}
public final K getKey(a) { return key; }
public final V getValue(a) { return val; }
public final int hashCode(a) { return key.hashCode() ^ val.hashCode(); }
public final String toString(a){ return key + "=" + val; }
public final V setValue(V value) {
throw new UnsupportedOperationException();
}
public final boolean equals(Object o) { Object k, v, u; Map.Entry<? ,? > e;return ((o instanceofMap.Entry) && (k = (e = (Map.Entry<? ,? >)o).getKey()) ! =null&& (v = e.getValue()) ! =null &&
(k == key || k.equals(key)) &&
(v == (u = val) || v.equals(u)));
}
/** * Virtualized support for map.get(); overridden in subclasses. */
Node<K,V> find(int h, Object k) {
Node<K,V> e = this;
if(k ! =null) {
do {
K ek;
if(e.hash == h && ((ek = e.key) == k || (ek ! =null && k.equals(ek))))
return e;
} while((e = e.next) ! =null);
}
return null; }}Copy the codeConcurrentHashMap. ForwardingNode. Java classes
The head Node is converted to a ForwardingNode type during expansion of a slot linked list. The hash value of ForwardingNode is -1.
The specific operation is when the data of a card slot has been moved to the new array, but the overall expansion process is not complete. Slots in the old array that have moved data are set to type ForwardingNode, which holds a nextTable object (a reference to the new array object). When a thread needs to get data from the slot in the old array, it can reference nextTable in ForwardingNode to the new array.
/** * A node inserted at head of bins during transfer operations. */
static final class ForwardingNode<K.V> extends Node<K.V> {
final Node<K,V>[] nextTable;
ForwardingNode(Node<K,V>[] tab) {
super(MOVED, null.null.null);
this.nextTable = tab;
}
Node<K,V> find(int h, Object k) {
// loop to avoid arbitrarily deep recursion on forwarding nodes
// The new array object is being expanded
outer: for (Node<K,V>[] tab = nextTable;;) {
Node<K,V> e; int n;
if (k == null || tab == null || (n = tab.length) == 0 ||
(e = tabAt(tab, (n - 1) & h)) == null)
return null;
// The slot list or tree where the node is located is not null in the new array
for (;;) {
int eh; K ek;
// The list or tree of the slot has been moved
if((eh = e.hash) == h && ((ek = e.key) == k || (ek ! =null && k.equals(ek))))
return e;
// Hash <0, there are two cases: 1: expansion is being performed, that is, ForwardingNode type, 2: red-black tree, hash=-2
if (eh < 0) {
// Skip the next loop until the slot has been shifted
if (e instanceof ForwardingNode) {
tab = ((ForwardingNode<K,V>)e).nextTable;
continue outer;
}
/ / the red-black tree
else
return e.find(h, k);
}
// Find the element at the end
if ((e = e.next) == null)
return null; }}}}Copy the codeConcurrentHashMap.TreeBin.java
An entire red-black Tree references the object, holds root, and maintains a sequential list of nodes, much like the B+Tree, where all nodes are in the leaf and form a linked list. All the operations on the red-black tree are in here.
static final class TreeBin<K.V> extends Node<K.V> {
TreeNode<K,V> root;
/ / head
volatile TreeNode<K,V> first;
volatile Thread waiter;
volatile int lockState;
// values for lockState
static final int WRITER = 1; // set while holding write lock
static final int WAITER = 2; // set when waiting for write lock
static final int READER = 4; // increment value for setting read lock
// There is a lot of code to skip. }Copy the codeI really do not want to post so long code, the front of a red black tree principle 👉 red black tree principle and algorithm
ConcurrentHashMap.TreeNode.java
TreeNode is used to maintain a single node in a red-black tree.
/** * Nodes for use in TreeBins */
static final class TreeNode<K.V> extends Node<K.V> {
TreeNode<K,V> parent; // red-black tree links
TreeNode<K,V> left;
TreeNode<K,V> right;
TreeNode<K,V> prev; // needed to unlink next upon deletion
boolean red;
TreeNode(int hash, K key, V val, Node<K,V> next,
TreeNode<K,V> parent) {
super(hash, key, val, next);
this.parent = parent;
}
Node<K,V> find(int h, Object k) {
return findTreeNode(h, k, null);
}
/** * Returns the TreeNode (or null if not found) for the given key * starting at given root. */
final TreeNode<K,V> findTreeNode(inth, Object k, Class<? > kc) {
if(k ! =null) {
TreeNode<K,V> p = this;
do {
int ph, dir; K pk; TreeNode<K,V> q;
TreeNode<K,V> pl = p.left, pr = p.right;
if ((ph = p.hash) > h)
p = pl;
else if (ph < h)
p = pr;
else if((pk = p.key) == k || (pk ! =null && k.equals(pk)))
return p;
else if (pl == null)
p = pr;
else if (pr == null)
p = pl;
else if((kc ! =null|| (kc = comparableClassFor(k)) ! =null) && (dir = compareComparables(kc, k, pk)) ! =0)
p = (dir < 0)? pl : pr;else if((q = pr.findTreeNode(h, k, kc)) ! =null)
return q;
else
p = pl;
} while(p ! =null);
}
return null; }}Copy the codeA constructor
This part is consistent with HashMap, I will post the source code, see me parse Java8 HashMap source article 😂
,Parameterless constructor
/** * Creates a new, empty map with the default initial table size (16). */
public ConcurrentHashMap(a) {}Copy the code,Parameter constructor 1
Unlike HashMap, the initial array length is assigned to sizeCtl
/**
* Creates a new, empty map with an initial table size
* accommodating the specified number of elements without the need
* to dynamically resize.
*
* @param initialCapacity The implementation performs internal
* sizing to accommodate this many elements.
* @throws IllegalArgumentException if the initial capacity of
* elements is negative
*/
public ConcurrentHashMap(int initialCapacity) {
if (initialCapacity < 0)
throw new IllegalArgumentException();
int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1))? MAXIMUM_CAPACITY : tableSizeFor(initialCapacity + (initialCapacity >>>1) + 1));
this.sizeCtl = cap;
}
Copy the codeTableSizeFor Method description
,Parameterized constructor 2
/**
* Creates a new map with the same mappings as the given map.
*
* @param m the map
*/
public ConcurrentHashMap(Map<? extends K, ? extends V> m) {
this.sizeCtl = DEFAULT_CAPACITY;
putAll(m);
}
Copy the code· tableSizeFor
For the same principle as HashMap, see the article parsing the source code for Java8 HashMap
/** * Returns a power of two table size for the given desired capacity. * See Hackers Delight, SEC 3.2 */
private static final int tableSizeFor(int c) {
int n = c - 1;
n |= n >>> 1;
n |= n >>> 2;
n |= n >>> 4;
n |= n >>> 8;
n |= n >>> 16;
return (n < 0)?1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
}
Copy the codePut method
Heavy, really complicated, just this one method and the method I called for a long time.
put(key,value)
/**
* Maps the specified key to the specified value in this table.
* Neither the key nor the value can be null.
*
* <p>The value can be retrieved by calling the {@code get} method
* with a key that is equal to the original key.
*
* @param key key with which the specified value is to be associated
* @param value value to be associated with the specified key
* @return the previous value associated with {@code key}, or
* {@code null} if there was no mapping for {@code key}
* @throws NullPointerException if the specified key or value is null
*/
public V put(K key, V value) {
return putVal(key, value, false);
}
Copy the codeinitTable()
SizeCtl, the global shared variable used. Function: Initialize arrays, and sizeCtl variables to trigger expansion thresholds
/** * Initializes table, using the size recorded in sizeCtl. */
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
// The CAS operation flag begins with a while loop that keeps trying to initialize the array while it has not been initialized
while ((tab = table) == null || tab.length == 0) {
/**sizeCtl<0, 1)-1: initializing; 2)<-1: Expansion * From the judgment condition of the while loop in the previous step, it can be seen that there are other threads in the initialization process. * So the current thread gives in and waits for another thread to complete the initialization. * /
if ((sc = sizeCtl) < 0)
Thread.yield(); // lost initialization race; just spin
/** Check whether sizeCtl in main memory is the same as the value obtained in the previous step. If yes, no other thread is initializing. * if no, another thread is initializing. Continue with */
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
// Check again to see if any other threads have completed initialization
if ((tab = table) == null || tab.length == 0) {
/** The initial size of ConcurrentHashMap when the argument constructor 1 is called */
int n = (sc > 0)? sc : DEFAULT_CAPACITY;@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])newNode<? ,? >[n]; table = tab = nt; sc = n - (n >>>2); }}finally {
/** * If the try code block is successfully executed, the shared variable value is changed to trigger capacity expansion threshold; If the try block fails, the sizeCtl values will be restored
sizeCtl = sc;
}
break; }}return tab;
}
Copy the codecasTabAt(tab,i,c,v)
Using Unsafe, CAS changes the TAB [I] value
static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,
Node<K,V> c, Node<K,V> v) {
return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
}
Copy the codetabAt(tab,i)
Using Unsafe, retrieve the value of TAB [I] in main memory
static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {
return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
}
Copy the codeputVal(key,value,onlyIfAbsent)
/** Implementation for put and putIfAbsent * onlyIfAbsent False - Implementation for put and putIfAbsent * onlyIfAbsent False - Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
// Computes the hash value
int hash = spread(key.hashCode());
//
int binCount = 0;
// The outermost loop, CAS required
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
// If the array is not initialized
if (tab == null || (n = tab.length) == 0)
tab = initTable();
// Hash addresses I to null
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//CAS operation. If TAB [I]==null, add a new node at this location
if (casTabAt(tab, i, null.new Node<K,V>(hash, key, value, null)))
// Add successfully, out of the loop
break; // no lock when adding to empty bin
}
// The current list is marked as moved, indicating that the current array is being expanded
else if ((fh = f.hash) == MOVED)
// The current thread helps with helpTransfer
tab = helpTransfer(tab, f);
// A linked list exists at TAB [I] and is not being expanded
else {
V oldVal = null;
// Add elements to the list to ensure atomicity
synchronized (f) {
/** check whether TAB [I] is in the same table as f */ ** check whether TAB [I] is in the same table
if (tabAt(tab, i) == f) {
// The hash value of the node is greater than 0
if (fh >= 0) {
// Number of linked list nodes
binCount = 1;
for (Node<K,V> e = f;; ++binCount) {
K ek;
// If the hash, key, and value values of this object are the same as the key-value to be inserted
if(e.hash == hash && ((ek = e.key) == key || (ek ! =null && key.equals(ek)))) {
oldVal = e.val;
/ / cover
if(! onlyIfAbsent) e.val = value;break;
}
Node<K,V> pred = e;
// If the end of the list has been traversed, insert the new node directly into the end of the list
if ((e = e.next) == null) {
pred.next = new Node<K,V>(hash, key,
value, null);
break; }}}// If it is a tree
else if (f instanceof TreeBin) {
Node<K,V> p;
binCount = 2;
if((p = ((TreeBin<K,V>)f).putTreeVal(hash, key, value)) ! =null) {
oldVal = p.val;
if(! onlyIfAbsent) p.val = value; }}}}// If the number of nodes is greater than TREEIFY_THRESHOLD, the list becomes a red-black tree
if(binCount ! =0) {
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if(oldVal ! =null)
return oldVal;
break; }}}/** * concurrentHashMap number of elements plus 1 * binCount in addCount() is an identifier. If the number of competing threads is greater than 1, check whether expansion is required. */
addCount(1L, binCount);
return null;
}
Copy the codeSpread (key.hashcode ()) = spread(key.hashcode ()) = spread(key.hashcode ()) = spread(key.hashcode ()))
· An outer loop, CAS is necessary, convenient to restart after failure ~(~ ▽ ~)~*
· Performed in the circulatory body
· If the array is not initialized, call the initTable method and continue with the next loop
· If null at subscript I is obtained by hash and length operation (tabAt method), insert the new node at TAB [I] using CAS operation (casTabAt method), and continue the next cycle if CAS operation fails; Success breaks out of the cycle.
· If the linked list at TAB [I] is being processed for capacity expansion, the current thread enters the helpTransfer() method.
· All that remains is that the array is initialized and not being expanded, and the TAB [I] position already has a linked list/tree; The next logical step is to insert the new node into the list/tree
· Thread locks are first applied to the head/root of the list. After all, TAB [] is a shared variable, and volatile trim is used to ensure immediate visibility, but not atomicity (see thread memory model). · Determine the hash value of the node at TAB [I] (see the meaning of the special hash value of the node), if it is a chain phenotype, directly look back from the head, if it is a tree, Then call the method of inserting red-black tree · determine whether the chain phenotype needs to be converted to a red-black tree · call addCount
About the code in the U.CompareAndSwap* method at the beginning of the specific roleThe unsafe classRelevant methods
ConcurrentHashMap counter
ConcurrentHashMap’s counter is implemented in the same way as LongAdder’s. Its counter is used to count elements in multithreaded situations.
Counter general idea ：
It consists of two things:baseCount ,countCell[]
,At first, CAS is used when threads are not in contentionbaseCountIncrease value;
,When the CAS operation fails, it indicates that a thread is competing, and the thread starts the operationcountCell[] ;
,The thread generates a random number r, calculated with the array lengthr&(length-1)conclusionThe subscript IforcountCell[i]Add or subtract. It used to be multiple threads operating on onebaseCountThe stress ofcountCell[]In the array, divide and conquer!
,To get the total, just add all the values of baseCount and countCell[] ==>
AddCount (x, check) method
ConcurrentHashMap () function: Count the number of ConcurrentHashMap elements. Count, if the number reaches the threshold, expansion will be triggered. If a thread is already expanding capacity, the current thread also helps expand capacity. After capacity expansion is completed, the calculation value is counted again. The reason is that some threads are found in putVal step to help capacity expansion, and helpTransfer method is directly used to help capacity expansion, and put element is not continued until the completion of capacity expansion. Parameter: long x => Number of new elements added
Int check => If < 0, do not check whether expansion is required. If <=1, check whether capacity expansion is required when thread contention is not fierce.
In fact, I see that the logic of the code is that if the thread competition is fierce, after counting countCell[], return directly without judging check and expansion detection. Why is that? Well, it makes me curious. Check =1 indicates that there is only one node in the linked list at TAB [I]. Check =2 indicates that TAB [I] may be a red-black tree or a linked list with two nodes; Check >1 represents a list with multiple nodes at TAB [I]. If a thread enters the fullAddCount method, it is particularly competitive (high concurrency). The large range is in the counting cycle, if the expansion is detected after the completion of counting, the high concurrency will continue to the expansion, the program continues to be under great pressure. As we all know, high concurrency is not a good thing, it takes up a lot of resources, and we want to avoid it whenever possible. Therefore, return directly after the count is complete to end the high concurrency as soon as possible. Wait until the first thread after the high concurrency ends to check whether expansion is needed.
ThreadLocalRandom
Function: generate random number in multi-threaded environment. Instead of multiple threads competing for a seed from Random, ThreadLocalRandom monopolizes a seed from each thread, thus greatly improving performance.
ThreadLocalRandom.getProbe()
If ThreadLocalRandom does not execute localInit, the Thread’s probe value is 0(the current Thread has not generated a random number).
About the code in the U.CompareAndSwap* method at the beginning of the specific roleThe unsafe classRelevant methods
/** * Adds to count, and if table is too small and not already * resizing, initiates transfer. If already resizing, helps * perform transfer if work is available. Rechecks occupancy * after a transfer to see if another resize is already needed * because resizings are lagging additions. * *@param x the count to add
* @param check if <0, don't check resize, if <= 1 only check if uncontended
*/
private final void addCount(long x, int check) {
CounterCell[] as; long b, s;
// If CountCells[](counter array is not null) or CAS fails to modify baseCount 👉 concurrency exists
if((as = counterCells) ! =null| |! U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
CounterCell a; long v; int m;
// No race flags for multiple threads
boolean uncontended = true;
/** * CAS failed to modify as[I] when the counter array is not initialized or the slot of the counter array is null * or the array subscript I is not null. Enter the fullAddCount method */
if (as == null || (m = as.length - 1) < 0 ||
(a = as[ThreadLocalRandom.getProbe() & m]) == null| |! (uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) { fullAddCount(x, uncontended);// It indicates that there are too many threads and the program is busy
return;
}
/** baseCount * /** baseCount * /** baseCount * Check =1 indicates that a linked list with only 1 nodes has been added to the array of * TAB [], indicating that no * collisions have occurred and concurrency exists. Therefore, the capacity expansion of * */ is not detected for the time being
if (check <= 1)
return;
/** * count total */
s = sumCount();
}
/** * no concurrency. The program is not busy, you can perform capacity expansion detection. * /
if (check >= 0) {
Node<K,V>[] tab, nt; int n, sc;
// The current number of data reaches the threshold, the array is initialized, and the array length does not reach the maximum length
while (s >= (long)(sc = sizeCtl) && (tab = table) ! =null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
/ * * * using Integer. NumberOfLeadingZeros (n) | 1 < < 15 generated a 16th is the value of 1. * for example: 0000 0000 0000 0000 1000 0000 0010 1101 */
int rs = resizeStamp(n);
// concurrentHashMap is being expanded
if (sc < 0) {
/** * RESIZE_STAMP_SHIFT=16 * (sc >>> RESIZE_STAMP_SHIFT)! * SC == RS +1, SC == RS + MAX_RESIZERS: Well (▔, ▔)ㄏ,sc= =sc+1,sc=sc+MAX_RESIZERS * (nt=nextTable)== NULL, transferIndex<=0: expansion end flag */
if((sc >>> RESIZE_STAMP_SHIFT) ! = rs || sc == rs +1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0)
break;
// The current thread is added to the expansion process. Sc = SC +1
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
transfer(tab, nt);
}
/** * RESIZE_STAMP_SHIFT =16 * It is well known that the int data type takes up 4 bytes of 32-bit binary, and identifies the symbol bit in the computer's highest bit, 1-negative, 0-positive. * When the first thread expands, it changes the sizeCtl 16 bits high to generate a token stamp, with a maximum of 1(i.e., negative) since bit 16 is 1. SizeCtl is a negative number in *. * /
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);
/** The first thread starts capacity expansion. Before the end of the putVal method, * other threads will enter capacity expansion gradually, including those that have entered capacity expansion before inserting elements into putVal. So I need to recalculate */s = sumCount(); }}}Copy the codefullAddCount(long x,boolean wasUncontended)
In the case of high concurrency, a large number of threads modify the baseCount variable, resulting in performance loss, and the number of CAS modifications will be more. After all, only one CAS can be successfully modified at a time. In this case, a divide-and-conquer approach is used, with multiple threads modifying multiple variables, each changing its own. Each thread randomly selects a slot of countCell[] to increment it.
About the code in the U.CompareAndSwap* method at the beginning of the specific roleThe unsafe classRelevant methods
// See LongAdder version for explanation
private final void fullAddCount(long x, boolean wasUncontended) {
int h;
// If the current thread has not been initialized by ThreadLocalRandom
if ((h = ThreadLocalRandom.getProbe()) == 0) {
// Initialize the current thread's probe value and seed
ThreadLocalRandom.localInit(); // force initialization
h = ThreadLocalRandom.getProbe();
wasUncontended = true;
}
// True if the last slot is non-empty
boolean collide = false; // True if last slot nonempty
/ / spin
for (;;) {
CounterCell[] as; CounterCell a; int n; long v;
// If the counter array is already initialized
if((as = counterCells) ! =null && (n = as.length) > 0) {
/** addresses subscript I with the probe value of the current thread if the array subscript I is null */
if ((a = as[(n - 1) & h]) == null) {
/** *cellBusy= 1: countCell[] is adding count units or expanding. *cellBusy=0: other */
if (cellsBusy == 0) { // Try to attach new Cell
// Try adding a new count unit
CounterCell r = new CounterCell(x); // Optimistic create
//CAS mode change cellBusy to 1
if (cellsBusy == 0 &&
U.compareAndSwapInt(this, CELLSBUSY, 0.1)) {
// The flag has been added
boolean created = false;
try { // Recheck under lock
// Double check the array is initialized.
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;
}
// Add a new count unit, jump out of the spin
if (created)
break;
// If adding fails, continue to try to modify the spin
continue; // Slot is now non-empty
}
}
collide = false;
}
/** When wasUncontended=false *, the card slot was calculated using the original probe value, and the CAS modification failed. So this means that there is another thread competing for the slot, so change the probe value of the current thread to find a new slot */
else if(! wasUncontended)// CAS already known to fail
wasUncontended = true; // Continue after rehash
// In this case, it is found that the card slot has an existing value, and there is no competition at present
else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))
break;
/** * else if the CAS fails to modify the count unit * and the array is expanded or reaches the maximum array length. NCPU is the number of logical * processors in the system, one thread per logical processor, assuming NCPU=8 *. CountCell [] maxLen>=8, meaning that only 8 threads will modify the count unit in the array at any one time. * /
else if(counterCells ! = as || n >= NCPU) collide =false; // At max size or stale
Keep me oncollide =true and keep me oncollide =true. Keep me oncollide =true and keep me onCollide (keep me onCollide)
else if(! collide) collide =true;
/** * No other thread is adding new count units or expanding capacity. Mark cellsBusy * /
else if (cellsBusy == 0 &&
U.compareAndSwapInt(this, CELLSBUSY, 0.1)) {
try {
/** Again check whether the counter array has been expanded and whether countCell[] in the current thread * working memory is obsolete */
if (counterCells == as) {// Expand table unless stale
// Expand and copy the array elements
CounterCell[] rs = new CounterCell[n << 1];
for (int i = 0; i < n; ++i) rs[i] = as[i]; counterCells = rs; }}finally {
cellsBusy = 0;
}
collide = false;
// Whether the capacity expansion is successful or not, enter the next loop and retry the CAS modification
continue; // Retry with expanded table
}
// Reset the probe value of the current thread to find the card slot again
h = ThreadLocalRandom.advanceProbe(h);
}
// The array is not initialized yet, and no other thread is initializing it
else if (cellsBusy == 0 && counterCells == as &&
U.compareAndSwapInt(this, CELLSBUSY, 0.1)) {
boolean init = false;
try { // Initialize table
// Double check is the basic operation in multithreaded cases
if (counterCells == as) {
// The initial length is 2
CounterCell[] rs = new CounterCell[2];
// Put the count of the current thread into an array
rs[h & 1] = new CounterCell(x);
counterCells = rs;
init = true; }}finally {
cellsBusy = 0;
}
if (init)
break;
}
/** * failed to modify cellBusy, indicating that another thread is operating on the counter array, indicating that the baseCount is under less pressure. Try changing baseCount again */
else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x))
break; // Fall back on using base}}Copy the codesumCount()
Function: Computes the number of elements in 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 codecapacity
This is where concurrentHashMap comes in. How to control concurrent expansion in multiple threads?
helpTransfer
Function: If the current ConcurrentHashMap instance is expanding, the current thread enters the expansion method and assists other threads in expanding.
/** * Helps transfer if a resize is in progress. */
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
Node<K,V>[] nextTab; int sc;
// If capacity expansion is being performed, the current thread joins the capacity expansion
if(tab ! =null && (f instanceofForwardingNode) && (nextTab = ((ForwardingNode<K,V>)f).nextTable) ! =null) {
/** * What follows is the same as before parsing the addCount() method */
int rs = resizeStamp(tab.length);
// Check whether the capacity is being expanded
while (nextTab == nextTable && table == tab &&
(sc = sizeCtl) < 0) {
// Check whether expansion is complete
if((sc >>> RESIZE_STAMP_SHIFT) ! = rs || sc == rs +1 ||
sc == rs + MAX_RESIZERS || transferIndex <= 0)
break;
//sc adds 1. The current thread enters the capacity expansion method to assist capacity expansion
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
transfer(tab, nextTab);
break; }}// The expanded array reference is returned after expansion
return nextTab;
}
// If the thread enters this method and has completed the expansion, it returns the current array object
return table;
}
Copy the codetransfer
Now, here’s the point.
When I first parsed the source code of ConcurrentHashMap expansion, I always wondered how to ensure the accuracy of get() method to get data during expansion. · The address reference to the current array object TAB does not change. · Holds a nextTable reference to the in-memory address of the new array for expansion purposes. · After successfully moving the data in a slot to the new array nextTable, insert a bridge into the slot in the current array TAB that leads to the new array. · Therefore, when the thread needs to read the data of the slot currently using the array TAB, it will get the bridge, reach the new array through the bridge, and search in the new array.
The bridge is 👉ForwardingNode class
General processing ideas:
· The subscript range to be handled by expansion is between 0 and (TAB [].length-1) (the subscript range of the old array).
· Each thread enters the Transfer method and first calculates the subscript interval segment to be processed. For example, if the old array length is 28, the total range of subscripts to deal with is [0,28-1]. The number of slots to be processed by each thread is calculated based on the number of logical processor cores on the device (closely related to the number of parallel threads). Assuming that the number of logical processors on the device is 8, the number of slots to be processed by the thread is 16.
The specific scenario of multi-threaded instances is as follows ↓
· assuming that the first thread a enters the capacity expansion method, it first calculates the number of card slots it needs to process as 16, the interval is [240,255], and marks the number of unprocessed slots as 256-16=240, then the total interval left unprocessed is [0,239]. Then it starts to process the data in this range. After processing, it is detected that if other threads are also expanding, its expansion journey is directly ended. · The second thread B enters the capacity expansion method, calculates the number of card slots to be processed as 16, interval [224,239], and marks the number of remaining unprocessed slots as 224, so the total remaining unprocessed interval is [0 223]. Let’s start working on this interval. If other threads are also expanding, the process ends. · This is true for each thread entering capacity expansion until the number of unprocessed threads is 0.
If the thread is single, the following ↓ is used
· Thread A enters capacity expansion and obtains that the current number of unprocessed items is 28, and the total range is [0,28-1]. The number of slots to be processed is 16, the number of slots to be processed is [240,255], and the number of unprocessed slots to be processed is 240. And then you start processing that part of the interval. · After the loop processing [240,255] is completed, the detection finds that the current number of unprocessed is still 240, indicating that there is no other thread participating in capacity expansion. Then the current thread A continues to calculate the next interval to be processed as [224,239], and the number of remaining unprocessed marks is 224. And then we go down to this interval. · There are two triggering points when thread A terminates capacity expansion. ① When it finishes processing an interval each time, it finds that the number of remaining unprocessed items marked before is different from the value obtained from main memory, so it ends directly, equivalent to handing over to another thread. (2) When the number of unprocessed data is 0, that is, all data has been migrated.
/** * Moves and/or copies the nodes in each bin to new table. See * above for explanation. */
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
/** * n: the length of the old array, the stride thread to handle the number of card slots */
int n = tab.length, stride;
/** * Calculate the number of card slots to be processed * NCPU: number of logical processors (a logical processor can only support one thread at a time) * MIN_TRANSFER_STRIDE: minimum number of card slots to be processed in a thread 16 */
if ((stride = (NCPU > 1)? (n >>>3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
// If the nextTab parameter is null, capacity expansion has just started. Initialize some global shared variables related to capacity expansion
if (nextTab == null) { // initiating
try {
// The array size is doubled
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])newNode<? ,? >[n <<1];
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
// Memory runs out when array size is 2 times. Therefore, the capacity cannot be expanded. Set the threshold to the maximum value of the Integer type, which is difficult to trigger capacity expansion.
sizeCtl = Integer.MAX_VALUE;
return;
}
// Point the global variable nextTable to the new array object
nextTable = nextTab;
// Can be interpreted as the number of unprocessed card slots
transferIndex = n;
}
int nextn = nextTab.length;
// Bridge the old array to the new array during expansion
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
boolean advance = true;
// Whether the expansion process is complete
boolean finishing = false; // to ensure sweep before committing nextTab
/ / spin
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
// Calculate the processing subscript range and mark the number of remaining unprocessed slots
while (advance) {
NextBound is the minimum value of the interval, and nextindex-1 is the maximum value of the interval
int nextIndex, nextBound;
//-- I >=bound indicates that data has been processed, and finishing=true indicates that data has been expanded
if (--i >= bound || finishing)
advance = false;
//transferIndex<=0 indicates that all slots have been processed or are being processed
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
// Modify transferIndex, which can be interpreted as changing the number of unprocessed slots to the original number of unprocessed slots minus the number of slots to be processed by the current thread
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
bound = nextBound;
i = nextIndex - 1;
advance = false; }}/** * the subscript of the processing is equal to or beyond the boundary of the overall processing interval [0,n] * three cases: *① The thread of the processing interval [0,stride-1], the processing is completed. While (I = -1, I = -1, I = -1, I = -1, I = -1, I = -1, I = -1, I = -1, I = -1
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
// If all threads have already processed capacity expansion
if (finishing) {
// Set the global shared variable nextTable to null so that other programs can use this to determine whether expansion has been completed
nextTable = null;
// The new array is in use
table = nextTab;
// Set sizeCtl to the next capacity expansion threshold
sizeCtl = (n << 1) - (n >>> 1);
return;
}
//sc=sc-1, indicating that the current thread has finished processing the corresponding interval it is responsible for
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
/** * When the first thread starts to expand, it will set * SC to (resizeStamp(n)<
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
return;
/** * all threads have completed capacity expansion, but to be on the safe side, you need to check again */
finishing = advance = true;
i = n; // recheck before commit}}/** * TAB [I]; /** * TAB [I]; /** * TAB [I]
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
/** * The slot's data has already been processed, set advance=true, that is, the next loop will enter * to evaluate the processing interval code to determine whether to continue processing the next interval */
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
/** * The data in this slot has not been processed */
else {
/** * Moving split linked list/tree data is an operation to write/modify public variables in memory, and the steps are multi-step, from the previous CAS knowledge, it is necessary to ensure its primitive */, to prevent ABA problems
synchronized (f) {
// Double-check is required in multi-threaded environments to ensure that the data in the slot is the same as the original data
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;
// Linear list
/** ** ** ** ** ** ** ** ** ** ** ** ** ** ** ** ** ** ** ** ** ** ** ** ** ** ** ** **
if (fh >= 0) {
int runBit = fh & n;
Node<K,V> lastRun = f;
for(Node<K,V> p = f.next; p ! =null; p = p.next) {
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)
ln = new Node<K,V>(ph, pk, pv, ln);
else
hn = new Node<K,V>(ph, pk, pv, hn);
}
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
// In slot I of the old array, put a bridge
setTabAt(tab, i, fwd);
/* * that is, the next loop enters the computation interval code to determine whether to continue processing the next interval */
advance = true;
}
// The tree is a red black tree
else if (f instanceof TreeBin) {
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; }}// Determine whether the split red-black tree needs to be disassembled and converted into a straight listln = (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;
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd);
advance = true;
}
}
}
}
}
}
Copy the codeAh-ha ha ha, this girl I finally put the whole process of writing, can make me cow bad ~~
