For each cycle in the list that grows simultaneously
I have every loop in java.
Each of them works in myList
a stream. myList
can grow at the same time. I have several questions:
- If I run the for-each loop and after it runs, the item has been added to the list, will the for-each loop do things?
- Suppose the answer to the question above is no. I have a problem. The for-each loop runs after some time (true), so it will start. I want the for-each loop to run once on each element. I am unable to remove elements during the for-each loop because I am getting
ConcurrentModificationException
. So my solution is to remove all list items after each loop ends. But, thus, if an item added to the list after starting each loop was also removed, the for-each loop will never work on that item.
My goal is to create for each one that works on a list that can grow at the same time. I want the for-each loop to never loop through an element and never run the same element twice or more. What's the solution?
source to share
Using Iterator.remove will allow you not to hit the ConcurrentModificationException, but another solution would be to not use a foreach loop and just a loop, like this:
// assuming that this is a list of Strings
List<String> list = ...
while(!list.isEmpty())) {
String data = list.remove(0);
...process data...
}
This will allow you to process each item added to the list and only do it once. There is a small window above, though, where isEmpty can return true and a new item can be added to the list (this can happen in a multi-threaded environment).
source to share
This is a typical consumer-producer problem. You shouldn't be using List or any list implementations. Since List is adding / removing elements based on indices, the list changes the indices of other elements.
Trying to use any queue implementation.
In your case, other threads (Producers) are queued, and the code / thread block (consumer) that starts the foreach block should just exit the queue and do your processing.
Let me know if this serves a purpose. If my understanding of your use case is incorrect, please clarify.
-
Vinod
source to share
I think you are looking for some sort of double buffering of your list.
I've tested this with multiple manufacturers and multiple consumers and it seems to work great.
Essentially, you need to keep the list, which is replaced with a new empty list when requested. Correct threading when replacing adds a bit of complexity. This handles multiple threads adding to the list as well as multiple threads grabbing lists for iteration.
Please note that a small change in your architecture (pulling entries one at a time in the list from the list) means you can use BlockingQueue
which might be the best 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!");
}
}
}
source to share