Java: how to return intermediate results from a stream

Question

Using Java 7 I am trying to create an observer that monitors a data store (some type of collection) and then returns some elements from it at certain points. In this case, they are timestamps when the timestamp skips the current time, when I want it to be returned to the original thread. See code below.

@Override public void run() { while (!data.isEmpty()) { for (LocalTime dataTime : data) { if (new LocalTime().isAfter(dataTime)) { // return a result but continue running } } } } 

I read about future and defiant ones, but they seem to stop the thread on return.

I don’t really want to return the value and stop the thread, and then start another task if you use the called, if that is not the best way.

What are the best methods to find this? It seems like such a wide selection.

thanks

Answer 1

You can put intermediate results in a lock queue so that results are available to consumer threads when and when they become available:

 private final LinkedBlockingQueue<Result> results = new LinkedBlockingQueue<Result>(); @Override public void run() { while (!data.isEmpty()) { for (LocalTime dataTime : data) { if (new LocalTime().isAfter(dataTime)) { results.put(result); } } } } public Result takeResult() { return results.take(); } 

Consumer threads can simply call the takeResult method to use intermediate results. The advantage of using a blocking queue is that you do not need to reinvent the wheel, as it looks like a typical producer-consumer scenario, which can be solved using a blocking data structure.

Note Here Result may be a "POJO" representing an intermediate result object.

Answer 2

You are on the right track. Suppose that the correct synchronization will be there, and you will receive all your timestamps in time :) You should ideally choose a data structure that does not require scanning of all elements. Select something like a mini-heap or some up / down lists, and now that you are repeating, just remove the item from this data store and put it in a blocking queue. there is a thread that listens on this queue to continue on.