- HashMap в jdk1.7 и jdk1.8 был представлен ранее [Коллекция JAVA: углубленный анализ HashMap (сравнение версий)], проанализировано далее в статьеHashMap не является потокобезопаснымпричина, то какПоточно-ориентированная Hashtable и ConcurrentHashMapКак добиться потокобезопасности?
Hashtable
/**
* @since JDK1.0
*/
public class Hashtable<K,V>
extends Dictionary<K,V>
implements Map<K,V>, Cloneable, java.io.Serializable
- Как видно из исходного кода в jdk1.8 выше, Hashtable доступен из jdk1.0 и является потокобезопасным. Позже HashMap был создан, потому что Hashtable был слишком неэффективным. Чтобы добиться эффективности, HashMap удалил гарантию потокобезопасныйsynchronizedключевые слова.
Основное различие между Hashtable и HashMap
- Размер по умолчанию: размер Hashtable по умолчанию равен 11, и может быть установлено значение больше 0. Размер HashMap по умолчанию равен 16. Учитывая значение, оно будет инициализировано до наименьшего кратного 2 большего, чем заданное значение (см. ссылку выше для HashMap).
public Hashtable() {
this(11, 0.75f);
}
public Hashtable(int initialCapacity, float loadFactor) {
if (initialCapacity < 0)
throw new IllegalArgumentException("Illegal Capacity: "+
initialCapacity);
if (loadFactor <= 0 || Float.isNaN(loadFactor))
throw new IllegalArgumentException("Illegal Load: "+loadFactor);
if (initialCapacity==0)
initialCapacity = 1;
this.loadFactor = loadFactor;
table = new Entry<?,?>[initialCapacity];
threshold = (int)Math.min(initialCapacity * loadFactor, MAX_ARRAY_SIZE + 1);
}
- Расширение: расширение Hashtable в 2 раза превышает существующую емкость + 1, расширение HashMap в 2 раза превышает существующую емкость.
int newCapacity = (oldCapacity << 1) + 1;
- Причины безопасности потоков Hashtable:
public synchronized int size() {略}
public synchronized boolean isEmpty(){}
public synchronized boolean contains(Object value) {}
public synchronized boolean containsKey(Object key) {}
public synchronized V get(Object key) {}
public synchronized V put(K key, V value) {略}
public synchronized V remove(Object key) {}
public synchronized void putAll(Map<? extends K, ? extends V> t) {}
public synchronized void clear() {}
public synchronized Object clone() {}
public synchronized int hashCode() {}
..........
从源码中可以看出，Hashtable的方法上基本上都加上了synchronized关键字，而当一个线程访问加了synchronized关键字的方法时，会先获得实例对象的锁，而其他线程就得不到对象的锁，也就不能访问加了synchronized关键字的方法，这就相当于此线程Блокирует всю таблицу объекта экземпляра Hashtable.，从而使Hashtable是线程安全的。
ConcurrentHashMap
ConcurrentHashMap同样是线程安全的，但是却比Hashtable效率要高，是如何做到的呢？
Кратко поговорим о классе sun.misc.Unsafe. Класс Unsafe может напрямую манипулировать памятью, и все операции являются атомарными.
- Метод objectFieldOffset получает смещение свойств объекта в памяти.
- Метод putObject(Object var1, long var2, Object var4) устанавливает значение по адресу объекта var1, смещение var2 на var4.
- Метод getObject(object,offset) получает значение атрибута по адресу объекта object offset offset
- Метод getObjectVolatile(object,offset) получает значение атрибута по адресу объекта object offset offset (атрибут оформлен с помощью Volatile)
CAS (атомарная операция)
cas, Compare и Swap — это сравнение и обмен. Методы cas в Unsafe:
public final native boolean compareAndSwapObject(Object var1, long var2, Object var4, Object var5);
public final native boolean compareAndSwapInt(Object var1, long var2, int var4, int var5);
public final native boolean compareAndSwapLong(Object var1, long var2, long var4, long var6);
Метод compareAndSwapObject имеет 4 параметра, объект var1, значение смещения var2, ожидаемое значение var4, измененное значение var5, если значение по адресу памяти var1, смещенному на var2, равно var4, то значение в памяти равно изменяется на var5 и возвращает true, в противном случае возвращает false.
JDK1.7
структура
В jdk1.7 ConcurrentHashMap состоит из сегментов, каждый сегмент имеет таблицу, а в таблице есть связанный список, то есть вся таблица в HashMap разбита на несколько сегментов, а отдельные сегменты блокируются при мульти- многопоточные операции.Вместо блокировки всей таблицы, как в HashTable, гранулярность лучше, одновременно может работать больше потоков, а эффективность выше.
определение
public class ConcurrentHashMap<K, V> extends AbstractMap<K, V>
implements ConcurrentMap<K, V>, Serializable
ConcurrentHashMap наследуется от класса AbstractMap и реализует интерфейсы ConcurrentMap и Serializable.
Переменные-члены
- Значение размера инициализации по умолчанию 16
static final int DEFAULT_INITIAL_CAPACITY = 16;
- Размер коэффициента нагрузки по умолчанию 0,75
static final float DEFAULT_LOAD_FACTOR = 0.75f;
- Количество сегментов по умолчанию (максимальное количество одновременных потоков)
static final int DEFAULT_CONCURRENCY_LEVEL = 16;
- Максимальная емкость
static final int MAXIMUM_CAPACITY = 1 << 30;
- Минимальная вместимость стола в каждом сегменте
static final int MIN_SEGMENT_TABLE_CAPACITY = 2;
- Максимальное количество сегментов
static final int MAX_SEGMENTS = 1 << 16;
- Количество попыток метода containsValue без блокировки таблицы
static final int RETRIES_BEFORE_LOCK = 2;
segment
segment继承了ReentrantLock，就有了加锁和解锁的方法。
static final class Segment<K,V> extends ReentrantLock implements Serializable {
private static final long serialVersionUID = 2249069246763182397L;
//自旋等待尝试加锁次数，单核为1，多核为64，Runtime.getRuntime().availableProcessors()方法获取CPU核心数
static final int MAX_SCAN_RETRIES =
Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;
//表，即HashEntry数组（每个segment中都有一个table）
transient volatile HashEntry<K,V>[] table;
//segment中元素个数
transient int count;
//修改次数
transient int modCount;
//扩容阀值
transient int threshold;
//负载因子
final float loadFactor;
/**
* 构造函数
*/
Segment(float lf, int threshold, HashEntry<K,V>[] tab) {
this.loadFactor = lf;
this.threshold = threshold;
this.table = tab;
}
}
- положить метод
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
/**
* 尝试进行加锁，如果加锁失败，则执行scanAndLockForPut方法，尝试加锁一定次数之后调用线程自中断方法（自旋等待）。
*/
HashEntry<K,V> node = tryLock() ? null :
scanAndLockForPut(key, hash, value);
V oldValue;
try {
HashEntry<K,V>[] tab = table;
int index = (tab.length - 1) & hash;
HashEntry<K,V> first = entryAt(tab, index);
for (HashEntry<K,V> e = first;;) { //for循环查找key是否存在，如果找到了，替换value值，返回oldValue
if (e != null) {
K k;
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
oldValue = e.value;
if (!onlyIfAbsent) {
e.value = value;
++modCount;
}
break;
}
e = e.next;
}
else { //如果没找到，新建HashEntry节点，放到first节点前面
if (node != null) //scanAndLockForPut自旋等待时如果已经新建了节点，设置next值即可，setNext方法实现了延迟写。
node.setNext(first);
else
node = new HashEntry<K,V>(hash, key, value, first);
int c = count + 1; //元素个数加1，如果超过了阀值，则进行rehash，进行扩容
if (c > threshold && tab.length < MAXIMUM_CAPACITY)
rehash(node);
else
setEntryAt(tab, index, node); //把新建节点放在链表的头位置
++modCount;
count = c;
oldValue = null;
break;
}
}
} finally {
unlock(); //最后释放锁
}
return oldValue;
}
- метод scanAndLockForPut, спин-блокировка, попытка блокировки определенное количество раз и все еще не может прервать поток.Этот метод сначала вычисляет позицию хеш-значения в таблице, зацикливает связанный список в этой позиции, чтобы найти значение ключа, если он не существует, создает новый узел, а затем пытается заблокировать MAX_SCAN_RETRIES раз и приостанавливает текущий поток, если он все время терпит неудачу. Если тем временем изменяется глава связанного списка, процесс перезапускается.
private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
HashEntry<K,V> first = entryForHash(this, hash);
HashEntry<K,V> e = first;
HashEntry<K,V> node = null;
int retries = -1;
while (!tryLock()) { //获取锁失败时进入循环
HashEntry<K,V> f;
if (retries < 0) { //循环链表，找到key值或者不存在新建节点
if (e == null) {
if (node == null) // speculatively create node
node = new HashEntry<K,V>(hash, key, value, null);
retries = 0;
}
else if (key.equals(e.key)) //如果找到了key值，
retries = 0;
else
e = e.next;
}
/**
* 找到key值或者key值不存在新建节点之后，尝试加锁一定次数进入等待状态
* 尝试次数，单核为1，多核为64
*/
else if (++retries > MAX_SCAN_RETRIES) {
lock();
break;
}
else if ((retries & 1) == 0 &&
(f = entryForHash(this, hash)) != first) { //如果尝试加锁过程中发现链表头变化了，重置retries为-1，重新开始
e = first = f;
retries = -1;
}
}
return node;
}
- Метод rehash, расширяет текущую таблицу, размер становится вдвое больше исходного размера, элементы в ней будут переназначены, элементы в связанном списке на oldTable[idx] могут быть перехешированы в newTable[idx] и newTbale[idx] +n] в связанном списке, n — размер oldTable.
private void rehash(HashEntry<K,V> node) {
HashEntry<K,V>[] oldTable = table;
int oldCapacity = oldTable.length;
int newCapacity = oldCapacity << 1; //newTable的大小为oldTable的2被
threshold = (int)(newCapacity * loadFactor);
HashEntry<K,V>[] newTable =
(HashEntry<K,V>[]) new HashEntry[newCapacity];
int sizeMask = newCapacity - 1;
for (int i = 0; i < oldCapacity ; i++) {
HashEntry<K,V> e = oldTable[i];
if (e != null) {
HashEntry<K,V> next = e.next;
int idx = e.hash & sizeMask; //计算节点在newTable中的位置idx
if (next == null) //如果链表只有一个节点，直接放到newTable的idx上
newTable[idx] = e;
else {
/**
* 与重新计算每个节点在newTable中的位置并依次进行头插法插入链表头相比，这里进行了优化
* 1.计算链表中每个节点在newTable中的位置，但是并不立即插入链表头
* 2.记住最后一个与它的上一个节点在新表中位置不同的节点lastRun，即链表中此节点之后的节点在newTable中的位置都相同
* 3.把lastRun放到newTable中，它之后的节点会带过来
* 4.计算lastRun之前的节点在newTable中的位置并依次进行头插法插入newTable中。
*/
HashEntry<K,V> lastRun = e;
int lastIdx = idx;
for (HashEntry<K,V> last = next;
last != null;
last = last.next) {
int k = last.hash & sizeMask;
if (k != lastIdx) {
lastIdx = k;
lastRun = last;
}
}
newTable[lastIdx] = lastRun;
// Clone remaining nodes
for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {
V v = p.value;
int h = p.hash;
int k = h & sizeMask;
HashEntry<K,V> n = newTable[k];
newTable[k] = new HashEntry<K,V>(h, p.key, v, n);
}
}
}
}
int nodeIndex = node.hash & sizeMask; // 把新节点放入newTable中
node.setNext(newTable[nodeIndex]);
newTable[nodeIndex] = node;
table = newTable;
}
- remove, сначала попытайтесь получить блокировку, если блокировка не удалась, scanAndLock вращается и ждет (аналогично методу put выше), после получения блокировки (tab.length - 1) & hash вычисляет индекс удаленного узла в таблице, если связанный список в этой позиции в таблице не пуст, зациклить, чтобы определить, равен ли узел в связанном списке удаленному узлу (когда значение равно нулю, ключ равен, в противном случае ключ и значение должен быть равен), если удаленный узел существует, установите указатель следующего узла перед узлом, который может указывать на следующий узел.
final V remove(Object key, int hash, Object value) {
if (!tryLock()) //尝试加锁
scanAndLock(key, hash); //加锁失败，则自旋等待
V oldValue = null;
try {
HashEntry<K,V>[] tab = table;
int index = (tab.length - 1) & hash; //计算hash值在table中的下标
HashEntry<K,V> e = entryAt(tab, index);
HashEntry<K,V> pred = null;
while (e != null) {
K k;
HashEntry<K,V> next = e.next;
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
V v = e.value;
if (value == null || value == v || value.equals(v)) { //key相等时，value为null或者value也相等即为删除节点
if (pred == null)
setEntryAt(tab, index, next); //如果删除节点是头节点，设置头节点为next节点
else
pred.setNext(next); //否则设置上一个节点的next指针指向next节点
++modCount; //修改次数加1
--count; //节点数量减1
oldValue = v;
}
break;
}
pred = e;
e = next;
}
} finally {
unlock(); //释放锁
}
return oldValue;
}
Метод хеширования использует операции смещения и XOR, чтобы сделать распределение значений k в сегменте максимально равномерным.
private int hash(Object k) {
int h = hashSeed;
if ((0 != h) && (k instanceof String)) {
return sun.misc.Hashing.stringHash32((String) k);
}
h ^= k.hashCode();
h += (h << 15) ^ 0xffffcd7d;
h ^= (h >>> 10);
h += (h << 3);
h ^= (h >>> 6);
h += (h << 2) + (h << 14);
return h ^ (h >>> 16);
}
метод put (метод put ConcurrentHashMap очень прост, сначала вычислите, в каком сегменте находится значение ключа, а затем вызовите метод put сегмента)
public V put(K key, V value) {
Segment<K,V> s;
if (value == null)
throw new NullPointerException();
int hash = hash(key);
int j = (hash >>> segmentShift) & segmentMask; //计算key落在哪个segment中
if ((s = (Segment<K,V>)UNSAFE.getObject
(segments, (j << SSHIFT) + SBASE)) == null)
s = ensureSegment(j); //如果segment不存在则初始化
return s.put(key, hash, value, false); //调用segment的put方法
}
метод get сначала вычисляет в какой сегмент попадает ключ, если сегмент не нулевой и таблица не нулевая, tab.length - 1)&h вычисляет индекс в таблице, и сравниваются узлы кругового связанного списка , если ключи равны или Методы hash и equals равны, и возвращается значение.
public V get(Object key) {
Segment<K,V> s; // manually integrate access methods to reduce overhead
HashEntry<K,V>[] tab;
int h = hash(key);
long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
(tab = s.table) != null) {
for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile
(tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);
e != null; e = e.next) {
K k;
if ((k = e.key) == key || (e.hash == h && key.equals(k))) //如果key相等，或者重载的hash方法和equals方法相等
return e.value;
}
}
return null;
}
метод удаления (существует два метода удаления, один параметр является только ключом, а другой — ключом и значением, поэтому, когда значение в методе удаления сегмента равно нулю, ключи равны.)
public V remove(Object key) {
int hash = hash(key); //计算hash值
Segment<K,V> s = segmentForHash(hash); //计算hash值落在哪个segment中
return s == null ? null : s.remove(key, hash, null); //调用segment的remove方法
}
/**
* {@inheritDoc}
*
* @throws NullPointerException if the specified key is null
*/
public boolean remove(Object key, Object value) {
int hash = hash(key);
Segment<K,V> s;
return value != null && (s = segmentForHash(hash)) != null &&
s.remove(key, hash, value) != null;
}
JDK1.8
В jdk1.8 ConcurrentHashMap претерпел большие изменения: он больше не является сегментной структурой, а использует метод, аналогичный оптимистической блокировке, для достижения многопоточной безопасности.
определение
ConcurrentHashMap наследует AbstractMap и реализует интерфейс ConcurrentMap.
public class ConcurrentHashMap<K,V> extends AbstractMap<K,V>
implements ConcurrentMap<K,V>, Serializable
постоянный
- Максимальная емкость
private static final int MAXIMUM_CAPACITY = 1 << 30;
- Инициализированная емкость по умолчанию равна 16, емкость должна быть кратна 2, а минимум равен 1.
private static final int DEFAULT_CAPACITY = 16;
- Максимальная емкость массива, используемая toArray и родственными методами
static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
- Количество сегментов по умолчанию (Больше не используется, совместим со старыми версиями)
private static final int DEFAULT_CONCURRENCY_LEVEL = 16;
- Коэффициент нагрузки по умолчанию
private static final float LOAD_FACTOR = 0.75f;
- Порог преобразования связанного списка в дерево
static final int TREEIFY_THRESHOLD = 8;
- Порог преобразования дерева в связанный список
static final int UNTREEIFY_THRESHOLD = 6;
- Минимальная емкость древовидной структуры (когда длина одного из связанных списков в таблице достигает 8, а общее количество узлов в таблице достигает 64, связанный список будет преобразован в древовидную структуру, а если количество узлов в таблице меньше 64, древовидная структура выполняться не будет. Вместо этого таблица расширяется для уменьшения длины связанного списка.)
static final int MIN_TREEIFY_CAPACITY = 64;
- Количество интервалов для каждого переноса ядра при масштабировании
private static final int MIN_TRANSFER_STRIDE = 16;
private static int RESIZE_STAMP_BITS = 16;
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
- Хэш-значение узла
static final int MOVED = -1; // 表示该节点正在处理中
static final int TREEBIN = -2; // 表示该节点是树的根节点
static final int RESERVED = -3; // 暂时保留
static final int HASH_BITS = 0x7fffffff; // 正常节点的hash值可用的位数
- Количество ядер процессора
static final int NCPU = Runtime.getRuntime().availableProcessors();
Атрибуты
- table таблица, volatile модификация (когда поток модифицирует атрибут, он будет немедленно записан в основную память, то есть сразу же будет виден другим потокам), переходный модификатор (этот атрибут игнорируется при сериализации, то есть атрибут существует только в память, а не сохраняется на диск)
transient volatile Node<K,V>[] table;
- newTable, новая таблица будет создана во время расширения, и другие потоки обнаружат, что таблица не пуста, что указывает на то, что есть потоки, уже выполняющие операции расширения, которые помогут расширить данные в oldTable в эту новую таблицу и завершить операцию расширения. вместе.
private transient volatile Node<K,V>[] nextTable;
- Одно из свойств baseCount, используемое для расчета размера
private transient volatile long baseCount;
- Свойства, управляющие инициализацией и расширением таблицы
- 0 , значение инициализации
- -1, означает инициализацию
- -N, указывающий, что потоки N-1 расширяются вместе
- N , когда таблица равна нулю, значение представляет собой инициализированный размер, если таблица не равна нулю, значение представляет размер следующего расширения.
private transient volatile int sizeCtl;
- Индекс следующей таблицы при расширении
private transient volatile int transferIndex;
- Идентификатор блокировки для расширения и CounterCells
private transient volatile int cellsBusy;
private transient volatile CounterCell[] counterCells;
private transient KeySetView<K,V> keySet;
private transient ValuesView<K,V> values;
private transient EntrySetView<K,V> entrySet;
узел
- Узел узла
static class Node<K,V> implements Map.Entry<K,V> {
final int hash; //hash值
final K key; //key值
volatile V val; //value值
volatile Node<K,V> next; //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() { return key; }
public final V getValue() { return val; }
public final int hashCode() { return key.hashCode() ^ val.hashCode(); }
public final String toString(){ 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 instanceof Map.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;
}
}
- Узел ForwardingNode (структура головного узла связанного списка, с которым работает поток при раскрытии), переписать метод find
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
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;
for (;;) {
int eh; K ek;
if ((eh = e.hash) == h &&
((ek = e.key) == k || (ek != null && k.equals(ek))))
return e;
if (eh < 0) {
if (e instanceof ForwardingNode) {
tab = ((ForwardingNode<K,V>)e).nextTable;
continue outer;
}
else
return e.find(h, k);
}
if ((e = e.next) == null)
return null;
}
}
}
}
Конструктор
Значения sizeCtl, устанавливаемые конструкторами разных параметров, неодинаковы, и при инициализации таблицы размер таблицы тоже разный.
public ConcurrentHashMap() {
}
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;
}
public ConcurrentHashMap(Map<? extends K, ? extends V> m) {
this.sizeCtl = DEFAULT_CAPACITY;
putAll(m);
}
public ConcurrentHashMap(int initialCapacity, float loadFactor) {
this(initialCapacity, loadFactor, 1);
}
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {
if (!(loadFactor > 0.0f) || 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);
this.sizeCtl = cap;
}
основной метод
- CAS, jdk1.8 в основном использует три операции cas для обеспечения безопасности потоков, эти три методаатомарностьработать.
/**
* tabAt,读取tab[i]的值
*/
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);
}
/**
* casTabAt，如果tab[i]的值等于c，用v替换c并返回true，否则返回false
*/
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);
}
/**
* setTabAt，设置tab[i]=v
*/
static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
}
- tableSizeFor, вычисляет размер таблицы, возвращает наименьшее кратное 2, большее или равное заданному целому числу
private static final int tableSizeFor(int c) {
int n = c - 1;
n |= n >>> 1; //把最前面的1复制到第二位，使下一位也为1。例0100变为0110
n |= n >>> 2; //把上一步的最前面的两位1复制到下面两位，使后两位也为1。例01100010变为01111010
n |= n >>> 4; //...
n |= n >>> 8;
n |= n >>> 16;
return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
}
- положить метод
public V put(K key, V value) {
return putVal(key, value, false);
}
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException(); //ConcurrentHashMap的key和value都不能为null
int hash = spread(key.hashCode());
int binCount = 0;
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
if (tab == null || (n = tab.length) == 0)
tab = initTable(); //如果table为null或者length为0，则进行初始化操作
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { //如果链表头为null，新建节点
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break; //如果多个线程同时执行casTabAt，因为是原子性操作，所以只有一个线程成功并结束for循环，其他线程继续for循环
}
else if ((fh = f.hash) == MOVED) //如果链表头节点的hash值为-1，说明table可能正在进行扩容，调用helpTransfer方法帮助扩容
tab = helpTransfer(tab, f);
else {
V oldVal = null;
synchronized (f) { //这里使用synchronized关键字，对链表头节点f加锁，从而实现多线程安全
if (tabAt(tab, i) == f) { //加锁之后需要进行再判断一次，保证f在加锁之前没有被其他线程修改
if (fh >= 0) {
binCount = 1; //计算链表的长度
for (Node<K,V> e = f;; ++binCount) { //循环链表，如果找到key则替换，否则新建节点
K ek;
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;
if (!onlyIfAbsent)
e.val = value;
break;
}
Node<K,V> pred = e;
if ((e = e.next) == null) {
pred.next = new Node<K,V>(hash, key,
value, null);
break;
}
}
}
else if (f instanceof TreeBin) { //如果头节点是树结构的，则调用putTreeVal方法
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 (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD) //如果链表长度大于等于8，则转化为树结构
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount); //节点数量+1
return null;
}
static final int spread(int h) {
return (h ^ (h >>> 16)) & HASH_BITS;
}
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
Node<K,V>[] nextTab; int sc;
if (tab != null && (f instanceof ForwardingNode) &&
(nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) { //如果table正在进行扩容
int rs = resizeStamp(tab.length); //扩容标识
while (nextTab == nextTable && table == tab &&
(sc = sizeCtl) < 0) {
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || //如果状态变化了，说明扩容结束
sc == rs + MAX_RESIZERS || transferIndex <= 0)
break;
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) { //sizCtl加1，多1个线程同时扩容
transfer(tab, nextTab);
break;
}
}
return nextTab;
}
return table;
}
- метод инициализации initTable
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
if ((sc = sizeCtl) < 0)
Thread.yield(); // 如果有其他线程已经开始初始化了，则释放cpu资源，等待其他线程初始化
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try { //初始化，设置sizeCtl为-1，表示有线程正在初始化
if ((tab = table) == null || tab.length == 0) {
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
sc = n - (n >>> 2); //扩容阈值0.75*n
}
} finally {
sizeCtl = sc; //设置sizeCtl为扩容阈值
}
break;
}
}
return tab;
}
- получить метод
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
int h = spread(key.hashCode());
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
if ((eh = e.hash) == h) { //判断头节点
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
else if (eh < 0) //如果头节点的hash<0，表明该节点是ForwardingNode节点或者树节点，调用子类的find方法
return (p = e.find(h, key)) != null ? p.val : null;
while ((e = e.next) != null) { //循环查找链表中的节点
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}
- метод расширения передачи
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // 转移节点时，下标跨越的步幅
if (nextTab == null) { // 初始化nextTab
try {
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;
transferIndex = n; //从下标n开始转移节点到新表
}
int nextn = nextTab.length;
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
boolean advance = true; //已处理标识
boolean finishing = false; // 结束标识
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
while (advance) { //如果该table[i]处理过，则--i
int nextIndex, nextBound;
if (--i >= bound || finishing)
advance = false;
else if ((nextIndex = transferIndex) <= 0) { //下一个要处理的下标<=0，跳出while循环
i = -1;
advance = false;
}
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) { //处理下一个范围
bound = nextBound;
i = nextIndex - 1;
advance = false;
}
}
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
if (finishing) { //如果结束，nextTable置空，table变成新表，sizeCtl扩容阈值
nextTable = null;
table = nextTab;
sizeCtl = (n << 1) - (n >>> 1);
return;
}
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) { //当前线程扩容结束，sizeCtl减1
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) //如果不等，说明还有其他线程没有结束扩容
return;
finishing = advance = true;
i = n; // recheck before commit
}
}
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
else if ((fh = f.hash) == MOVED)
advance = true; // 该链表已经在处理中
else {
synchronized (f) { //对链表头节点进行加锁
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn; //ln表示放在下标i的头节点，hn表示放在下标i+n的头节点
if (fh >= 0) { //fh>=0表示是Node类型节点
/**
* n是2的倍数
* fh & (n-1)计算hash值在oldTable中的下标i，
* fh & n==0表示在newTable中下标仍为i，
* fh & n==1表示在newTable中的下标为i + n
*/
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) { //如果为0，表示在新表中的下标仍为i
ln = lastRun;
hn = null;
}
else { //否则下标为i+n
hn = lastRun;
ln = null;
}
for (Node<K,V> p = f; p != lastRun; p = p.next) { //把lastRun之前的节点用头插法插入ln和hn为头节点的两条链表中
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); //ln为头节点的链表放入newTable[i]中
setTabAt(nextTab, i + n, hn); //hn为头节点的链表放入newTable[i+n]中
setTabAt(tab, i, fwd); //oldTable[i]修改为ForwardingNode节点，表示已处理
advance = true; //已处理标识
}
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;
/**
* 循环链表（ConcurrentHashMap中的树节点也有next指针，也是一条链表）
* lo表示将要放到下标i的链表的头节点，loTail用来构建新链表，lc表示链表节点数量
* ho表示将要放到下标i的链表的头节点，hoTail用来构建新链表，hc表示链表节点数量
*/
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;
else
hiTail.next = p;
hiTail = p;
++hc;
}
}
/**
* 判断两条子链表的长度如果小于等于6，则从树结构转换为链表结构
* 如果大于6，需要新建两个子树
* 如果其中一个没有节点，直接用原先的树结构即可
*/
ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
(hc != 0) ? new TreeBin<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;
}
}
}
}
}
}
- метод удаления
public V remove(Object key) {
return replaceNode(key, null, null);
}
final V replaceNode(Object key, V value, Object cv) {
int hash = spread(key.hashCode()); //计算hash值
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
if (tab == null || (n = tab.length) == 0 ||
(f = tabAt(tab, i = (n - 1) & hash)) == null) //如果表是空的或者链表是空的，结束
break;
else if ((fh = f.hash) == MOVED) //如果链表头节点的hash值是-1，说明正在处理中，如果在扩容，则帮助扩容
tab = helpTransfer(tab, f);
else {
V oldVal = null;
boolean validated = false;
synchronized (f) { //对链表头节点加锁
if (tabAt(tab, i) == f) {
if (fh >= 0) { //如果是Node节点
validated = true;
for (Node<K,V> e = f, pred = null;;) { //循环链表查找key值
K ek;
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) { //如果key值相等
V ev = e.val;
if (cv == null || cv == ev ||
(ev != null && cv.equals(ev))) { //如果给定的value为null或者value也相等
oldVal = ev;
if (value != null) //如果给定value值不为null，替换value
e.val = value;
else if (pred != null) //前一个节点的next指向下一个节点，删除当前节点
pred.next = e.next;
else
setTabAt(tab, i, e.next); //前一个节点为null，说明是链表头节点，插入头节点，next指向原头节点的next
}
break;
}
pred = e;
if ((e = e.next) == null) //如果循环链表没找到key，结束
break;
}
}
else if (f instanceof TreeBin) { //如果是树节点
validated = true;
TreeBin<K,V> t = (TreeBin<K,V>)f;
TreeNode<K,V> r, p;
if ((r = t.root) != null &&
(p = r.findTreeNode(hash, key, null)) != null) { //调用树节点的findTreeNode方法查找key值
V pv = p.val;
if (cv == null || cv == pv ||
(pv != null && cv.equals(pv))) {
oldVal = pv;
if (value != null)
p.val = value;
else if (t.removeTreeNode(p)) //调用树节点removeTreeNode方法删除节点，返回true说明节点太少，转化为链表结构
setTabAt(tab, i, untreeify(t.first));
}
}
}
}
}
if (validated) { //如果锁住头节点之后执行了删除操作（有可能加锁之前，其他线程进行了扩容操作，那么就不会执行删除节点操作，该值就为false）
if (oldVal != null) {
if (value == null)
addCount(-1L, -1); //count减1
return oldVal;
}
break;
}
}
}
return null;
}
- метод addCount, добавляющий и вычитающий количество узлов check
private final void addCount(long x, int check) {
CounterCell[] as; long b, s;
//如果counterCells不为null或者增加baseCount变量值失败
if ((as = counterCells) != null ||
!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
CounterCell a; long v; int m;
boolean uncontended = true; //表示无竞争
/**
* 如果as为null或者length为0
* 或者数组中随机一个元素为null，ThreadLocalRandom.getProbe()获取一个随机值
* 或者CAS加值失败
* 调用fullAddCount增加count值
*/
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);
return;
}
if (check <= 1)
return;
s = sumCount(); //计算节点数量
}
if (check >= 0) {
Node<K,V>[] tab, nt; int n, sc;
while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
int rs = resizeStamp(n);
if (sc < 0) { //如果小于0，正在扩容或进行初始化
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0) //如果已结束，推出循环
break;
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) //否则帮助扩容，sizeCtl+1表示多一个线程进行扩容
transfer(tab, nt);
}
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2)) //首次扩容，rs << RESIZE_STAMP_SHIFT) + 2)赋值给sizeCtl
transfer(tab, null);
s = sumCount();
}
}
}
- метод размера, вычислить количество узлов, вычислитьbaseCount и counterCellsСумма
public int size() {
long n = sumCount();
return ((n < 0L) ? 0 :
(n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
(int)n);
}
final long sumCount() {
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;
}
Суммировать
- ConcurrentHashMap в основном обеспечивает атомарность операций через метод CAS класса sun.misc.Unsafe.
- В jdk1.7 ConcurrentHashMap использует структуру сегмента для сегментации, а сегмент наследует ReentrantLock для реализации блокировки, что обеспечивает безопасность при многопоточности.
- В jdk1.8 ConcurrentHashMap больше не использует сегментную структуру, а использует ключевое слово synchronized для блокировки головного узла связанного списка в таблице, а степень детализации меньше, так что количество одновременно работающих потоков больше. и КПД выше.
- В jdk1.8, когда связанный список слишком длинный, он будет преобразован в древовидную структуру для повышения эффективности.Древовидную структуру см. в предыдущей статье.TreeMap красно-черное дерево, общая идея в основном та же.