Java - 并发清除列表答案

【问题标题】：Java - concurrent clear of the listJava - 并发清除列表
【发布时间】：2014-03-05 15:17:38
【问题描述】：

我正在努力寻找实现以下 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 完成。这是或多或少正确的方法吗？

【问题讨论】：

标签： java multithreading locking copyonwritearraylist

【解决方案1】：

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

线程 A 在 add() 紧跟在 store.get() 之后被搁置
线程 B 在processAndClear()，替换列表，处理旧列表的所有元素，然后返回。
线程 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 可以在添加新元素时在旧元素上运行。但是，并发示例可能会更快，因为您可以让多个线程同时进行处理。

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

【讨论】：

好地方-如果我想确保在给定时间只有一个线程清理列表，我可以加锁或同步方法对吗？添加到列表中的线程很好。
如果两种方法是同步的/使用相同的循环，那么是的，那会起作用。但另一方面，您将使用这些锁完全破坏该对象上的线程。
嗯，为什么我需要使两个方法同步？ I如果我只锁定进程方法，这意味着在任何给定时间只有一个线程可以访问它。但是，考虑到一个线程安全的集合，并且我不关心在处理它时是否有任何其他线程添加到集合中，我不需要在 add 方法上加锁，不是吗？
如果只锁定processAndClear 方法，我描述的错误仍然可能发生。一次只能处理一个线程，但同时任何其他线程都可以在不方便的时候添加其他元素。
添加好像没问题吧？队列仍然是线程安全的，例如 LinkedBlockingQueue，因此即使在我们处理时从其他线程发生插入，这在线程安全方面也很好。

【解决方案2】：

该类将存储对象，并在调用 processAndClear 时遍历当前存储的对象，以某种方式处理它们，然后清除存储。

看来您应该使用BlockingQueue 来完成此任务。您的 add(...) 方法将添加到队列中，您的消费者将调用 take() 阻止等待下一个项目。 BlockingQueue（ArrayBlockingQueue 是一个典型的实现）为您处理所有的同步和信令。

这意味着您不必拥有CopyOnWriteArrayList 或AtomicReference。您将失去的是一个集合，并且您可以出于其他原因进行迭代，而不是您当前的帖子。

【讨论】：