AtomicReference на изменяемый объект и видимость

Допустим, у меня есть 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 (batchSize))); ).

Поскольку количество строк в тексте не кратно размеру пакета, последние объекты попадают в пакет, который не сбрасывается, поскольку он не заполнен. Таким образом, эти оставшиеся партии вставляются основным потоком.

Я использую AtomicReference.getAndSet () , чтобы заменить полный пакет пустым. Верна ли эта программа в отношении многопоточности?

8
задан Jan Van den bosch 21 February 2012 в 13:59
поделиться