Java - 并发清除列表

Question

我正在尝试找到一种实现以下 API 的好方法:

void add(Object o);
void processAndClear();

该类将存储对象，并且在调用 processAndClear 时将遍历当前存储的对象，以某种方式处理它们，然后清除存储。这个类应该是线程安全的。

明显的方法是使用锁定，但我想更“并发”。这是我将使用的方法:

class Store{
    private AtomicReference<CopyOnWriteArrayList<Object>> store = new AtomicReference<>(new CopyOnWriteArrayList <>());

    void add(Object o){
        store.get().add(o);
    }

    void processAndClear(){
        CopyOnWriteArrayList<Object> objects = store.get();
        store.compareAndSet(objects, new CopyOnWriteArrayList<>());
        for (Object object : objects) {
            //do sth
        }
    }
}

这将允许尝试添加对象的线程几乎立即进行，而无需任何锁定/等待 xlearing 完成。这是或多或少正确的方法吗？

Answer 1

您上面的代码不是线程安全的。想象一下:

1. 线程 A 在 store.get() 之后立即在 add() 处暂停
2. 线程 B 在 processAndClear() 中，替换列表，处理旧列表的所有元素，然后返回。
3. 线程 A 恢复并向现在已过时的列表添加一个永远不会被处理的新项目。

这里可能最简单的解决方案是使用 LinkedBlockingQueue ，这也会大大简化任务:

class Store{
  final LinkedBlockingQueue<Object> queue = new LinkedBlockingQueue<>();

  void add(final Object o){
    queue.put(o); // blocks until there is free space in the optionally bounded queue
  }

  void processAndClear(){
    Object element;
    while ((element = queue.poll()) != null) { // does not block on empty list but returns null instead
      doSomething(element);
    }
  }
}

编辑:如何使用synchronized来做到这一点:

class Store{
  final LinkedList<Object> queue = new LinkedList<>(); // has to be final for synchronized to work

  void add(final Object o){
    synchronized(queue) { // on the queue as this is the shared object in question
      queue.add(o);
    }
  }

  void processAndClear() {
    final LinkedList<Object> elements = new LinkedList<>(); // temporary local list
    synchronized(queue) { // here as well, as every access needs to be properly synchronized
      elements.addAll(queue);
      queue.clear();
    }

    for (Object e : elements) { 
      doSomething(e); // this is thread-safe as only this thread can access these now local elements
    }
  }
}

为什么这不是一个好主意

虽然这是线程安全的，但与并发版本相比它要慢得多。假设您的系统有 100 个线程，这些线程经常调用 add，而一个线程调用 processAndClear。那么就会出现以下性能瓶颈:

• 如果一个线程调用 add，则其他 99 个线程同时被搁置。
• 在 processAndClear 的第一部分，所有 100 个线程都被搁置。

如果您假设这 100 个添加线程没有其他事情可做，您可以很容易地证明该应用程序的运行速度与单线程应用程序的运行速度相同，减去同步成本。这意味着:添加 100 个线程实际上比添加 1 个线程要慢。如果您像第一个示例那样使用并发列表，情况就不同了。

但是，处理线程的性能会有所提高，因为 doSomething 可以在添加新元素时在旧元素上运行。但是并发示例可能会更快，因为您可以让多个线程同时进行处理。

Effectively synchronized 也可以使用，但你会自动引入性能瓶颈，可能导致应用程序作为单线程运行得更慢，迫使你进行复杂的性能测试。此外，扩展功能总是存在引入线程问题的风险，因为锁定需要手动完成。
相比之下，并发列表无需额外代码即可解决所有这些问题，并且以后可以轻松更改或扩展代码。