Basic introduction
ConcurrentHashMapThe schematic diagram of the structure is as followsHashMapIs similar in structure to,TreeBinA node is a proxy node for a tree-like red-black tree node,FWDThe node indicates the bucket after capacity expansionnextTable.
constant
ConcurrentHashMap constants are as follows:
/* ---------------- Constants -------------- */
/** * Hash array maximum limit */
private static final int MAXIMUM_CAPACITY = 1 << 30;
/** * Default hash table size */
private static final int DEFAULT_CAPACITY = 16;
/** * The largest possible (non-power of two) array size. * Needed by toArray and related methods. */
static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
/** * Concurrency level, left over from JDk1.7, 1.8 is only used for initialization. * does not represent concurrency level. * /
private static final int DEFAULT_CONCURRENCY_LEVEL = 16;
/** * Load factor. In JDK1.8 ConcurrentHashMap is fixed */
private static final float LOAD_FACTOR = 0.75 f;
/** * Tree threshold, which specifies that a tree operation is possible if the bucket list length reaches 8. * /
static final int TREEIFY_THRESHOLD = 8;
/** * The threshold for converting a red-black tree to a linked list */
static final int UNTREEIFY_THRESHOLD = 6;
/** * joins TREEIFY_THRESHOLD to control whether the bucket level is treified. Only when the length of the table array reaches 64 and the length of the linked list in a bucket level reaches 8 will the bucket level be treified */
static final int MIN_TREEIFY_CAPACITY = 64;
/** * The minimum step size of thread migration data, which controls the minimum interval of thread migration task */
private static final int MIN_TRANSFER_STRIDE = 16;
/** * Calculates an identifier */ generated during capacity expansion
private static int RESIZE_STAMP_BITS = 16;
/** * 65535 indicates the maximum number of concurrent threads */
private static final int MAX_RESIZERS = (1< < (32 - RESIZE_STAMP_BITS)) - 1;
/** * Capacity expansion related */
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
// When the hash value of Node is -1, it indicates that the Node is FWD and has been migrated
static final int MOVED = -1; // hash for forwarding nodes
// When the hash value of Node is -2, the current Node is treed and the current Node is a TreeBin object. The TreeBin object agent operates red-black trees
static final int TREEBIN = -2; // hash for roots of trees
static final int RESERVED = -3; // hash for transient reservations
// 0x7FFFFFFF => Convert to 31 binary bits 1 0111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111
/** * 0111 1111 1111 1111 1111 1111 1111 1111 1111 1111 * You can take a negative number, that is, a number with the highest sign bit of 1, and get a positive number by performing the bit-sum operation & with it, but not by taking the absolute value of */
static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash
/** Number of CPUS, to place bounds on some sizings
static final int NCPU = Runtime.getRuntime().availableProcessors();
/** For serialization compatibility. * JDK1.8 serializes */ For building compatibility with JDk1.7 ConcurrentHashMap
private static final ObjectStreamField[] serialPersistentFields = {
new ObjectStreamField("segments", Segment[].class),
new ObjectStreamField("segmentMask", Integer.TYPE),
new ObjectStreamField("segmentShift", Integer.TYPE)
};
Copy the codeMember variables
The member variables of ConcurrentHashMap are as follows:
/** * The array of bins. Lazily initialized upon first insertion. * Size is always a power of two. Accessed directly by Iterators. * Hash table, length must be 2 power */
transient volatile Node<K,V>[] table;
/** * The next table to use; Non-null only while resizing. * During the capacity expansion, the new table in the capacity expansion will be assigned to keep references to nextTable, which will be set to NULL */ after the capacity expansion
private transient volatile Node<K,V>[] nextTable;
/** * Base counter value, used mainly when there is no contention, * But also as a fallback during table initialization * races. Updated via CAS While the current LongAdder is locked, the increment is trapped in baseCount */
private transient volatile long baseCount;
/** * sizeCtl < 0 * 1. -1 indicates that the current table is being initialized (a thread is creating an array of tables). * 2. Indicates that the current table array is being expanded. 16 bits higher indicates that the identifier of the expansion is 16 bits lower. SizeCtl = 0, indicating that the DEFAULT_CAPACITY of the table array is set to size > 0 * * 1. If the table is not initialized, it indicates the initialization size * 2. If the table is initialized, it indicates the triggering condition (threshold) for the next capacity expansion */
private transient volatile int sizeCtl;
/** * Record the current progress during capacity expansion. All threads need to assign interval tasks from transferIndex to perform their own tasks. * /
private transient volatile int transferIndex;
/** * Spinlock (locked via CAS) used when resizing and/or creating CounterCells. * LongAdder in cellsBuzy 0 indicates that the LongAdder is currently unlocked, and 1 indicates that the LongAdder is currently locked */
private transient volatile int cellsBusy;
/** * Table of counter cells. When non-null, size is a power of 2. * The thread computs the hash value to its own cell, adding increments to the specified cell * total = sum(cells) + baseCount */
private transient volatile CounterCell[] counterCells;
Copy the code-
Table: Node is lazily loaded and initialized only when put is called for the first insert. Array size is always a power of 2.
-
NextTable: During expansion, new tables in the expansion are referenced to nextTable and will be set to Null after expansion.
-
BaseCount: The addCount method is used to calculate the number of elements in the hash table. The addCount method is used in the same way as the Base attribute in LongAdder. When multiple threads modify the baseCount in a race, the above attribute counterCells is used to split the hot spots. You only need to modify the value in the bucket bit. When calculating the total number of elements in the hash table, add the value of baseCount to the value of the array counterCells. The principle of LongAdder can be viewed [LongAdder source code analysis] this article, here will not expand.
-
CellsBusy: This property is used in the same way as in LongAdder, where cellsBuzy 0 indicates that the LongAdder object is currently unlocked and 1 indicates that the LongAdder object is currently locked
-
The sizeCtl attribute values look like this:
- -1: indicates that the table array is being initialized
- -n: The table array is being expanded. 16 bits higher indicates the expansion identifier. 16 bits lower indicates the number of threads (1 + nThread) participating in concurrent expansion.
- If the value is 0, DEFAULT_CAPACITY is 16 when creating the table array
- Greater than 0: indicates the initialization size if the table is not initialized, or the triggering condition (threshold) for the next expansion if the table is initialized.
Static code block
The static code block for ConcurrentHashMap is as follows:
UnsafeClass throughU.objectfieldoffset (k.getdeclaredfield (" attribute name "));To obtain the offset address of the corresponding attribute for subsequent passUnsafethecasModify the corresponding attribute value.ABASE = U.arrayBaseOffset(ak);:ABASEforNodeThe header address of the arrayint scale = U.arrayIndexScale(ak);: represents the space occupied by array units,scalesaidNode[]The amount of space occupied by each cell in the array.Integer.numberOfLeadingZeros(scale)This method retrieves the maximum number of consecutive zeros in the binary representation of a parameter, as a simple example, assumingNodeSize of space occupied by a nodescalefor4, binary is100Because there is a common32 -, then100The front and29Bits are consecutive zeros, soInteger.numberOfLeadingZeros(4)The value of29ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);: Why is this calculation done here? What is the purpose? Assuming thatNodeSize of space occupied by a nodescalefor4, thenASHIFT = 31 - 29 = 2For example, I want to get the offset address of the fifth element in the array:ABASE + 5 * scale = ABASE + 5 * 4That is, the starting address of the first element plus the element subscript, we can also passABASE + (5 << ASHIFT)If we move 5 two Spaces to the left, that also meansABASE + 5 * 4In this way, it is also possible to obtain the corresponding offset address, and this way uses bit operation, faster.
static {
try{ U = sun.misc.Unsafe.getUnsafe(); Class<? > k = ConcurrentHashMap.class;/** * the following are the memory offsets for the corresponding attributes */
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);// Represents the size of the array cell,scale represents the size of each cell in the Node[] array
int scale = U.arrayIndexScale(ak);
// Since scale must be powers of 2, 1 0000&0 1111 = 0
if ((scale & (scale - 1)) != 0)
throw new Error("data type scale not a power of two");
//numberOfLeadingZeros() returns the number of zeros in a row from the highest digit to the lowest digit.
//8 => 1000 numberOfLeadingZeros(8) = 28
//4 => 100 numberOfLeadingZeros(4) = 29
// you can run between two streams.
// Suppose scale is 4, for example I want to get the offset address of the fifth element in the array: ABASE + 5 * scale
//ABASE + 5 * scale = (1 << ASHIFT
ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);
} catch (Exception e) {
throw newError(e); }}Copy the codeThe inner class
Node
The Node structure is similar to that of HashMap1.8, except that val and next in ConcurrentHashMap use volatile to indicate memory visibility and ensure the safety of multithreaded operations.
static class Node<K.V> implements Entry<K.V> {
final int hash;
final K key;
volatile V val;// The volatile modifier indicates memory visibility
volatile Node<K,V> next;// The volatile modifier indicates memory visibility
}
Copy the codeTreeNode
TreeNode nodes, namely red-black tree nodes, are as follows:
/** * Nodes for use in TreeBins */
static final class TreeNode<K.V> extends Node<K.V> {
TreeNode<K,V> parent; // Parent node
TreeNode<K,V> left; // Left child node
TreeNode<K,V> right; // Right child node
TreeNode<K,V> prev; // needed to unlink next upon deletion
boolean red; // The default is red
}
Copy the codeForwardingNode
ForwardingNode identifies bucket bits for capacity expansion. The structure is as follows:
/** * 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;// New table array after expansion
ForwardingNode(Node<K,V>[] tab) {
super(MOVED, null.null.null);
this.nextTable = tab; }}Copy the codeWhen capacity expansion occurs, the ForwardingNode in concurrentHashMap works as shown in the following figure
Of course, data migration of high-low linked list and TreeNode nodes will be involved, which will be explained in details in the capacity expansion method below.
Utility methods
Spread (Hash) source analysis
The spread method is a perturbation function, similar to the hash method in HashMap1.8, to make the hash table more diffuse and reduce hash collisions. The code is as follows:
/** * example: * 1100 0011 1010 0101 0001 1100 0001 1110 ==> h * 0000 0000 0000 0000 1100 0011 1010 0101 ==> h >>> 16 * 1100 0011 1010 0101 1101 1111 1011 1011 ==> h ^ (h >>> 16 * --------------------------------------- * 1100 0011 1010 0101 1101 1111 1011 1011 ==> h ^ (h >>> 16 * 0111 1111 1111 1111 1111 1111 1111 1111 ==> HASH_BITS * 0100 0011 1010 0101 1101 1111 1011 1011 ==> (h ^ (h >>> 16)) & HASH_BITS; HASH_BITS = &hash_bits = &hash_bits = 0 To get a positive hash value, because HASH_BITS are 31 1 */
static final int spread(int h) {
return (h ^ (h >>> 16)) & HASH_BITS;
}
Copy the code1. Why do you want to leavekeyHow about moving the hashCode 16 bits right and xor ^ with the original hashCode?
Since we know that the routing function is hash & (the length of the hash array is -1), the length of the hash table is usually several powers of 2. For example, if the length of the hash table is 16, 16-1=15 and converted to binary 1111, the final effect of hash & 1111 is the last few bits, and the high bits do not work. So h ^ (h >>> 16) is used to make the hash table less likely to collide.
2. Why& HASH_BITS?
/** * 0111 1111 1111 1111 1111 1111 1111 1111 1111 1111 * You can take a negative number, that is, a number with the highest sign bit of 1, and get a positive number by performing the bit-sum operation & with it, but not by taking the absolute value of */
static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash
Copy the codeSince HASH_BITS has the value 0x7fffffff, i.e. 0111 1111 1111 1111 1111 1111 1111 1111 1111 1111 there are 31 1s in total, & HASH_BITS is used to make the first digit of the final hash value 0, which is a positive number (the first digit 1 indicates a negative number).
TabAt (TAB,index) source code analysis
(long) I << ASHIFT) + ABASE; As explained in the static code block above, the code looks like this:
(long) I << ASHIFT) + ABASE: the start address of the first element of the table array */
@SuppressWarnings("unchecked")
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 codeCasTabAt (TAB,index, C, V) source analysis
This method modifies the table bucket element of the corresponding subscript by cas, and the code is as follows:
/** * alter table Node element * with subscript I@paramC Expected value *@paramV Change the 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 codeSetTabAt (TAB,index, V) source analysis
This method stores elements in the bucket of the corresponding subscript of table. The code is as follows:
/** * stores Node */ at the subscript I position in the table array
static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
}
Copy the codeResizeStamp (INT N) source code analysis
During capacity expansion, a capacity expansion identifier will be calculated. Because the capacity expansion operation may be executed by multiple threads, the thread can assist in capacity expansion only when it gets the same capacity expansion identifier identifier as other threads.
For example, expanding capacity from 16 to 32,
n = 16 = 1000.Integer.numberOfLeadingZeros(n) = 27 = 0000 0000 0001 1011RESIZE_STAMP_BITS = 16Is a fixed value,1 << (RESIZE_STAMP_BITS - 1) = 1000 0000 0000 0000 = 327680000 0000 0001 1011 1000 0000 0000 0000 | = 1000, 0000, 0001, 1011
That is, the expansion identifier from 16 to 32 is a fixed value.
The code is as follows:
/** * Returns the stamp bits for resizing a table of size n. * Must be negative when shifted left by RESIZE_STAMP_SHIFT. * A capacity expansion identifier is calculated during capacity expansion. * Since capacity expansion may be performed by multiple threads, Each thread to the expansion of table logo stamp is the same * 16 - > expansion and from 16 to 32 * 32 numberOfLeadingZeros (16) = > 1 = = > > 27 0000 0000 0000 0001 1011 * | * (1 < < (RESIZE_STAMP_BITS - 1)) => 1000 0000 0000 0000 => 32768 * -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- - * 0000, 0000, 0001, 1011 * 1000, 0000, 0000, 0000 * 1000, 0000 0001 * 1011 * Integer. NumberOfLeadingZeros (n) : The maximum number of consecutive 0's in the 32-bit bits of N, such as 16 => 10000, is the maximum of the preceding 27 consecutive 0 * RESIZE_STAMP_BITS, a fixed value of 16 */
static final int resizeStamp(int n) {
return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));
}
Copy the codeTableSizeFor c (int) method
This method has the same effect as in JDK1.8 HashMap, which is to get the smallest power of 2 that returns >=c. The code is as follows:
/** * Returns a power of two table size for the given desired capacity. * See Hackers Delight, The SEC 3.2 * returns the smallest power of 2 > = c number 28 * * c = n = 27 = > 0 b = > 11111 * 11111 11011 * 11011 | 01101 | 11111 * 00111 = >... * => 11111 + 1 =100000 = 32 */
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 codeA constructor
Common constructors are as follows:
/** * Creates a new, empty map with the default initial table size
public ConcurrentHashMap(a) {}public ConcurrentHashMap(int initialCapacity) {
if (initialCapacity < 0)
throw new IllegalArgumentException();
int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1))? MAXIMUM_CAPACITY ://tableSizeFor returns a power greater than the minimum of 2 of the passed parameter
tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
/** * sizeCtl > 0 * When the table is not initialized, sizeCtl indicates the initialized capacity */
this.sizeCtl = cap;
}
/** * This (initialCapacity, loadFactor, 1) calls the following constructor */
public ConcurrentHashMap(int initialCapacity, float loadFactor) {
this(initialCapacity, loadFactor, 1);
}
/** * In ConcurrentHashMap, the load factor is final,o.75f * concurrencyLevel has no real meaning */
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {
if(! (loadFactor >0.0 f) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
if (initialCapacity < concurrencyLevel) // Use at least as many bins
initialCapacity = concurrencyLevel; // as estimated threads
long size = (long) (1.0 + (long)initialCapacity / loadFactor);
int cap = (size >= (long)MAXIMUM_CAPACITY) ?
MAXIMUM_CAPACITY : tableSizeFor((int)size);
/** * sizeCtl > 0 * When the table is not initialized, sizeCtl indicates the initialized capacity */
this.sizeCtl = cap;
}
Copy the codeCore method write operation put(k,v)
PutVal () method
public V put(K key, V value) {
return putVal(key, value, false);
}
/** Implementation for put and putIfAbsent */
//onlyIfAbsent: Overwrites the value of value if the key passed in exists. If true, overwrites the value of value
final V putVal(K key, V value, boolean onlyIfAbsent) {
// Control k and v cannot be null
if (key == null || value == null) throw new NullPointerException();
// With the spread method, we can make the high position also participate in the forward addressing operation.
int hash = spread(key.hashCode());
//binCount Indicates the subscript position of the linked list in the bucket after the current K-V is encapsulated as node and inserted into the specified bucket
//0 indicates that the current bucket bit is null and node can be directly placed
//2 Indicates that the bucket bit may be a red-black tree
int binCount = 0;
// TAB refers to the table of the map object
/ / spin
for (Node<K,V>[] tab = table;;) {
//f is the head of the bucket
//n indicates the length of the hash table array
// I indicates the bucket bit index obtained after the key is addressed
//fh indicates the hash value of the bucket header
Node<K,V> f; int n, i, fh;
//CASE1: yes, indicating that the table in the current map is not initialized..
if (tab == null || (n = tab.length) == 0)
// Eventually the current thread gets the latest map.table reference.
tab = initTable();
//CASE2: I indicates that key uses the routing addressing algorithm to obtain the subscript position of the table array corresponding to key. TabAt obtains the head node F of the specified bucket
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
// Enter CASE2 code block preconditions The current table array I bucket is Null.
// Use CAS to set the bucket bit of array I to new Node
(hash, key, value, null), and the expected value is NULL
,v>
// If the cas operation succeeds, the cas operation is ok
// the cas operation failed, indicating that another thread set the value to the specified I bucket before the current thread.
// The current thread can only spin again to go through other logic.
if (casTabAt(tab, i, null.new Node<K,V>(hash, key, value, null)))
break; // no lock when adding to empty bin
}
//CASE3: the precondition is that the bucket header must not be null.
// If the condition is true, the head node of the current bucket is FWD, which indicates that the map is being expanded.
else if ((fh = f.hash) == MOVED)
// After seeing the FWD node, the current node has the obligation to help the current map to complete the migration of data
// We'll see after the expansion.
tab = helpTransfer(tab, f);
//CASE4: The current bucket bit can be either a linked list or a red-black tree proxy node, TreeBin
else {
// When the insert key exists, the old value is assigned to oldVal, which is returned to the put method call..
V oldVal = null;
// Use sync to lock the "head node"
synchronized (f) {
// Check whether the current bucket header is the same as the previous one.
// To prevent other threads from modifying the header of the bucket and causing the current thread to lock from sync, there is a problem. I don't have to do anything after that.
if (tabAt(tab, i) == f) {// Select * from lock ();
// If the condition is true, the current bucket level is an ordinary bucket level.
if (fh >= 0) {
//1. If the current insert key is inconsistent with the key of all elements in the linked list, the current insert operation is appended to the end of the linked list. BinCount indicates the length of the linked list
//2. If the current insert key is the same as the key of an element in the linked list, the current insert operation is probably a replacement. BinCount indicates the location of the conflict (bincoun-1)
binCount = 1;
// The list of buckets in the current iteration loop, e is the node processed for each iteration.
for (Node<K,V> e = f;; ++binCount) {
// The current loop node key
K ek;
// condition 1: e.hash == hash indicates that the hash value of the current element of the loop is the same as the hash value of the inserted node
/ / condition 2: (= e.k (ek ey) = = key | | (ek! = null && key.equals(ek)))
// true: the current node of the loop is in conflict with the key of the inserted node
if(e.hash == hash && ((ek = e.key) == key || (ek ! =null && key.equals(ek)))) {
// Assign the value of the current loop element to oldVal
oldVal = e.val;
if(! onlyIfAbsent) e.val = value;break;
}
// When the current element does not match the key of the inserted element, the following procedure is performed.
//1. The update loop processes the node as the next node of the current node
Check whether the next node is null. If the value is null, the current node is at the end of the queue, and data needs to be appended to the end of the node.
Node<K,V> pred = e;
if ((e = e.next) == null) {
pred.next = new Node<K,V>(hash, key,
value, null);
break; }}}// The bucket bit must not be a list
// If the condition is true, the current bucket bit is the red-black tree agent node TreeBin
else if (f instanceof TreeBin) {
//p means that putTreeVal returns a reference to a node in a red-black tree that conflicts with the key you inserted.
Node<K,V> p;
// Force binCount to be 2, because binCount <= 1 has other meanings, so set it to 2. Go back to addCount.
binCount = 2;
// Condition 1: yes, the key of the node being inserted is the same as that of a node in the red-black tree
if((p = ((TreeBin<K,V>)f).putTreeVal(hash, key, value)) ! =null) {
// Assign the value of the conflicting node to oldVal
oldVal = p.val;
if(! onlyIfAbsent) p.val = value; }}}}// Indicates that the bucket bit is not null and may be a red-black tree or a linked list
if(binCount ! =0) {
// If binCount>=8, the bucket bit must be a linked list
if (binCount >= TREEIFY_THRESHOLD)
// Call the method that converts the linked list to a red-black tree
treeifyBin(tab, i);
The data key inserted by the current thread conflicts with the original k-v, and the original data v needs to be returned to the caller.
if(oldVal ! =null)
return oldVal;
break; }}}//1. Count the number of data in the table
//2. Check whether the capacity expansion threshold is reached.
addCount(1L, binCount);
return null;
}
Copy the code-
1. Calculate the hash value from spread(key.hashcode ()), and then check whether the table array is initialized. If not, call initTable to initialize it.
-
2. Then (n-1) & hash) calculates the table bucket subscript of the current hash value. If the bucket is empty, create a Node and populate the bucket using casTabAt (underlying cas of the Unsafe class) to complete the operation.
-
3. If the table bucket bit corresponding to (n-1) & hash is not empty, check whether the hash value of the bucket element fh = F. hash) == MOVED. If the hash value is -1, the bucket is an FWD node, and the map is being expanded. HelpTransfer is then called to help the other threads complete the expansion.
-
4. If the condition of Step 3 is not true, it indicates that the current bucket level may be a linked list or a red-black tree agent node TreeBin. Then use synchronized exclusive lock to lock the current bucket level node to ensure that only one thread can modify all elements of the bucket level. If the hash value of the bucket is greater than 0, it indicates that the bucket is an ordinary bucket. If the hash value of the bucket is the same as that of the bucket to be inserted and the key is the same, or if the equals method returns true, the bucket is overwritten. Otherwise insert the insert node to the end of the one-way list. If the current bucket bit f instanceof TreeBin is a TreeBin node, the putTreeVal method is called to insert the node (explained in more detail in TreeBin below).
-
5. After node insertion, check whether binCount is greater than TREEIFY_THRESHOLD, that is, whether it is greater than 8. If so, call treeifyBin to tree the unidirectional list
-
6. Finally, run the addCount method to count the number of data in the table and determine whether the capacity expansion threshold is reached.
InitTable () method
Initialize the table array as follows:
/** * Initializes table, SizeCtl < 0 * * 1. -1 indicates that the table is being initialized (a thread is creating an array of tables). The current thread needs to spin wait.. * * 2. Indicates that the current table array is being expanded. The higher 16 bits indicates that the expansion identifier is lower 16 bits: (1 + nThread) Number of threads participating in concurrent capacity expansion * * * * sizeCtl = 0, indicating that the DEFAULT_CAPACITY is set to * * * * sizeCtl > 0 * * * * 1. If the table is not initialized, it indicates the initialization size * * 2. If the table is initialized, it indicates the triggering condition (threshold) for the next capacity expansion */
private final Node<K,V>[] initTable() {
/ / TAB reference map. The table
//sc sizeCtl temporary value
Node<K,V>[] tab; int sc;
// Spin condition: map.table is not initialized
while ((tab = table) == null || tab.length == 0) {
if ((sc = sizeCtl) < 0)
// The probability is -1, indicating that another thread is in the process of creating the table, and the current thread is not competing for the lock of initializing the table.
//Thread.yield(); Causes the current thread to release CPU run permissions
Thread.yield(); // lost initialization race; just spin
//1. SizeCtl = 0, indicating that DEFAULT_CAPACITY is used to create the table array
//2. If the table is not initialized, the size is initialized
//3. If the table has been initialized, it indicates the triggering condition (threshold) for the next capacity expansion.
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
// Why do we need to judge? Prevent the current thread from initializing again after another thread has initialized. Data is lost.
// If this condition is true, no other thread has entered the if block, and the current thread is entitled to initialize the table.
if ((tab = table) == null || tab.length == 0) {
If sc is greater than 0, use sc as the specified size when creating a table. Otherwise, use the default value 16.
int n = (sc > 0)? sc : DEFAULT_CAPACITY;@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])newNode<? ,? >[n];// Finally assign to map.table
table = tab = nt;
/ / n > > > 2 = > equal to 1/4 n n - (1/4) n = 3/4 n = > 0.75 * n
//sc 0.75N indicates the trigger condition for the next capacity expansion.
sc = n - (n >>> 2); }}finally {
//1. If the current thread is the thread that creates the map.table for the first time, sc indicates the threshold for the next capacity expansion
//2. Indicates that the current thread is not the thread that created map.table for the first time, and the current thread enters the else if block
//sizeCtl set to -1, then change it to the entry value.
sizeCtl = sc;
}
break; }}return tab;
}
Copy the codeThe core method counts the addCount() method
The addCount method does two main things
- Statistics how much data is present in the table how much data is present in the hash table
- Check whether the capacity expansion threshold is reached. Because in addition to inserting elements
putValMethod is called to delete the elementremoveMethod is also called.
/** * 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
* @paramcheck if <0, don't check resize, if <= 1 only check if uncontended * * 1. How much data does the hash table have? * 2. Check whether the capacity expansion threshold is reached. * /
private final void addCount(long x, int check) {
/ / as said LongAdder. Cells
/ / b said LongAdder. Base
//s indicates the number of elements in the current map.table
CounterCell[] as; long b, s;
// Condition 1: true-> indicates that cells have been initialized and the current thread should use hash addressing to find the appropriate cell to add data to
// false-> indicates that the current thread should add data to base
False -> Write base successfully, add data to base, do not create cells
// true-> failed to write base, contention with other threads on base, current thread should try to create cells.
if((as = counterCells) ! =null| |! U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
// How many cases go into the if block?
True -> indicates that cells have been initialized and the current thread should use hash addressing to find the appropriate cell to add data to
//2. True -> failed to write base, contention with other threads on base, current thread should try to create cells.
// A indicates the cell matching the hash address of the current thread
CounterCell a;
//v represents the expected value of the current thread when it writes the cell
long v;
//m represents the length of the current cells array
int m;
//true -> uncontended false-> Contended
boolean uncontended = true;
A: / / conditions as = = null | | (m = as length - 1) < 0
//true-> indicates that the current thread failed to enter the if block by writing to base, so it needs to call the fullAddCount method to expand or retry. LongAdder.longAccumulate
/ / condition 2: a = as [ThreadLocalRandom. GetProbe () & m]) = = null precondition: cells has been initialized
//true-> Indicates that the cell table hit by the current thread is empty, and the current thread needs to enter the fullAddCount method to initialize the cell and place it in the current position.
// If (! (uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x)
False -> If false is obtained, the current thread successfully updates the matched cell in CAS mode
// true-> If the value is reversed, the current thread fails to update the matched cell in CAS mode. You need to enter fullAddCount for retries or expand cells.
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 is the longAccumulate method in LongAdder
fullAddCount(x, uncontended);
// In the case of fullAddCount, the current thread does not participate in the expansion logic, and directly returns to the call point.
return;
}
if (check <= 1)
return;
// Get the number of current hash elements, which is an expected value
// is the sum method in LongAdder
s = sumCount();
}
/** * check >= 1 indicates the length of the bucket list corresponding to the inserted element * check == 0 indicates that the bucket bit corresponding to the inserted element key is empty, that is, it is directly inserted into the bucket bit. Check is 0 * check == 2 which means that the bucket key is already treed * * Check < 0 means that instead of calling the addCount of the insert element method put, it means that the addCount of the insert element method remove, It will not enter the following if block */
// addCount for a call to a put operation
if (check >= 0) {
/ / map TAB said. The table
/ / nt said map. NextTable
//n indicates the length of the map.table array
//sc represents the temporary value of sizeCtl
Node<K,V>[] tab, nt; int n, sc;
/** * sizeCtl < 0 * 1. -1 indicates that the current table is being initialized (a thread is creating an array of tables). * 2. Indicates that the current table array is being expanded. 16 bits higher indicates that the identifier of the expansion is 16 bits lower. SizeCtl = 0, indicating that the DEFAULT_CAPACITY of the table array is set to size > 0 * * 1. If the table is not initialized, it indicates the initialization size * 2. If the table is initialized, it indicates the triggering condition (threshold) for the next capacity expansion */
/ / spin
//条件一：s >= (long)(sc = sizeCtl)
// true-> 1. If the current sizeCtl is a negative value, it indicates that the capacity is being expanded. The current thread should assist in the expansion
// 2. The current sizeCtl value is a positive number, indicating the capacity expansion threshold
// false-> Indicates that the table does not meet the capacity expansion condition
//条件二：(tab = table) != null
// set true
(n = tab.length) < MAXIMUM_CAPACITY
// true-> If the current table length is smaller than the maximum value, you can expand the table.
while (s >= (long)(sc = sizeCtl) && (tab = table) ! =null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
// Unique identifier of expansion batch
//16 -> 32 The expansion id is 1000 0000 0001 1011, that is, the expansion id obtained from 16 to 32 is the same
int rs = resizeStamp(n);
// Conditional: The table is being expanded
// The current thread should theoretically assist the table to complete the expansion
if (sc < 0) {
// condition 1 :(sc >>> RESIZE_STAMP_SHIFT)! 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000
// true-> Indicates that the unique capacity expansion id obtained by the current thread is not the current batch capacity expansion id
// false-> Indicates that the unique identification stamp obtained by the current thread is the current batch capacity expansion
Jira: = sc == (rs << 16) + 1
// true-> The current thread does not need to participate in the capacity expansion
// false-> Expansion is still in progress, the current thread can participate
// Jira: = sc == (RS <<16) + MAX_RESIZERS
// true-> Indicates that the number of concurrent expansion threads reaches the maximum value of 655,5-1
// false-> Indicates that the current thread can participate
// condition 4 :(nt = nextTable) == null
// true-> Indicates that capacity expansion is complete
// false-> Expansion in progress
if((sc >>> RESIZE_STAMP_SHIFT) ! = rs || sc == rs +1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0)
break;
// Prerequisites: The table is being expanded.. The current thread has the opportunity to participate in the expansion.
// If the condition is true, the current thread successfully participates in the capacity expansion task, and the lower 16-bit value of SC is increased by 1, it indicates that another thread participates in the task, i.e. (1 + nThread).
SizeCtl: // condition failed: 1. There are many threads trying to modify sizeCtl. Another thread successfully modified sizeCtl, causing your SC expected value to be inconsistent with the memory value
// 2. Transfer task internal threads sizeCtl has also been modified.
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
// Helps expand threads and holds nextTable
transfer(tab, nt);
}
//RESIZE_STAMP_SHIFT to 16, shift the expansion stamp 16 bits to the left and add 2
//1000 0000 0001 1011 0000 0000 0000 0000 +2 => 1000 0000 0001 1011 0000 0000 0000 0010
// 1000 0000 0001 1011 0000 0000 0000 0010 The lower 16 bits are (1 + nThread) 1+1=2
// If the condition is true, it indicates that the current thread is the first thread to trigger capacity expansion, and the sizeCtl is changed to negative, because the first sign bit is 1, some capacity expansion preparation needs to be done in the transfer method
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
// The thread that triggers the expansion condition does not hold nextTable
transfer(tab, null); s = sumCount(); }}}Copy the code-
1. Use the same method as LongAdder to finally call sumCount to calculate the total number of elements in the current hash table, and of course add baseCount to the sum of elements in the array CounterCell.
-
2. Then check whether the second parameter check of the addCount method is greater than 0. The value of the check passed by the addCount method is greater than 0 when the addCount method is called inside the putVal method, while the value of the check passed by the addCount method is less than 0 when the addCount method is called inside the remove method.
-
3. If check is greater than 0, then call resizeStamp in the while loop to calculate the unique identification stamp of expansion, and then judge whether the value of attribute sizeCtl is less than 0. As we know above, if the value of attribute sizeCtl is less than 0, it means that the table is currently expanding. The current thread should theoretically help the table to complete the expansion, and then increse the sizeCtl value by 1 via CAS. When the first thread expands, it will change the sizeCtl value to << RESIZE_STAMP_SHIFT) + 2, that is, the sizeCtl stamp is moved 16 bits to the left, that is, the higher 16 bits is the expansion stamp, the lower 16 bits is 0, then add 2 to the lower 1 bit, 2 means that there are 1 + N threads currently expanding. That is, when only one thread is expanding, the value of the lower 16 bits is 2. Since the value of the highest bit is 1, and the sign bit is 1, which means negative, the value of sizeCtl becomes less than 0 after the first expansion thread changes the value of sizeCtl. So if I go back up why do I add 1? That makes sense, right? SizeCtl increase the number of threads in the lower 16 bits by 1. Transfer (TAB, NT) is then called to assist other threads with capacity expansion.
-
If the sizeCtl value is not less than 0 for the first time, it indicates that the current thread is the first thread participating in capacity expansion. Then, according to step 3 above, change the sizeCtl value to the capacity expansion identifier stamp << RESIZE_STAMP_SHIFT) + 2, and finally call the Transfer method to expand capacity.
Super core method expansion transfer method
The capacity expansion flowchart is as follows:
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
//n indicates the length of the table array before capacity expansion
//stride represents the step size assigned to the thread task, which requires a thread to carry out data migration with the number of bucket bits with the length of stride
int n = tab.length, stride;
// convenient explanation source stride fixed to 16
if ((stride = (NCPU > 1)? (n >>>3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
// If the condition is true, the current thread is the thread triggering capacity expansion, and some preparations need to be made for capacity expansion
// The condition is not true: the current thread is the thread assisting capacity expansion..
if (nextTab == null) { // initiating
try {
// Create a table twice as large as before expansion
@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;
}
// Assigns to the object property nextTable to help expand threads get new tables
nextTable = nextTab;
// A marker to record the overall location of the migrated data. The index count starts at 1.
transferIndex = n;
}
// Represents the length of the new array
int nextn = nextTab.length;
// FWD node. After the data processing of a bucket is complete, that is, all data of the bucket is migrated. Set the bucket to FWD.
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
// Advance the mark
boolean advance = true;
// Complete the tag
boolean finishing = false; // to ensure sweep before committing nextTab
// I indicates the bucket bit assigned to the current thread
//bound represents the lower bound assigned to the current thread task
int i = 0, bound = 0;
/ / spin
for (;;) {
// the head of the f bucket
// Hash the fh header
Node<K,V> f; int fh;
/** * 1. Assign a task range to the current thread * 2. Maintain the task progress of the current thread (I indicates the bucket bit currently being processed) * 3. Maintains the progress of the map object globally */
while (advance) {
// Assign the task to the start subscript
// Assign the end subscript of the task
int nextIndex, nextBound;
//CASE1:
-- I >= bound
// hold: indicates that the current thread has not completed its task, and there is a corresponding range of buckets to process, -- I causes the current thread to process the next bucket bit.
// Failed: indicates that the current thread task is completed or not assigned
if (--i >= bound || finishing)
advance = false;
//CASE2:
// Preconditions: the current thread task is completed or not assigned
Set the I variable of the current thread to -1, and execute the procedures related to the exit migration task after breaking out of the loop
// Condition not true: Indicates that bucket bits in the global scope of the object have not been allocated
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
//CASE3:
// Prerequisites: 1. The current thread needs to allocate a task range. 2
// If yes, the task is successfully assigned to the current thread
// Conditional failure: failed to allocate to the current thread
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
// Lower bound of bucket bits processed by the current thread
bound = nextBound;
// The current thread processes the upper limit of bucket bits from bottom up
i = nextIndex - 1;
advance = false; }}/ / CASE1:
// if I < 0, I =-1
// True: no task is assigned to the current thread
if (i < 0 || i >= n || i + n >= nextn) {
// Save the sizeCtl variables
int sc;
if (finishing) {
// Before the last expansion thread exits
nextTable = null;
table = nextTab;
// sizeCtl = 0.75的2n
sizeCtl = (n << 1) - (n >>> 1);
return;
}
If sizeCtl is set to sizeCtl, sizeCtl specifies whether (1 + nThread) threads are expanding. If sizeCtl is set to sizeCtl, sizeCtl specifies whether (1 + nThread) threads are expanding
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
//1000 0000 0001 1011 0000 0000 0000 0000 0000 0000
// Conditional: Indicates that the current thread is not the last thread to exit the transfer task
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
// Exit normally
return;
// Before the last expansion thread exits, we need to do some check work in case some missing bucket data has not been migrated
finishing = advance = true;
i = n; // recheck before commit}}// Preconditions: [CASE2~CASE4] The current thread task is not finished, in progress
//CASE2:
// Conditional: No data is stored in the bucket. You only need to set this bucket to FWD.
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
//CASE3:
// If the condition is true, the current bucket level has been migrated, and the current thread does not need to process it any more. The current thread can directly update the task index of the current thread to process the next bucket level or other operations
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
//CASE4:
// Prerequisites: Data exists in the bucket and the node node is not the FWD node. The data needs to be migrated.
else {
//sync locks the current bucket header
synchronized (f) {
// Prevent the current bucket header object from being modified by another writer thread before you add the lock object.
if (tabAt(tab, i) == f) {
//ln represents a low level list reference
//hn stands for high level list references
Node<K,V> ln, hn;
// Conditional: Indicates that the bucket level is a linked bucket level
if (fh >= 0) {
//lastRun
// We can get the node at the end of the current list
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 lastRun refers to a low level list, then let ln refer to the low level list
if (runBit == 0) {
ln = lastRun;
hn = null;
}
// If lastRun refers to the high-order list, then hn points to the high-order list
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);
setTabAt(tab, i, fwd);
advance = true;
}
// If the condition is true, the current bucket bit is a red-black tree agent node TreeBin
else if (f instanceof TreeBin) {
// Convert the header to treeBin reference t
TreeBin<K,V> t = (TreeBin<K,V>)f;
// Lo points to the head of the low list. LoTail points to the tail of the low list
TreeNode<K,V> lo = null, loTail = null;
// Lo points to the head of the loTail points to the tail of the high list
TreeNode<K,V> hi = null, hiTail = null;
//lc represents the number of low list elements
//hc represents the number of high-level list elements
int lc = 0, hc = 0;
// Iterate through the bidirectional list in TreeBin, from the beginning node to the end node
for(Node<K,V> e = t.first; e ! =null; e = e.next) {
// h means to loop through the hash of the current element
int h = e.hash;
// A new TreeNode is constructed using the current node
TreeNode<K,V> p = new TreeNode<K,V>
(h, e.key, e.val, null.null);
// If the condition is true, the current loop node belongs to the low chain node
if ((h & n) == 0) {
// Condition true: Indicates that there is no data in the current low level list
if ((p.prev = loTail) == null)
lo = p;
// The current element is appended to the end of the list
else
loTail.next = p;
// Point the low list tail pointer to the p node
loTail = p;
++lc;
}
// The current node is a high-chain node
else {
if ((p.prev = hiTail) == null)
hi = p;
elsehiTail.next = p; hiTail = p; ++hc; } } ln = (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 code- 1. First of all, how to judge what comes in
nextTabIf yes, it indicates that the current thread is the first thread for capacity expansion. Preparations before capacity expansion need to be made:- Create a new array twice the size of the previous table array
Node<K,V>[] nt = (Node<K,V>[])new Node<? ,? >[n << 1]. - Assign the new table array to the property
nextTable. - Change the length of the old table array
nAssigned totransferIndexIs used below to record the overall location of the migration data. The index count starts at 1.
- Create a new array twice the size of the previous table array
- 2. Then in the for spin while loop, the current thread is assigned the bucket interval for data migration from
boundtoiBut let’s deal with thatData migration is done from the bottom upThat is, I is processed first, then I -1, I -2… Until the bound. - 3. If I is less than 0, it indicates that the current thread is not allocated to the data migration interval, and then modify it through CAS
sizeCtlThe value of the lower 16 bits is reduced by 1, which means that the current thread has no work to handle, so the number of threads is reduced by 1(sc - 2) ! = resizeStamp(n) << RESIZE_STAMP_SHIFT, includingRESIZE_STAMP_SHIFTThe value of 16, why is that? To determine whether the current thread is the last thread to expand. Because we know when we expandsizeCtlIf the current thread is not the last thread, exit the expansion method directly. Otherwise, some additional processing needs to be done, such as:- Check whether all bucket data has been migrated. If any bucket data has not been migrated, migrate it
- After all bucket data has been migrated, the new array is assigned to the property
tableAttributes,nextTableValue is set to null. - will
sizeCtlIs set to(n << 1) - (n >>> 1), that is, 2N for the next capacity expansion whose threshold is 0.75. Exit the capacity expansion method.
- 4. Then the following logic is for the task interval from
iStart toboundDuring data migration, check whether the bucket bit is empty. If so, pass the data directlycasTabAt(tab, i, null, fwd)Set the current bucket bit to FWD node - 5. If the current bucket element
hashIf the value is -1, yesFWDNode, the current bucket has been migrated, the current thread does not need to process. - 6. If the current bucket is not empty and the current bucket is not empty
FWDNode, indicating that the current bucket may be a linked list or a linked bucketTreeBinNode. Then use thesynchronized (f)Lock the current bucket node, then determine the current bucket elementhashIf the value is greater than 0, it indicates that the current bucket bit is the head node of the linked list. The nodes are connected using the high and low linked list and stored in the new hash table. - 7. If the current bucket node is
TreeBinNodes, the same high and low bidirectional linked list will be usedTreeNodeNodes are strung together and callednew TreeBin<K,V>(lo)TreeBin’s method of constructing a red-black tree using a bidirectional linked list constructs the red-black tree, which is eventually stored in a new hash table.
Data migration for capacity expansion is completed by multiple threads, as shown in the following figure
The principle of high and low linked list migration is shown in the figure below
&Operation, if(h & n) == 0Note The high hash value of this object is 0. The index of the bucket saved after capacity expansion remains unchanged, and the index of the bucket saved after capacity expansion is 1i+n, corresponding to the high-order linked list. s
HelpTransfer () of core methods to assist capacity expansion
Methods for assisting capacity expansion are as follows:
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
//nextTab references FWD. nextTable == map.nextTable theoretically.
/ / sc save map. SizeCtl
Node<K,V>[] nextTab; int sc;
// condition 1: TAB! = null constant true
// condition b :(f instanceof ForwardingNode) always true
// condition three :((ForwardingNode
)f).nexttable)! = null constant true
,v>
if(tab ! =null && (f instanceofForwardingNode) && (nextTab = ((ForwardingNode<K,V>)f).nextTable) ! =null) {
Assume 16 -> 32 Capacity expansion: 1000 0000 0001 1011
int rs = resizeStamp(tab.length);
// Condition 1: nextTab == nextTable
// Yes: The expansion is in progress
// Not available: 1. NextTable is set to Null and will be set to Null after expansion
// 2. The nextTab we got is also out of date...
// Table == TAB
// Yes: Expansion is in progress
// Not available: Expansion is complete. After expansion, the last thread to exit will set nextTable to Table
// (sc = sizeCtl) < 0
// Yes: Expansion is in progress
// No: The value of sizeCtl is greater than 0, which represents the threshold for the next capacity expansion. The current capacity expansion is complete.
while (nextTab == nextTable && table == tab &&
(sc = sizeCtl) < 0) {
// condition 1 :(sc >>> RESIZE_STAMP_SHIFT)! = rs
// true-> Indicates that the unique capacity expansion id obtained by the current thread is not the current batch capacity expansion id
// false-> Indicates that the unique identification stamp obtained by the current thread is the current batch capacity expansion
Jira: = sc == (rs << 16) + 1
// true-> The current thread does not need to participate in the capacity expansion
// false-> Expansion is still in progress, the current thread can participate
// Jira: = sc == (RS <<16) + MAX_RESIZERS
// true-> Indicates that the number of concurrent expansion threads reaches the maximum value of 655,5-1
// false-> Indicates that the current thread can participate
// transferIndex <= 0
// true-> Indicates that the global map task has been assigned, and the current thread has no work to do.
// false-> There are tasks to assign.
if((sc >>> RESIZE_STAMP_SHIFT) ! = rs || sc == rs +1 ||
sc == rs + MAX_RESIZERS || transferIndex <= 0)
break;
// Update the lower 16 bits of sizeCtl. The number of threads currently participating in the expansion is +1
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
transfer(tab, nextTab);
break; }}return nextTab;
}
return table;
}
Copy the codeThe get method
Obtain value by key as follows:
public V get(Object key) {
/ / TAB reference map. The table
//e The current element
//p target node
//n Table array length
//eh Current element hash
//ek Current element key
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
// The perturbation gives a more hashed hash value
int h = spread(key.hashCode());
//条件一：(tab = table) != null
//true-> Indicates that data has been put and the table inside the map has been initialized
//false-> Indicates that no data is put after the map is created. The table inside the map is delayed initialization and the creation logic is triggered only when data is written for the first time.
// condition two :(n = tab.length) > 0 true-> indicates that the table is initialized
(e = tabAt(TAB, (n-1) &h))! = null
//true-> The bucket bit addressed by the current key has a value
//false-> If the bucket is null, return null
if((tab = table) ! =null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) ! =null) {
// Prerequisites: Data exists on the current bucket
// Compare the hash of the header with that of the query key
// If the condition is true, the hash value of the header is the same as that of the query Key
if ((eh = e.hash) == h) {
// Compare the query key with the header key
// if the query is true, the header is the query data
if((ek = e.key) == key || (ek ! =null && key.equals(ek)))
return e.val;
}
// Conditional:
-1 FWD Indicates that the table is being expanded and the data of the bucket queried has been migrated
2.-2 The TreeBin node is queried using the find method provided by TreeBin.
else if (eh < 0)
return(p = e.find(h, key)) ! =null ? p.val : null;
// The bucket level has already formed a linked list
while((e = e.next) ! =null) {
if(e.hash == h && ((ek = e.key) == key || (ek ! =null && key.equals(ek))))
returne.val; }}return null;
}
Copy the code
The find method in FWD is implemented as follows:
/** * Find elements */ after migrating data
Node<K,V> find(int h, Object k) {
// loop to avoid arbitrarily deep recursion on forwarding nodes
// TAB must not be empty
Node<K,V>[] tab = nextTable;
outer: for (;;) {
//n indicates the length of the new table to be created for capacity expansion
//e represents the bucket header generated by the addressing algorithm when a new table is created during capacity expansion
Node<K,V> e; int n;
// Condition 1: never true
// Condition 2: never true
// Condition 3: Never true
// Condition 4: Reposition the hash header in the new expansion table
//true -> 1. In oldTable, the corresponding bucket bit is null before migration
// 2. If another writer thread exists after capacity expansion, set the bucket bit to NULL
if (k == null || tab == null || (n = tab.length) == 0 ||
(e = tabAt(tab, (n - 1) & h)) == null)
return null;
// Prerequisites: The bucket bit corresponding to the expanded table must not be null, and e is the head of the bucket bit
// What node types can e be?
/ / 1. The node type
/ / 2. TreeBin type
/ / 3. FWD type
for (;;) {
// Hash of the current node of the specified bucket after eh expansion
// Key of the node with the specified bucket in the table after ek expansion
int eh; K ek;
// If the condition is true, the data that matches the bucket in the newly expanded table is the desired data.
if((eh = e.hash) == h && ((ek = e.key) == k || (ek ! =null && k.equals(ek))))
return e;
//eh<0
TreeBin = TreeBin; FWD = TreeBin; FWD = TreeBin;
if (eh < 0) {
if (e instanceof ForwardingNode) {
tab = ((ForwardingNode<K,V>)e).nextTable;
continue outer;
}
else
The bucket is a TreeBin node. Run the treebin. find command to find the corresponding node in the red-black tree.
return e.find(h, k);
}
// Prerequisites: The current bucket header does not match the query, indicating that the bucket is a linked list
//1. Point the current element to the next element in the list
//2. Check whether the next position of the current element is empty
// true-> Returns Null after iterating to the end of the list if no corresponding data is found
if ((e = e.next) == null)
return null; }}}Copy the codeThe remove method
public V remove(Object key) {
return replaceNode(key, null.null);
}
final V replaceNode(Object key, V value, Object cv) {
// Computes the hash of the perturbed key
int hash = spread(key.hashCode());
/ / spin
for (Node<K,V>[] tab = table;;) {
//f indicates the bucket header
//n indicates the length of the current table array
// I indicates that the hash matches the bucket subscript
//fh indicates the hash of the bucket header
Node<K,V> f; int n, i, fh;
/ / CASE1:
// TAB == null true-> map.table is not initialized.. False -> Already initialized
(n = tab.length) == 0 true-> Map.table is not initialized.. False -> Already initialized
If (f = tabAt(TAB, I = (n-1) & hash)) == null true -> if (f = tabAt(TAB, I = (n-1) & hash)) == null true -> If (f = tabAt(TAB, I = (n-1) & hash)) == null
if (tab == null || (n = tab.length) == 0 ||
(f = tabAt(tab, i = (n - 1) & hash)) == null)
break;
/ / CASE2:
CASE2 to CASE3: The current bucket bit is not null
// If the condition is true, the table is being expanded and a write operation is performed. Therefore, the current thread needs to assist the table in expanding the capacity.
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
//CASE3:
CASE2 to CASE3: The current bucket bit is not null
// The current bucket bit can be either a "linked list" or a "red-black tree" TreeBin
else {
// Preserve the data reference before the replacement
V oldVal = null;
// Checkmark
boolean validated = false;
// Lock the current bucket header, after successful lock will enter the code block.
synchronized (f) {
// Check whether sync lock is the current bucket head node to prevent other threads from modifying the bucket head node before the current thread successfully locks.
// The current bucket header is still f, and other threads have not changed it.
if (tabAt(tab, i) == f) {
// The bucket is a linked list or a single node
if (fh >= 0) {
validated = true;
//e indicates the current loop processing element
//pred indicates the last node of the current loop node
Node<K,V> e = f, pred = null;
for (;;) {
// Current node key
K ek;
// Condition 1: e.hash == hash true-> Indicates that the hash of the current node is the same as that of the search node
/ / condition 2: (= e.k (ek ey) = = key | | (ek! = null && key.equals(ek)))
// If the condition is true, the key is the same as the query key.
if(e.hash == hash && ((ek = e.key) == key || (ek ! =null && key.equals(ek)))) {
// The value of the current node
V ev = e.val;
// CV == null true-> Replace the value of null then it is a delete operation
/ / condition 2: CV = = ev | | (ev! = null && CV. Equals (ev)) then it is a replacement operation
if (cv == null|| cv == ev || (ev ! =null && cv.equals(ev))) {
// Delete or replace
// Assign the value of the current node to oldVal for subsequent returns
oldVal = ev;
// This is a replacement operation
if(value ! =null)
// Direct substitution
e.val = value;
// If the condition is true, the current node is not a header
else if(pred ! =null)
// The previous node of the current node points to the next node of the current node.
pred.next = e.next;
else
// The current node is the head node. You only need to set the bucket bit to the next node of the head node.
setTabAt(tab, i, e.next);
}
break;
}
pred = e;
if ((e = e.next) == null)
break; }}// Conditional: TreeBin node.
else if (f instanceof TreeBin) {
validated = true;
// Convert to the actual type TreeBin t
TreeBin<K,V> t = (TreeBin<K,V>)f;
//r represents the root of a red-black tree
//p indicates that nodes with the same key are found in the red-black tree
TreeNode<K,V> r, p;
// (r = t.oot)! = null theoretically true
Treenode.findtreenode searches for the key (including its own node) from the current node.
// true-> The node corresponding to the key is found. Will be assigned to p
if((r = t.root) ! =null &&
(p = r.findTreeNode(hash, key, null)) != null) {
// save p.val to pv
V pv = p.val;
// CV == null
/ / condition 2: CV = = pv | | (pv! = null && cv.equals(pv)) : Indicates that the value of the pair is the same as that of the current p node
if (cv == null|| cv == pv || (pv ! =null && cv.equals(pv))) {
// Replace or delete operations
oldVal = pv;
// Replace the operation
if(value ! =null)
p.val = value;
// Delete operation
else if (t.removeTreeNode(p))
// There is no judgment here. confused
// Convert a bidirectional list to a unidirectional list
setTabAt(tab, i, untreeify(t.first));
}
}
}
}
}
When the bucket header is modified by another thread and the sync header of the current thread locks the wrong object, the validated value is false and the next for spin is entered
if (validated) {
if(oldVal ! =null) {
// Replace the value of null, indicating that the current deletion operation, oldVal! =null if yes, the deletion is successful and the current number of elements counter is updated.
if (value == null)
/ / check to 1
addCount(-1L, -1);
return oldVal;
}
break; }}}return null;
}
Copy the codeTreeBin source
The TreeBin structure is as follows:
static final class TreeBin<K.V> extends Node<K.V> {
// Red black tree root node is a red black tree
TreeNode<K,V> root;
// The first node of the bidirectional list
volatile TreeNode<K,V> first;
// wait thread (current lockState is read lockState)
volatile Thread waiter;
The write lock state is exclusive. In terms of the hash, only one writer thread actually enters the TreeBin at a time. Read lock status Read locks are shared. Multiple threads can access the TreeBin object at the same time. Each thread gives the lockState + 4 * 3. The wait state (the writer thread is waiting), and while a reader thread in the TreeBin is currently reading data, the writer thread cannot modify the data, so set the lowest 2 bits of the lockState to 0b 10 */
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
}
Copy the codeThe value of the lockState property can be in three states, as follows:
- The write lock state is exclusive. In terms of the hash, there is only one writer thread that actually enters the TreeBin at a time, with a value of 1.
- Read lock status Read locks are shared. Multiple threads can access the TreeBin object to obtain data at the same time. Every thread will give
lockState + 4 - 3. The wait state (the writer thread is waiting), when a reader thread in the TreeBin is currently reading data and the writer thread cannot modify the data, then the
lockStateThe lowest 2 bits of the0b 10
Construct a red-black tree using a bidirectional linked list in TreeNode:
/** * Creates bin with initial set of nodes headed by b. ** Creates bin with initial set of nodes headed by b
TreeBin(TreeNode<K,V> b) {
// Setting hash to -2 indicates that the node is a TreeBin node
super(TREEBIN, null.null.null);
// Use first to reference the treeNode list
this.first = b;
//r red black tree root reference
TreeNode<K,V> r = null;
//x represents the current node traversed
for(TreeNode<K,V> x = b, next; x ! =null; x = next) {
next = (TreeNode<K,V>)x.next;
// Forces the left and right subtrees of the currently inserted node to be null
x.left = x.right = null;
// If this condition is true, the current red-black tree is an empty tree, so set the inserted element to the root node
if (r == null) {
// The parent of the root node must be null
x.parent = null;
// Change the color to black
x.red = false;
// let r refer to the object to which x points.
r = x;
}
else {
// If it's not the first time, the else branch will be added, and the red-black tree will already have data
//k indicates the key of the node to be inserted
K k = x.key;
//h indicates the hash of the node to be inserted
int h = x.hash;
//kc indicates the class type of the key of the inserted nodeClass<? > kc =null;
//p represents a temporary node to find the parent of the inserted node
TreeNode<K,V> p = r;
for (;;) {
//dir (-1, 1)
//-1 indicates that the hash value of the inserted node is greater than that of the current node P
//1 indicates that the hash value of the inserted node is smaller than that of the current node P
//ph p represents a hash for a temporary node that looks for the parent node of the inserted node
int dir, ph;
// Temporary node key
K pk = p.key;
// The hash value of the inserted node is smaller than that of the current node
if ((ph = p.hash) > h)
// Inserting a node may require inserting it into the left child of the current node or continuing the search in the left child tree
dir = -1;
// The hash value of the inserted node is greater than that of the current node
else if (ph < h)
// To insert a node, you may need to insert it into the right child of the current node or continue searching in the right child tree
dir = 1;
// If CASE3 is executed, the hash of the current inserted node is the same as that of the current node, and the final hash will be sorted in CASE3. In the end
// Get dir not 0, (-1, 1)
else if ((kc == null &&
(kc = comparableClassFor(k)) == null) ||
(dir = compareComparables(kc, k, pk)) == 0)
dir = tieBreakOrder(k, pk);
// XP wants to represent the parent of the inserted node
TreeNode<K,V> xp = p;
// If the condition is true, node P is the parent node of the node to be inserted
// If the condition is not true, it indicates that there are layers under the p node. You need to point P to the left or right child node of P to continue the search.
if ((p = (dir <= 0)? p.left : p.right) ==null) {
// Set the parent of the node to be the current node
x.parent = xp;
// It is smaller than the P node and needs to be inserted into the left child of the P node
if (dir <= 0)
xp.left = x;
// It is larger than P and needs to be inserted into the right child of P
else
xp.right = x;
// After inserting nodes, the red-black tree nature may be broken, so you need to call the balancing method
r = balanceInsertion(r, x);
break; }}}}// Assign r to the root reference of the TreeBin object.
this.root = r;
assert checkInvariants(root);
}
Copy the codeThe final TreeBin structure is shown below
In the above get method, it is possible to call FWD’s find method as well as TreeBin’s find method, as follows:
final Node<K,V> find(int h, Object k) {
if(k ! =null) {
//e indicates that the current node of the loop iterates over the list referenced by first
for(Node<K,V> e = first; e ! =null;) {//s stores the temporary lock state
// Key of the current node in the EK list
int s; K ek;
//(WAITER|WRITER) => 0010 | 0001 => 0011
//lockState & 0011 ! If the = 0 condition is true, the current TreeBin has a waiting thread or a write thread is currently locking
// If the tree is locked and not blocked, query the bidirectional list. That's what bidirectional lists do
if(((s = lockState) & (WAITER|WRITER)) ! =0) {
if(e.hash == h && ((ek = e.key) == k || (ek ! =null && k.equals(ek))))
return e;
e = e.next;
}
// Preconditions: There are no waits or writers in the current TreeBin
// If the condition is true, the read lock is successfully added
else if (U.compareAndSwapInt(this, LOCKSTATE, s,
s + READER)) {
TreeNode<K,V> r, p;
try {
// Query the red-black tree operation
p = ((r = root) == null ? null :
r.findTreeNode(h, k, null));
} finally {
//w represents the wait thread
Thread w;
//U.getAndAddInt(this, LOCKSTATE, -READER) == (READER|WAITER)
//1. The current thread is finished querying the red-black tree, and the current thread's read lock is released by setting the lockstate to -4
/ / (READER | WAITER) = 0110 = > said that the current is only one thread is reading, and "there is a thread waiting for"
// The current reader thread is the last one in the TreeBin.
//2.(w = waiter) ! = null indicates that a writer thread is waiting for the read operation to complete. That is, the lockSupport. park method is called.
if (U.getAndAddInt(this, LOCKSTATE, -READER) == (READER|WAITER) && (w = waiter) ! =null)
// Use unpark to restore the writer thread to its running state.
LockSupport.unpark(w);
}
returnp; }}}return null;
}
Copy the code-
In the find method above, if the red-black tree in the current TreeBin has a waiting thread (the writer thread is waiting) or has a write lock, the data can only be read from the bidirectional list, otherwise it can be read directly from the red-black tree.
-
Otherwise, run the CAS command to change the lockState value to lockState + 4, indicating that a read lock is added. Then, start from the root node in the red-black tree and compare the hash value to find the corresponding node.
-
After the node is found using the second method, lockstate-4 is used to determine if only the writer thread is currently waiting. If so, call locksupport.unpark (w); Wake up the writer thread
Inserts a node putTreeVal method into a red-black tree and returns the node if its key is the same as the one to be inserted
/**
* Finds or adds a node.
* @returnNull if added * Adds a node to the red-black tree. Returns the node */ if the node has the same key as the one to be added
final TreeNode<K,V> putTreeVal(int h, K k, V v) { Class<? > kc =null;
boolean searched = false;
for (TreeNode<K,V> p = root;;) {
int dir, ph; K pk;
if (p == null) {
first = root = new TreeNode<K,V>(h, k, v, null.null);
break;
}
else if ((ph = p.hash) > h)
dir = -1;
else if (ph < h)
dir = 1;
else if((pk = p.key) == k || (pk ! =null && k.equals(pk)))
return p;
else if ((kc == null &&
(kc = comparableClassFor(k)) == null) ||
(dir = compareComparables(kc, k, pk)) == 0) {
if(! searched) { TreeNode<K,V> q, ch; searched =true;
if(((ch = p.left) ! =null&& (q = ch.findTreeNode(h, k, kc)) ! =null) || ((ch = p.right) ! =null&& (q = ch.findTreeNode(h, k, kc)) ! =null))
return q;
}
dir = tieBreakOrder(k, pk);
}
TreeNode<K,V> xp = p;
if ((p = (dir <= 0)? p.left : p.right) ==null) {
// The current cycle node xp is the parent of node X
//x indicates the node to be inserted
//f old header
TreeNode<K,V> x, f = first;
// The new node points to the old node
first = x = new TreeNode<K,V>(h, k, v, f, xp);
// The list has data
if(f ! =null)
// Set the prefix reference of the old header to the current header.
f.prev = x;
if (dir <= 0)
xp.left = x;
else
xp.right = x;
if(! xp.red) x.red =true;
else {
// Indicates that after a new node is inserted, the new node and its parent node form a "red red connection".
// Lock the red black tree
lockRoot();
try {
// Balance the red-black tree to make it conform to the specification again.
root = balanceInsertion(root, x);
} finally {
// Release the red black treeunlockRoot(); }}break; }}assert checkInvariants(root);
return null;
}
/** * Acquires write lock for tree restructuring. */
private final void lockRoot(a) {
// lockState is not 0, indicating that another reader is reading data in the treeBin red-black tree.
if(! U.compareAndSwapInt(this, LOCKSTATE, 0, WRITER))
contendedLock(); // offload to separate method
}
/** * Releases write lock for tree restructuring. */
private final void unlockRoot(a) {
lockState = 0;
}
/** * Possibly blocks awaiting root lock. * Currently a reader thread is reading the red-black tree */
private final void contendedLock(a) {
boolean waiting = false;
// represents the lock value
int s;
/ / spin
for (;;) {
//~WAITER = 11111.... 01
// Condition true: No reader thread in TreeBin is currently accessing the red-black tree
// Condition not true: a thread is accessing the red black tree
if (((s = lockState) & ~WAITER) == 0) {
// If the condition is true, the writer thread succeeded in preempting the lock
if (U.compareAndSwapInt(this, LOCKSTATE, s, WRITER)) {
if (waiting)
// Set the TreeBin object to null
waiter = null;
return; }}//lock & 0000... If the waiter flag is set to 0, the current thread can be set to 1 and suspended.
else if ((s & WAITER) == 0) {
if (U.compareAndSwapInt(this, LOCKSTATE, s, s | WAITER)) {
waiting = true; waiter = Thread.currentThread(); }}// true: the current thread has set treebin.waiter to the current thread in CASE2 and set the lockState to 1
// At this point, the current thread is suspended.
else if (waiting)
LockSupport.park(this); }}Copy the codeAt the end of the insertion insertion method, lockRoot is required to obtain write locks through cas (lockState) and then release the locks at the end of the insertion method. Otherwise, call locksupport.park (this) to suspend.