java - For-each loop on a list that grows simultaneously

Question

I have a for-each loop in java.

The for-each runs on myList in a thread. myList can grow simultaneously. I have some questions:

If I started the for-each loop and after it started, an item was added to the list, will the actions in the for-each loop run on it?
Suppose that the answer to the question above is no. I have kind of a problem. The for-each loop is in a while(true) loop so it will start over. I want the for-each loop to run on each item once. I cannot remove items during the for-each loop because I get ConcurrentModificationException. So my solution is to remove all list items after the for-each loop ends. But, this way, if an item added to the list after the for-each loop started it was deleted too, the for-each loop will never run on this item.

My goal is to make a for-each that runs on a list that can grow simultaneously. I want the for-each loop to never miss an item and never run the on same item twice or more. What is the solution?

score 6 · Accepted Answer

Use of Iterator.remove would allow you not to hit the ConcurrentModificationException, but another solution would be to not use a foreach loop and simply loop, something like:

// assuming that this is a list of Strings
List<String> list = ...
while(!list.isEmpty())) {
    String data = list.remove(0);
    ...process data...
}

This would allow you to process each item added to the list and to do so only once. There is a small window above though where the isEmpty could return true and a new item could be added to the list (this could happen in a multi-threaded environment).

score 4 · Accepted Answer

It's a Typical producer consumer problem. You are not supposed to use List or any implementations of list. As List is index based addition/removal of items to the list modifies the indexes of other elements.

Trying using any implementation of Queue.

In your case other threads (Producers) would enqueue to the queue and the block of code/thread (consumer) which runs the foreach block should just dequeue from the queue and do your processing.

Let me know if that serves the purpose. If my understanding of your use case is wrong , please clarify.

--

Vinod

score 0 · Accepted Answer

I think you are looking for a kind of double-buffering of your list.

I have tested this with multiple producers and multiple consumers and it seems to work perfectly.

Essentially you need to hold a list which, when requested, is replaced by a new empty one. Handling threading correctly while swapping adds a little complexity. This handles multiple threads adding to the list as well as multiple threads grabbing lists for iteration.

Note that a slight change in your architecture (pulling entries one-at-a-time from the list) would mean you could use a BlockingQueue which may be a better solution for you.

public class DoubleBufferedList<T> {
  // Atomic reference so I can atomically swap it through.
  // Mark = true means I am adding to it so unavailable for iteration.
  private AtomicMarkableReference<List<T>> list = new AtomicMarkableReference<List<T>>(newList(), false);

  // Factory method to create a new list - may be best to abstract this.
  protected List<T> newList() {
    return new ArrayList<T>();
  }

  // Get and replace the current list.
  public List<T> getList() {
    // Atomically grab and replace the list with an empty one.
    List<T> empty = newList();
    List<T> it;
    // Replace an unmarked list with an empty one.
    if (!list.compareAndSet(it = list.getReference(), empty, false, false)) {
      // Failed to replace! 
      // It is probably marked as being appended to but may have been replaced by another thread.
      // Return empty and come back again soon.
      return Collections.EMPTY_LIST;
    }
    // Successfull replaced an unmarked list with an empty list!
    return it;
  }

  // Add an entry to the list.
  public void addToList(T entry) {
    List<T> it;
    // Spin on get and mark.
    while (!list.compareAndSet(it = list.getReference(), it, false, true)) {
      // Spin on mark.
    }
    // Successfully marked! Add my new entry.
    it.add(entry);
    // Unmark it. Should never fail because once marked it will not be replaced.
    if (!list.attemptMark(it, false)) {
      throw new IllegalMonitorStateException("it changed while we were adding to it!");
    }
  }
}

java - For-each loop on a list that grows simultaneously

3 回答 3

Related

Reference