Параллелизм java: стратегия CopyOnWriteArrayList
Я пытаюсь понять CopyOnWriteArrayList в своем коде:
Мой код:
public class AuditService {
private CopyOnWriteArrayList<Audit> copyWrite;
public void flush(Audit... audits) {
Collection<Audit> auditCollection = Arrays.asList(audits);
this.copyWrite.addAll(auditCollection);
this.copyWrite.forEach(audit -> {
try {
// save audit object on database
this.copyWrite.remove(audit);
} catch (DataAccessException e) {
// log it
}
});
}
}
Что делает этот код:
- Сначала результаты аудита сохраняются в буфере
CopyOnWriteArrayList. - Пытается сохранить аудит в базу данных
- Когда он сохранен, он удаляется из буфера
CopyOnWriteArrayList.
Другие:
AuditService- это одноэлементный классflushможет быть достигнут несколькими потоками.
Вопросы:
- Я предполагаю, что
this.copyWrite.forEach(audit -> {...может быть доступен одновременно несколькими потоками: Означает ли это, что один и тот же объект аудита можно дважды попытаться сохранить в базе данных? - Каждый раз, когда операция модификации выполняется на
CopyOnWriteArrayList, новая копия заполняется в других потоках? Как он заселен?
Ответы 1
Каждый раз, когда вызывается remove, создается новая копия внутреннего массива, поддерживающего CopyOnWriteArrayList. Будущие обращения к этому списку с помощью методов доступа и мутатора будут видны при обновлении.
тем не мение, метод метода CopyOnWriteArrayList#foreach выполняет итерацию по массиву, который был доступен во время его вызова. Это означает, что все исполнения метода flush, который вошел в foreachдо, любое обновление в списке будет повторяться для устаревшей версии массива.
Таким образом, во время параллельного выполнения метода flush одни и те же элементы Audit будут сохраняться более одного раза и до d раз, если d - это максимальное количество одновременных выполнений метода flush.
Другая проблема с использованием CopyOnWriteArrayList в этом случае заключается в том, что поскольку новая копия создается для каждого вызова remove, сложность вашего кода равна d.n^2, где n - длина списка, а d - определенная выше.
CopyOnWriteArrayList - неправильная реализация для использования здесь. Возможны несколько подходящих дизайнов. Один из них - использовать LinkedBlockingQueue следующим образом [*]:
public void flush(Audit... audits) {
Collection<Audit> auditCollection = Arrays.asList(audits);
this.queue.addAll(auditCollection);
Collection<Audit> poissonedAudits = new ArrayList<Audit>();
Audit audit = null;
while ((audit = this.queue.poll()) != null) {
try {
// save audit object on database
queue.remove(audit);
} catch (DataAccessException e) {
// log it
poissonedAudits.add(audit);
}
}
this.queue.addAll(poissonedAudits);
}
Вызов LinkedBlockingQueue#poll() является потокобезопасным и атомарным. Один и тот же элемент нельзя опрашивать несколько раз (если он не добавляется в очередь более одного раза, cf [*]). Сложность в n линейна.
Два момента, которые следует учитывать:
- Ваша коллекция аудита не должна содержать элементов
null, поскольку это запрещено дляLinkedBlockingQueue, посколькуnullиспользуется для обозначения того, что список пуст. - Вам следует позаботиться о том, чтобы не использовать методы блокировки, такие как
takeилиpoll(timeout, unit), для опроса очереди.
[*] В методе flush было внесено изменение: новое не выполняет копирование элементов Audit. Я не уверен, есть ли гарантия, что эти элементы различны, если метод flush вызывается параллельно. Если массив Audit один и тот же для всех вызовов flush, очередь должна заполняться только один раз перед любым вызовом flush.
Я только что понял, что вы удалили параметр Audits... audits. Я также читал [*] и не совсем понимаю, что именно вы имеете в виду. Я отредактировал ваш пост с моими предложениями. Пожалуйста, не стесняйтесь высказывать свои мысли.
Я отредактировал ваш ответ. Пожалуйста, не могли бы вы взглянуть на это?