предисловие
Интервью в конце прошлого года.купить много еды, В качестве доказательства есть фотки, разбираю писания, надеюсь вы не думаете, что уже поздно (это проклятое промедление 😂).
У Pinduoduo было видеоинтервью в 8 часов вечера в воскресенье, но весь персонал работал сверхурочно, а в офисе было полно людей, так что каждый должен подумать об этом, прежде чем идти в Duoduo. 🤣
Вопрос 1. Является ли массив потокобезопасным, в какой строке это отражено?
Ответ: Поток не является безопасным, и конкретная производительность заключается в операции присваивания вновь добавленного элемента.elementData[size++]=e.
Давайте сначала посмотрим, что такое потокобезопасность:
потокобезопасностьПри доступе из нескольких потоковвыполняется на данныхМеханизм блокировки (оптимистическая блокировка или пессимистическая блокировка), только один поток может нормально обращаться к нему, а другие потоки не могут получить к нему доступ, пока поток не закончит чтение.Не будет несогласованности данных или загрязнения данных.
Поток небезопасенПри доступе из нескольких потоковЗащита доступа к данным не предусмотрена, и может быть несколько потоков, изменяющих данные последовательно, поэтомуПолученные данные являются грязными данными.
Как показано на рисунке, в AbstractList есть два класса: один — небезопасный для потоков ArrayList, а другой — безопасный для потоков вектор.
С точки зрения исходного кода, поскольку ключевое слово synchronized добавляется перед методом Vector, что означает синхронизацию, он является потокобезопасным.
Итак, мы видим разницу между этими двумя:
Vector: потокобезопасный, но с низкой производительностью.
ArrayList: не потокобезопасный, но высокопроизводительный и эффективный.
Поэтому издревле нельзя иметь и рыбу, и медвежью лапу.
Давайте посмотрим на конкретный код.На следующем рисунке мы можем нарисовать два шага при добавлении элементов:
1. Сохраните этот элемент в расположении Items[Size];
2. Увеличьте значение Размер.
Давайте подумаем об этом:
существуетоднопоточная операцияВ случае , если Size = 0, то после добавления элемента этот элемент находится в позиции 0, а Size=1, все нормально;
и если вМногопоточностьВ случае, например, двух потоков,Поток A сначала сохраняет элемент в позиции 0, и его размер не увеличивается на единицу., но в это время ЦП планирует приостановить поток A, а поток B получает возможность запуститься.
Поток B также добавляет элементы в этот ArrayList, поскольку в этот момент Size по-прежнему равен 0.(обратите внимание, что мы предполагаем, что добавление элемента занимает два шага, а поток A завершает только шаг 1),
Таким образом, поток B также сохраняет элемент в позиции 0, и его размер не увеличивается на 1. Затем оба потока A и B продолжают работать, и оба увеличивают значение Size, и в это время size равен 2.
Что ж, теперь мы нашли проблему, на самом деле есть только один элемент, который хранится в позиции 0, а Size равен 2. Это "небезопасно для потоков".
Посмотрите на следующий код: В случае многопоточного параллелизма выдается сообщение об ошибке, и программа сообщает об исключении параллельной модификации.ConcurrentModificationException, мы также можем посмотреть на метод add() в нижнем слое ArrayList, который является операцией без блокировки.Когда несколько потоков совместно используют ресурс, могут возникнуть проблемы с потоками;Метод add() массиваList не заблокирован.
public static void main(String[] args) throws InterruptedException {
List<Integer> list = new ArrayList<>();
ExecutorService threadPool = Executors.newFixedThreadPool(30);
for (int i = 1; i <= 30; i++) {
int finalI = i;
threadPool.execute(() -> {
list.add(finalI);
System.out.println(list);
});
}
threadPool.shutdown();
}
Если мы хотим не сообщать об ошибках,Вы можете преобразовать список в потокобезопасную коллекцию, используя метод synchronizedList класса инструментов Collections., чтобы преобразовать его в потокобезопасный.
public static void main(String[] args) throws InterruptedException {
List<Integer> list = Collections.synchronizedList(new ArrayList<>());
//此处只是案例demo,真实使用线程池不建议用这种方式创建
ExecutorService threadPool = Executors.newFixedThreadPool(30);
for (int i = 1; i <= 30; i++) {
int finalI = i;
threadPool.execute(() -> {
list.add(finalI);
System.out.println(list);
});
}
threadPool.shutdown();
}
Вопрос 2. Почему хэш-карта использует красно-черное дерево, а не дерево AVL или дерево B+?
Ответ: Поскольку дерево AVL поддерживает более строгий баланс, чем красно-черное дерево, это дерево ценой более медленной вставки и удаления, вызванной большим количеством операций вращения, и все узлы дерева B+ скучены вместе. данных мало, он выродится в связанный список.
АВЛ-дерево
Самобалансирующееся двоичное дерево поиска, также известное как дерево AVL, обладает следующими свойствами:Это пустое дерево или абсолютное значение разности высот между его левым и правым поддеревьями не превосходит 1, причем и левое, и правое поддеревья являются сбалансированным бинарным деревом.
Поиск в дереве AVL обычно выполняется быстрее, но этоБольшее количество операций вращения приводит к более медленной вставке и удалению за счетиз. Поэтому, если вы хотите, чтобы количество операций поиска преобладало над количеством обновлений дерева, используйте дерево AVL.
На следующих двух рисунках показаны сбалансированные бинарные деревья поиска.
Итак, давайте посмотрим, как выглядит несбалансированное бинарное дерево поиска? Как видно из рисунка ниже, левое поддерево узла C имеет F, L и M, то есть два слоя, а правое поддерево узла C — нет, они отличаются на 2, поэтому это не сбалансированный двоичный файл. дерево поиска.
Левое поддерево левого поддерева вставить узел (слева слева)
Правое поддерево правого поддерева вставляет узел (справа справа)
Правое поддерево левого поддерева вставляет узел (слева и справа)
Левое поддерево правого поддерева вставить узел (справа слева)
В+ дерево
Неконечные узлы дерева B+ не хранят данные, поэтому каждый узел может хранить больше ключевых слов. Следовательно, дерево B+ лучше справляется с большим объемом данных. HashMap в jdk1.7 изначально был в виде массива + связанный список.Связанный список необходимо заменить древовидной структурой с более высокой эффективностью поиска из-за его медленной функции поиска.
Если используется дерево B+, данные будут «втиснуты» в узел, когда объем данных не очень велик. В это время эффективность обхода вырождается в связанный список
B-дерево порядка m имеет следующие характеристики:
1. Корневой узел имеет не менее двух дочерних узлов.
2. Каждый промежуточный узел содержит не менееceil(m / 2)Есть не более m детей.
3. Каждый листовой узел содержит k-1 элементов, где m/2
4. Все листовые узлы расположены на одном уровне.
5. Элементы в каждом узле упорядочены от меньшего к большему, и k-1 элементов в узле точно соответствуют диапазону значений элементов, содержащихся в k дочерних элементах.
Вопрос 3. Почему ConcurrentHashMap является потокобезопасным? Какие баллы гарантированы?
отвечать:инициализация массивакогдавращениеЧтобы убедиться, что инициализация может быть успешной, а затем передатьCAS устанавливает значение переменной SIZECTL, чтобы гарантировать, что только один поток может инициализировать массив одновременно., после успешного выполнения CAS такжеПроверьте еще раз, был ли инициализирован текущий массив, если он был инициализирован, он не будет инициализирован снова;
Добавить слотпройти, когдавращениеУбедитесь, что новое добавление прошло успешно, а затем передайтеCAS добавить, если встречается значение слота, заблокировав текущий слот или корневой узел красно-черного дерева;
При расширениипройти черезЗаблокируйте слот исходного массива, установите узел передачи и вращайтедождитесь операций, чтобы обеспечить потокобезопасность.
Общая идея ConcurrentHashMap по методу put:
1. Если массив пуст, инициализировать его, после завершения инициализации перейти к пункту 2;
2. Вычислите, имеет ли текущий слот значение. Если значение отсутствует, создается cas. Если это не удается, он продолжает вращаться (для бесконечного цикла) до тех пор, пока не добьется успеха. Если слот имеет значение, перейдите к 3;
3. Если слот является узлом передачи (расширяющимся), он будет продолжать вращаться и ждать завершения расширения перед добавлением вместо перехода к 4 для узла передачи;
4. Если у слота есть значение, сначала заблокируйте текущий слот, чтобы другие потоки не могли работать. Если это связанный список, добавьте значение в конец связанного списка. Если это красно-черное дерево, используйте метод добавления красно-черного дерева для его добавления;
5. После завершения добавления проверьте, требуется ли расширение, и при необходимости расширьте.
Конкретный исходный код выглядит следующим образом:
final V putVal (K key, V value,boolean onlyIfAbsent){
if (key == null || value == null) throw new NullPointerException();
//通过hashcode计算 hash
int hash = spread(key.hashCode());
int binCount = 0;
for (Node<K, V>[] tab = table; ; ) {
Node<K, V> f;
int n, i, fh;
//table为空，进行初始化工作，调用initTable
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//如果当前索引位置没有值，直接创建
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//cas 在 i 位置创建新的元素，当 i 位置是空时，即能创建成功，结束 for 自循，否则继续自旋
if (casTabAt(tab, i, null,
new Node<K, V>(hash, key, value, null)))
break; // no lock when adding to empty bin
}
//如果当前槽点是转移节点，表示该槽点正在扩容，就会一直等待扩容完成
//转移节点的 hash 值是固定的，都是 MOVED
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
//槽点上有值的
else {
V oldVal = null;
//锁定当前槽点，其余线程不能操作，保证了安全
synchronized (f) {
//这里再次判断 i 索引位置的数据没有被修改
//binCount 被赋值的话，说明走到了修改表的过程里面
if (tabAt(tab, i) == f) {
//链表
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;
}
}
}
//红黑树，这里没有使用 TreeNode,使用的是 TreeBin，TreeNode 只是红黑树的一个节点
//TreeBin 持有红黑树的引用，并且会对其加锁，保证其操作的线程安全
else if (f instanceof TreeBin) {
Node<K, V> p;
binCount = 2;
//满足 if 的话，把老的值给 oldVal
//在 putTreeVal 方法里面，在给红黑树重新着色旋转的时候
//会锁住红黑树的根节点
if ((p = ((TreeBin<K, V>) f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
//binCount 不为空，并且 oldVal 有值的情况，说明已经新增成功了
if (binCount != 0) {
// 链表是否需要转化成红黑树
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
//这一步几乎走不到。槽点已经上锁，只有在红黑树或者链表新增失败的时候
//才会走到这里，这两者新增都是自旋的，几乎不会失败
break;
}
}
}
//check 容器是否需要扩容，如果需要去扩容，调用 transfer 方法去扩容
//如果已经在扩容中了，check 有无完成
addCount(1L, binCount);
return null;
}
Потокобезопасность при инициализации массива
Когда массив инициализирован, сначала прокрутите, чтобы убедиться, что инициализация может быть успешной, а затем установите значение переменной SIZECTL через CAS, чтобы гарантировать, что только один поток может инициализировать массив одновременно. еще раз определить, был ли текущий массив Инициализация завершена. Если инициализация завершена, он не будет инициализирован снова. Потокобезопасность инициализации массива гарантируется с помощью вращения + CAS + двойная проверка. Исходный код выглядит как следует:
//初始化 table，通过对 sizeCtl 的变量赋值来保证数组只能被初始化一次
private final Node<K, V>[] initTable () {
Node<K, V>[] tab;
int sc;
//通过自旋保证初始化成功
while ((tab = table) == null || tab.length == 0) {
// 小于 0 代表有线程正在初始化，释放当前 CPU 的调度权，重新发起锁的竞争
if ((sc = sizeCtl) < 0)
Thread.yield(); // lost initialization race; just spin
// CAS 赋值保证当前只有一个线程在初始化，-1 代表当前只有一个线程能初始化
// 保证了数组的初始化的安全性
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
// 很有可能执行到这里的时候，table 已经不为空了，这里是双重 check
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);
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}
}
Потокобезопасность при добавлении значений слотов
На данный момент, чтобы обеспечить потокобезопасность, были сделаны четыре оптимизации:
- Гарантированно можно добавить успех через бесконечный цикл спина.
- Когда текущий слот пуст, добавьте его через CAS.
- Текущий слот имеет значение, и текущий слот заблокирован.
- Когда красно-черное дерево вращается, заблокируйте корневой узел красно-черного дерева, чтобы гарантировать, что текущее красно-черное дерево может вращаться только одним потоком одновременно.
Потокобезопасность при масштабировании
-
При копировании слота слот исходного массива будет заблокирован;
-
После успешного копирования слот исходного массива будет установлен в качестве узла передачи, так что, если есть данные, которые нужно поместить в узел, будет обнаружено, что слот является узлом передачи, и он будет ждать, пока не будет Расширение выполнено успешно, прежде чем продолжить размещение.Вы можете обратиться к методу helpTransfer в методе размещения;
-
Копировать с конца в начало, при успешном копировании в качестве узла переноса устанавливается слот исходного массива.
-
После того, как копирование расширения завершено, значение нового массива напрямую присваивается контейнеру массива, и данные, ожидавшие размещения, могут быть продолжены.
// Разложение в основном делится на 2 этапа, во-первых, создать новый пустой массив, а второй, чтобы переместить и скопировать каждый элемент в новый массив // закладка: исходный массив, NextTab: новый массив частная конечная недействительной передачи (узел
[] вкладка, узел [] NextTab) { // длина старого массива Int N = tab.length, шаг; если ((шаг = (NCPU> 1) (п >>> 3) / NCPU:? п) [] нт = (Узел []) новый узел [п FWD = новый ForwardingNode (NextTab); логический прогресс = TRUE; булева отделка = ложь; // чтобы обеспечить зачистку перед совершением NextTab // Бесконечный спина, значение я стартую от максимального значения исходного массива и медленно уменьшаться до 0 для (INT I = 0, связанный = 0;;) { Узел F; INT СПЧ; в то время как (заранее) { INT NEXTINDEX, nextBound; // флаг для завершения цикла если (--i> = связан || отделка) авансовый = FALSE; // уже скопировано иначе, если ((NEXTINDEX = transferIndex) шаг? NEXTINDEX - шаг: 0))) { оценка = nextBound; I = NEXTINDEX - 1; авансовый = FALSE; } } // если условие выполняется, то копия закончена если (я = п || я + п> = nextn) { INT СБН; // После копирования, присвоить значение непосредственно, потому что каждый раз, когда узел копируется узел переноса размещаются на исходном массиве, так что данные скопированный узел не изменятся снова. // Исходный массив оказывается перенос узла и не будет работать. Он будет ждать узла передачи исчезнуть, прежде чем продолжить. // То есть, когда узел массива помечается как узел передачи, не будет никаких дальнейших изменений, так что не будет никаких проблем безопасности потока // Таким образом, нет никаких проблем с назначением значения непосредственно здесь. если (чистовая) { nextTable = NULL; Таблица = NextTab; sizeCtl = (п >> 1); возвращение; } если (U.compareAndSwapInt (это, SIZECTL, SC = sizeCtl, подкожно - 1)) { если ((п - 2) = resizeStamp (п) пер, Hn; если (СПЧ> = 0) { INT runBit = ФХ & п; Узел LASTRUN = F; для (Node р = f.next;! р = NULL; р = p.next) { INT б = p.hash & п; если (б! = runBit) { runBit = Ь; LASTRUN = р; } } если (runBit == 0) { п = LASTRUN; кп = NULL; } еще { Hn = LASTRUN; п = NULL; } // Если узел имеет только один данные, копировать его непосредственно, если это связанный список, цикл несколько раз, чтобы сформировать связанный список копию для (Node р = F;! р = LASTRUN; р = p.next) { ИНТ фот = p.hash; К рк = p.key; V PV = p.val; если ((рН & п) == 0) п = новый узел (тела, рк, р, п); еще Hn = новый узел (тела, рк, р, Hn); } // поместить скопированное значение в новой позиции массива setTabAt (NextTab, я, п); setTabAt (NextTab, я + п, Нп); // Поместите узел ForwardingNode на старой позиции массива // Если положить, то оказывается, что он является узлом ForwardingNode, и данные этого узла не будут перемещены больше setTabAt (закладка, я, ВПЕРЕД); авансовый = TRUE; } // копия красно-черного дерева иначе если (е InstanceOf TreeBin) { // Экземпляр работа красно-черного дерева, то же самое, как и содержание HashMap, код игнорирует // Поместите узел ForwardingNode на старой позиции массива setTabAt (закладка, я, ВПЕРЕД); авансовый = TRUE; } } } } } }
Вопрос 4. Говоря о распределенных блокировках, давайте поговорим о дизайнерских идеях и решениях.
Ответ: Описание в основном основано на конкретных бизнес-сценариях (каждый проект здесь индивидуален, поэтому я не буду его расширять), в основном представляющих распределенные блокировки, реализованные с помощью Redis, которые должны быть гарантированы.Взаимное исключение (в любой момент только один клиент удерживает блокировку, используйте setnx),Невозможно заблокировать (установить срок действия),Убедитесь, что блокировка и разблокировка относятся к одному и тому же клиенту (установите разные значения), время работы слишком велико, что приводит к истечению срока действия блокировки (установите сторожевой таймер, автоматически продлите блокировку) и повторный вход в блокировку (используйте redis hset ).
Если в распределенной системе мы читаем часть данных из базы данных, а затем модифицируем и сохраняем ее, легко столкнуться с проблемами параллелизма. Поскольку чтение и сохранение обновления не являются атомарными операциями, при одновременном выполнении они приведут к неверным данным. Если это автономное приложение, его можно избежать, напрямую используя локальную блокировку. Если это распределенное приложение, приложение развертывается на нескольких JVM, и нет возможности его контролировать, поэтому для его решения вводятся распределенные блокировки.
взаимная исключительность
127.0.0.1:6379> setnx lock value1 #在键lock不存在的情况下，将键key的值设置为value1
(integer) 1
127.0.0.1:6379> setnx lock value2 #试图覆盖lock的值，返回0表示失败
(integer) 0
127.0.0.1:6379> get lock #获取lock的值，验证没有被覆盖
"value1"
127.0.0.1:6379> del lock #删除lock的值，删除成功
(integer) 1
127.0.0.1:6379> setnx lock value2 #再使用setnx命令设置，返回0表示成功
(integer) 1
127.0.0.1:6379> get lock #获取lock的值，验证设置成功
Блокировка: Используйте команду значения ключа setnx, если ключ не существует, установите значение (блокировка прошла успешно). Если блокировка уже существует (т. е. клиент удерживает блокировку), настройка не выполняется (сбой блокировки).
Разблокировать: используйте команду del, чтобы снять блокировку, удалив значение ключа.
Не могу зайти в тупик
Установите время истечения и удалите данные в точке, чтобы избежать того, что если клиент внезапно выйдет из строя при удерживании блокировки, это приведет к тому, что он не сможет разблокировать, тогда другие люди не смогут получить блокировку, блокировка будет существовать всегда и в конечном итоге приведет к взаимоблокировке.
срок действия блокировки истек
Как долго действует время?Если моя бизнес-операция превышает допустимое время, мой бизнес-код будет автоматически разблокирован для меня до того, как он будет выполнен, и он будет завершен.
Эта проблема немного сложна, и в Интернете есть много дискуссий.Первое решение — положиться на то, что программист поймет ее, оценит время, необходимое для выполнения бизнес-кода, а затем установит более длительный срок действия. чем время выполнения.Автоматическая разблокировка влияет на выполнение бизнес-кода на стороне клиента. Однако это не надежное решение, например, дрожание сети непредсказуемо, а также может привести к увеличению времени выполнения бизнес-кода, поэтому оно небезопасно. Есть более надежный метод,
этоОбновите замок. Идея реализации распределенных блокировок в фреймворке Redisson заключается в использовании механизма watchDog для обновления блокировки.. Когда блокировка будет успешно добавлена, поток демона будет запущен в то же время.Период действия по умолчанию составляет 30 секунд, и блокировка будет обновляться каждые 10 секунд до 30 секунд.Пока клиент, удерживающий блокировку, не отключен , можно гарантированно держать блокировку все время.Пока бизнес-код не будет выполнен, клиент разблокирует его сам.Если он не работает, он будет автоматически разблокирован после истечения срока действия.
Убедитесь, что и блокировка, и разблокировка выполняются одним и тем же клиентом.
Значение ключа может быть установлено в зависимости от бизнеса.Например, он используется центром пользователей.Его можно указать как USER_REDIS_LOCK.Уникальность значения может быть гарантирована с помощью uuid, который используется для идентификации заблокированного клиента и обеспечения того, что блокировка и разблокировка относятся к одному и тому же клиенту.
Каждый раз при разблокировке можно предварительно определить является ли значение блокировки текущим пользователем.Если да, то значит его можно разблокировать.Если нет, то разблокировать нельзя, что приведет к непониманию чужих блокировок.
Блокировка повторного входа
Реентерабельная блокировка означает, что после того, как внешний слой использует блокировку, внутренний слой все еще может быть использован, так в чем заключается идея реализации реентерабельной блокировки? Идея реализации повторных блокировок в Redisson использует хеш-таблицу Redis для хранения количества повторных входов, при успешной блокировке используется команда hset, и значение (количество повторных входов) равно 1.
При разблокировке сначала определите, больше ли повторяющееся число 0, и если оно больше 0, вычтите единицу, в противном случае удалите значение ключа и освободите ресурс блокировки.
Вопрос 5. Вопросы по алгоритму
Когда я увидел этот вопрос по алгоритму, я рассмеялся, разве это не первый вопрос Ликоу, хахаха, к счастью, я его прошел. Каждый может написать простой метод, насильственную операцию, но производительность имеет значение, но область комментариев написана очень умно для великого бога, поэтому я переместил ее напрямую.
тема:
Учитывая целочисленный массив nums и целочисленное целевое значение target, найдите два целых числа в массиве, где и являются целевым значением, и верните их индексы массива.
Можно предположить, что для каждого входа будет только один ответ. Однако один и тот же элемент массива не может повторяться в ответе.
Вы можете возвращать ответы в любом порядке.
решение:
Это решение не вычисляет ответ из вопроса, а начинает с ответа и видит, какие цифры нужны.
public int[] twoSum(int[] nums, int target) {
int[] indexs = new int[2];
// 建立k-v ，一一对应的哈希表
HashMap<Integer,Integer> hash = new HashMap<Integer,Integer>();
for(int i = 0; i < nums.length; i++){
if(hash.containsKey(nums[i])){
indexs[0] = i;
indexs[1] = hash.get(nums[i]);
return indexs;
}
// 将数据存入 key为补数 ，value为下标
hash.put(target-nums[i],i);
}
// // 双重循环 循环极限为(n^2-n)/2
// for(int i = 0; i < nums.length; i++){
// for(int j = nums.length - 1; j > i; j --){
// if(nums[i]+nums[j] == target){
// indexs[0] = i;
// indexs[1] = j;
// return indexs;
// }
// }
// }
return indexs;
}