предисловие
всем привет. HashMap был представлен в предыдущем блоге, и эта статья представляет вам ConcurrentHashMap.
[В одной статье рассказывается о прошлой и настоящей жизни HashMap]:Наггетс.Талант/пост/695958…
Как упоминалось выше, HashMap — это потоконебезопасный класс, а ConcurrentHashMap рекомендуется для операций с данными типа k-v в условиях многопоточности. Эта статья продолжит интерпретацию HashMap, начиная с ключевых переменных-членов, основных методов и общих точек интервью ConcurrentHashMap, чтобы помочь вам понять ConcurrentHashMap, основную структуру данных в java, простыми словами.
1. Введение в ConcurrentHashMap
ConcurrentHashMap, как и HashMap, также является структурой данных, которая работает с данными типа K-V. В HashMap modCount используется для предотвращения конфликтов при многопоточности, но решается с помощью механизма fast-fail. Невозможно решить проблему упорядоченного выполнения действий присваивания одному и тому же значению ключа в случае многопоточности. ConcurrentHashMap использует блокировки сегментов для управления одновременными операциями записи в версии 1.7 и использует ключевое слово CAS+synchronized для управления параллельными операциями записи в JDK 1.8. В этой статье будут представлены основные концепции ConcurrentHashMap для JDK1.8.
2. Введение в ключевые понятия
Ключевые концепции структуры данных ConcurrentHashMap согласуются с HashMap.
- множество
- Линейный связанный список
- бинарное дерево
- хеш-таблица
- хэш-коллизия
[Попросите понять прошлое и настоящее жизни HashMap:Наггетс.Талант/пост/695958…
3. Ключевые переменные-члены
//hashamp的最大容量，2的30次方
private static final int MAXIMUM_CAPACITY = 1 << 30;
//构造hashmap时，默认初始化容量为16
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;
//默认扩容的扩展因子，当hashmap中的元素个数达到当前容量的75%时，触发扩容
private static final float LOAD_FACTOR = 0.75f;
//ConcurrentHashMap数组节点上链表转换为红黑的的阈值，链表节点达到8个时转换为红黑树【注意：这里不是绝对，切往后看】
static final int TREEIFY_THRESHOLD = 8;
//链表节点数小于6个时，从红黑树转换为链表
static final int UNTREEIFY_THRESHOLD = 6;
//链表转化为红黑的第二个要求，与TREEIFY_THRESHOLD对应，最小的链转树的数组大小
static final int MIN_TREEIFY_CAPACITY = 64;
//2^15-1，help resize的最大线程数
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;
//32-16=16，sizeCtl中记录size大小的偏移量
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
//记录目标的hash值
static final int MOVED = -1;
//目标的红黑树根节点hash值
static final int TREEBIN = -2;
//ReservationNode的hash值
static final int RESERVED = -3;
//获取当前CPU核数
static final int NCPU = Runtime.getRuntime().availableProcessors();
//存放node的数组
transient volatile Node<K,V>[] table;
/*控制标识符，用来控制table的初始化和扩容的操作，不同的值有不同的含义
*当为负数时：-1代表正在初始化，-N代表有N-1个线程正在 进行扩容
*当为0时：代表当时的table还没有被初始化
*当为正数时：表示初始化或者下一次进行扩容的大小
*/
private transient volatile int sizeCtl;
//默认为null，初始化发生在第一次插入操作，默认大小为16的数组，用来存储Node节点数据，扩容时大小总是2的幂次方
transient volatile Node<K,V>[] table;
//默认为null，扩容时新生成的数组，其大小为原数组的两倍
private transient volatile Node<K,V>[] nextTable;
4. Анализ ключевых методов
4.1. Конструктор
ConcurrentHashMap имеет пустой и параметризованный конструктор.
4.1.1 Конструктор пустых параметров
/**
* Creates a new, empty map with the default initial table size (16).
*/
public ConcurrentHashMap() {
}
Как видно из комментариев, метод построения нулевого параметра создает пустую карту с размером по умолчанию 16. При построении с нулевыми параметрами карта будет инициализирована, когда мы поместим первый элемент. Давайте посмотрим на метод putVal в методе put.
Анализ метода initTable()
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
//如果一个线程发现sizeCtl<0，意味着另外的线程执行CAS操作成功，当前线程只需要让出cpu时间片
if ((sc = sizeCtl) < 0)
Thread.yield();
//如果cas成功，修改sizeCtl变量未-1，表示正在初始化
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
//table为空，设置一个默认大小的数组
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;
//设置一个默认扩容大小为12，根据扩容因子0.75得到。因为这里是空参构造方法，所有参数
为默认
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}
Из приведенного выше значения сам по себе пустой конструктор не имеет смысла, только инициализируются некоторые переменные со значениями по умолчанию, а для реальной инициализации данных структуры нужно дождаться ввода первого элемента put.
4.1.2 Конструкторы с параметрами
public ConcurrentHashMap(int initialCapacity) {
if (initialCapacity < 0)
throw new IllegalArgumentException();
int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
MAXIMUM_CAPACITY :
//将当前容量设置为1.5倍+1
tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
this.sizeCtl = cap;
}
Установка емкости гарантированно кратна 2. Здесь есть особенно важный момент.В примечаниях выше было указано, что при инициализации входящая емкость будет установлена в 1,5 раза больше входящей емкости + 1. Причина этого в том, что если вам нужно 7 элементов емкости карты , вы переходите в 7, по здравому смыслу вам будет установлена емкость 8, но вы уже знаете, что емкость равна 7, и когда количество элементов достигнет шестого, произойдет расширение. Чтобы избежать необоснованной установки начальной мощности и трудоемкой операции последующего расширения мощности, начальная мощность устанавливается в соответствии с коэффициентом расширения мощности 0,75f.
Если вы введете 7, вместимость составит 16.
Если вы введете 15, вместимость составит 32
4.2. метод ввода
public V put(K key, V value) {
return putVal(key, value, false);
}
//onlyIfAbsent参数为，当key值不存在的时候在进行设置
final V putVal(K key, V value, boolean onlyIfAbsent) {
//这里与HashMap不同，key与value都不允许为空
if (key == null || value == null) throw new NullPointerException();
//计算key的hash
int hash = spread(key.hashCode());
int binCount = 0;
//数组遍历
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
//如果为空参构造方法，table参数还未被初始化，则先进行初始化动作
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//如果根据hash定位到的数组节点为空则使用cas进行新节点的设置
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break; // no lock when adding to empty bin
}
//如果定位到的hash值为-1，则说明正在扩容，当前线程去帮忙进行扩容操作
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
else {
//最后一种情况，说明不是直接在数组上的节点，则遍历链表或者红黑树，遍历时使用synchronized加锁
V oldVal = null;
synchronized (f) {
//在节点 f 上进行同步，节点插入之前，再次利用tabAt(tab, i) == f判断，防止被其它线程修改
if (tabAt(tab, i) == f) {
//根据在第二点的值，hash值大于0时为链表节点，则进行链表遍历
if (fh >= 0) {
//链表节点计数
binCount = 1;
//链表遍历
for (Node<K,V> e = f;; ++binCount) {
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;
}
}
}
//这里判断节点是不是等于TreeBin，如果是，则进行红黑树遍历
else if (f instanceof TreeBin) {
Node<K,V> p;
binCount = 2;
//红黑树设置值
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
//计数值如果大于8
if (binCount != 0) {
//计数大于8则链表转换红黑树
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
//如果key值节点已经存在则进行返回
if (oldVal != null)
return oldVal;
break;
}
}
}
//容量增加，决定是否扩容
addCount(1L, binCount);
return null;
}
Метод put является ядром всего ConcurrentHashMap, а ключевое слово cas+synchronized используется для обеспечения потокобезопасности каждого узла. Я не знаю, есть ли у моих друзей какие-либо вопросы о том, почему они используют блокировку cas+synchronized, заменяющую метод Segment+ReentrantLock в версии JDK1.7. Основная причина здесь в том, что блокировка уточнена, Segment+ReentrantLock — это сегментированная блокировка, конфликт блокировок относительно высок, а cas+synchronized — для узла массива или всей цепочки в массиве, что значительно сокращает количество блокировок. конкурентный конфликт. После JDK1.6 были внесены ключевые улучшения производительности для ключевого слова synchronized, а также был выполнен ряд оптимизаций, таких как огрубление блокировки, обновление и устранение.
Поэтому с такой мелкозернистой блокировкой, как ConcurrentHashMap, производительность будет не хуже, чем у ReentrantLock.
Синхронизированный процесс обновления замка:Краткое описание.com/Fear/704 Evil Retribution 56 Ах ах...
4.3.получить метод
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
//hash值计算
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;
}
//小于0，则进行红黑树遍历，其中find方法点击进入，根据注释的值，将会调用子类TreeBin内部find方法，红黑树遍历查找
else if (eh < 0)
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;
}
Логика метода get относительно проста, и во всем процессе нет блокирующих действий. Тут у вас должны возникнуть сомнения: в процессе расширения узел был перехэширован, можно ли его пройти корректно? Здесь мы должны посмотреть на Node
static class Node<K,V> implements Map.Entry<K,V> {
final int hash;
final K key;
volatile V val;
volatile Node<K,V> next;
}
Данные узла узла модифицируются с помощью ключевого слова volatile, которое может быть загружено из основной памяти в последние данные адресной информации, поэтому нет необходимости его блокировать.
4.4.метод удаления
final V replaceNode(Object key, V value, Object cv) {
int hash = spread(key.hashCode());
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)
tab = helpTransfer(tab, f);
else {
V oldVal = null;
boolean validated = false;
synchronized (f) {
if (tabAt(tab, i) == f) {
if (fh >= 0) {
validated = true;
for (Node<K,V> e = f, pred = null;;) {
K ek;
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
V ev = e.val;
if (cv == null || cv == ev ||
(ev != null && cv.equals(ev))) {
oldVal = ev;
if (value != null)
e.val = value;
else if (pred != null)
pred.next = e.next;
else
setTabAt(tab, i, e.next);
}
break;
}
pred = e;
if ((e = e.next) == null)
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) {
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))
setTabAt(tab, i, untreeify(t.first));
}
}
}
}
}
if (validated) {
if (oldVal != null) {
if (value == null)
addCount(-1L, -1);
return oldVal;
}
break;
}
}
}
return null;
}
Метод remove аналогичен методу put в целом, это операция записи, если он находится непосредственно в головном узле массива, то он будет удален напрямую, иначе в обработку будет добавлено ключевое слово synchronized.
4.5 Расширение
Расширение происходит, когда емкости таблицы недостаточно, то есть количество элементов в таблице достигает порога емкости sizeCtl, и таблицу необходимо расширить.
Анализ триггерного входа:
1. метод put, количество вставляемых элементов достигает порога расширения
2. Инициировано tryPresize. Есть два возможных триггера для этого действия,
Первый: если количество узлов связанного списка одного блока в массиве достигает 8, но длина массива все еще меньше 64, он попытается расшириться.
Второй: это вызов метода, когда putAll.
хорошо, зная вход, мы нажимаем на послойный метод, чтобы найти метод расширения: перенос
private final void transfer(ConcurrentHashMap.Node<K,V>[] tab, ConcurrentHashMap.Node<K,V>[] nextTab) {
int n = tab.length, stride;
// 多线程扩容，每核处理的量小于16，则强制赋值16
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE;
// nextTab 为空，先实例化一个新的数组
if (nextTab == null) {
try {
@SuppressWarnings("unchecked")
// 新数组的大小是原来的两倍
ConcurrentHashMap.Node<K,V>[] nt = (ConcurrentHashMap.Node<K,V>[])new ConcurrentHashMap.Node<?,?>[n << 1];
nextTab = nt;
} catch (Throwable ex) {
sizeCtl = Integer.MAX_VALUE;
return;
}
// 更新成员变量
nextTable = nextTab;
// 更新转移下标，就是 老的 tab 的 length
transferIndex = n;
}
// bound ：该线程此次可以处理的区间的最小下标，超过这个下标，就需要重新领取区间或者结束扩容
// advance： 该参数
int nextn = nextTab.length;
// 创建一个 fwd 节点，用于占位。当别的线程发现这个槽位中是 fwd 类型的节点，则跳过这个节点。
ConcurrentHashMap.ForwardingNode<K,V> fwd = new ConcurrentHashMap.ForwardingNode<K,V>(nextTab);
// advance 变量指的是是否继续递减转移下一个桶，如果为 true，表示可以继续向后推进，反之，说明还没有处理好当前桶，不能推进
boolean advance = true;
// 完成状态，如果是 true，表示扩容结束
boolean finishing = false; // to ensure sweep before committing nextTab
// 死循环,i 表示下标，bound 表示当前线程可以处理的当前桶区间最小下标
for (int i = 0, bound = 0;;) {
ConcurrentHashMap.Node<K,V> f; int fh;
while (advance) {
int nextIndex, nextBound;
if (--i >= bound || finishing)
advance = false;
else if ((nextIndex = transferIndex) <= 0) {
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 = null;
table = nextTab;
sizeCtl = (n << 1) - (n >>> 1);
return;
}
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 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; // already processed
else {
synchronized (f) {
// 这儿多判断一次，是否为了防止可能出现的remove()操作
if (tabAt(tab, i) == f) {
// 旧链表上该节点的数据，会被分成低位和高位，低位就是在新链表上的位置跟旧链表上一样，
// 高位就是在新链表的位置是旧链表位置加上旧链表的长度
ConcurrentHashMap.Node<K,V> ln, hn;
if (fh >= 0) {
int runBit = fh & n;
ConcurrentHashMap.Node<K,V> lastRun = f;
for (ConcurrentHashMap.Node<K,V> p = f.next; p != null; p = p.next) {
int b = p.hash & n;
if (b != runBit) {
runBit = b;
lastRun = p;
}
}
if (runBit == 0) {
ln = lastRun;
hn = null;
}
else {
hn = lastRun;
ln = null;
}
for (ConcurrentHashMap.Node<K,V> p = f; p != lastRun; p = p.next) {
int ph = p.hash; K pk = p.key; V pv = p.val;
// 该节点哈希值与旧链表长度与运算，结果为0，则在低位节点上，反之，在高位节点上
if ((ph & n) == 0)
ln = new ConcurrentHashMap.Node<K,V>(ph, pk, pv, ln);
else
hn = new ConcurrentHashMap.Node<K,V>(ph, pk, pv, hn);
}
setTabAt(nextTab, i, ln);
// 在nextTable i + n 位置处插上链表
setTabAt(nextTab, i + n, hn);
// 在table i 位置处插上ForwardingNode 表示该节点已经处理过了
setTabAt(tab, i, fwd);
advance = true;
}
else if (f instanceof ConcurrentHashMap.TreeBin) {
// 如果是TreeBin，则按照红黑树进行处理，处理逻辑与上面一致
// 红黑树的逻辑跟节点一模一样，最后也会分高位和低位
ConcurrentHashMap.TreeBin<K,V> t = (ConcurrentHashMap.TreeBin<K,V>)f;
ConcurrentHashMap.TreeNode<K,V> lo = null, loTail = null;
ConcurrentHashMap.TreeNode<K,V> hi = null, hiTail = null;
int lc = 0, hc = 0;
for (ConcurrentHashMap.Node<K,V> e = t.first; e != null; e = e.next) {
int h = e.hash;
ConcurrentHashMap.TreeNode<K,V> p = new ConcurrentHashMap.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，那么转成链表，反之，创建一个新的树
ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
(hc != 0) ? new ConcurrentHashMap.TreeBin<K,V>(lo) : t;
hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
(lc != 0) ? new ConcurrentHashMap.TreeBin<K,V>(hi) : t;
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd);
advance = true;
}
}
}
}
}
}
5. Резюме
В этой статье подробно представлена основная логика и реализация структуры данных ConcurrentHashMap. Теперь сделайте следующее резюме
1. Начальная емкость ConcurrentHashMap равна степени 2, емкость пустой конструкции параметров по умолчанию равна 16, а емкость параметризованной конструкции больше чем в 1,5 раза + 1 параметр в минимальной степени 2.
2. В методе put массив cas spin используется для установки значения, связанный список или ключевое слово synchronized красно-черного дерева используется для установки значения блокировки, а текущий статус узла оценивается в соответствии с хешем. значение в массиве суждений. -1 означает, что происходит расширение, а больше 0 означает значение хеш-функции. Значение напрямую перезаписывается. Если тип узла TreeBin, это красно-черное дерево. В противном случае выполняется обход связанного списка и присвоено.
3. Требования преобразования красно-черного дерева, количество узлов связанного списка больше или равно 8, а длина массива больше или равна 64.
4. Метод get не заблокирован, поддерживает параллелизм, обеспечивает видимость памяти с помощью ключевого слова volatile и получает последние данные узла.
6. Ссылка
woo woo Краткое описание.com/fear/eat 5 oh 024's 9…
сегмент fault.com/ah/119000002…
7. Свяжитесь со мной
Диндин: louyanfeng25
WeChat: baiyan_lou