Допустим, у меня есть AtomicReference
для списка объектов:
AtomicReference<List<?>> batch = new AtomicReference<List<Object>>(new ArrayList<Object>());
Поток A добавляет элементы в этот список: batch.get (). Add (o);
Позже поток B берет список и, например, сохраняет его в БД: insertBatch (batch.get ());
Нужно ли мне выполнять дополнительную синхронизацию при записи ( Thread A) и чтение (Thread B), чтобы убедиться, что поток B видит список так, как его оставил A, или об этом позаботится AtomicReference?
Другими словами: если у меня есть AtomicReference для изменяемого объекта, и один поток изменяет этот объект, другие потоки сразу видят это изменение?
Edit:
Может быть, какой-то пример кода в порядке:
public void process(Reader in) throws IOException {
List<Future<AtomicReference<List<Object>>>> tasks = new ArrayList<Future<AtomicReference<List<Object>>>>();
ExecutorService exec = Executors.newFixedThreadPool(4);
for (int i = 0; i < 4; ++i) {
tasks.add(exec.submit(new Callable<AtomicReference<List<Object>>>() {
@Override public AtomicReference<List<Object>> call() throws IOException {
final AtomicReference<List<Object>> batch = new AtomicReference<List<Object>>(new ArrayList<Object>(batchSize));
Processor.this.parser.parse(in, new Parser.Handler() {
@Override public void onNewObject(Object event) {
batch.get().add(event);
if (batch.get().size() >= batchSize) {
dao.insertBatch(batch.getAndSet(new ArrayList<Object>(batchSize)));
}
}
});
return batch;
}
}));
}
List<Object> remainingBatches = new ArrayList<Object>();
for (Future<AtomicReference<List<Object>>> task : tasks) {
try {
AtomicReference<List<Object>> remainingBatch = task.get();
remainingBatches.addAll(remainingBatch.get());
} catch (ExecutionException e) {
Throwable cause = e.getCause();
if (cause instanceof IOException) {
throw (IOException)cause;
}
throw (RuntimeException)cause;
}
}
// these haven't been flushed yet by the worker threads
if (!remainingBatches.isEmpty()) {
dao.insertBatch(remainingBatches);
}
}
Что происходит здесь, так это то, что я создаю четыре рабочих потока для синтаксического анализа некоторого текста (это Считыватель в параметре
метода process ()
). Каждый рабочий сохраняет строки, которые он проанализировал в пакете, и сбрасывает пакет, когда он заполнен ( dao.insertBatch (batch.getAndSet (new ArrayList
).
Поскольку количество строк в тексте не кратно размеру пакета, последние объекты попадают в пакет, который не сбрасывается, поскольку он не заполнен. Таким образом, эти оставшиеся партии вставляются основным потоком.
Я использую AtomicReference.getAndSet ()
, чтобы заменить полный пакет пустым. Верна ли эта программа в отношении многопоточности?