I just started learning reactive streams with RxJava. After reading a couple of books and a lot of articles our there I am still having trouble understanding how to coordinate multiple threads.
I would appreciate a critique of the following code, particularly if anyone knows a better way to do it.
I basically want to execute multiple tasks in the background, and as results start to get back I want to get immediately notified of the results. I want the background tasks to run in individual threads, but I want the notification thread to run in a single thread.
My idea is that the notification code does not need to worry about synchronization and I can achieve that by making sure the notification code only runs in one thread.
This was my best attempt of that:
public class Question {
public static void main(String[] args) {
//A hypothetical list of tasks to run asynchronously
List<Callable<String>> tasks = Arrays.asList(
() -> "One: " + Thread.currentThread().getName(),
() -> "Two: " + Thread.currentThread().getName(),
() -> "Three: " + Thread.currentThread().getName()
);
//A blocking queue to hold the results when ready
BlockingQueue<String> queue = new ArrayBlockingQueue<>(3);
//when a task is done, this observer puts it in the queue.
//observer code then will run in the currently processing thread
Observer<String> observer = Observers.create(queue::offer);
tasks.stream().map(Question::async).forEach(result -> result.subscribe(observer));
//as tasks get resolve and enter the queue, this other observer process the results
//in my current thread, not in any of the task threads.
consumer(queue, 3).forEach(item -> {
System.out.println("Received " + item + " at " + Thread.currentThread().getName());
});
}
static <T> Observable<T> async(Callable<T> supplier) {
return Observable.<T>create(subscriber -> {
try {
subscriber.onNext(supplier.call());
subscriber.onCompleted();
}
catch (Exception ex) {
if (!subscriber.isUnsubscribed()) {
subscriber.onError(ex);
}
}
}).subscribeOn(Schedulers.computation());
}
static <T> Observable<T> consumer(BlockingQueue<T> queue, int count) {
return Observable.<T>create(subscriber -> {
for (int i = 0; i < count; i++) {
try {
T text = queue.take();
subscriber.onNext(text);
}
catch (InterruptedException ex) {
subscriber.onError(ex);
}
}
subscriber.onCompleted();
}).observeOn(Schedulers.immediate());
}
}
And I get:
Received One: RxComputationThreadPool-1 at main Received Two: RxComputationThreadPool-2 at main Received Three: RxComputationThreadPool-3 at main
So, this works like a charm, but I have the feeling it is still too verbose and that maybe there is a way to make it simpler with RxJava somehow.