Coordinating Threads with Reactive Streams
You are jumping through a lot of hoops to get your queue, consumer, and observer strategies aligned. I look at your code, and I don't really follow it all. Part of the reason is that I don't know RxJava at all, but, normally, I can follow these things without much research.

So, your code streams the tasks, for each one, it creates an Observable for it which is subscribed on the Reactive computation engine. The observer for the Observable is set to add the result to the queue.

Now, for the queue, you create an Observer that registers one action for each expected result. You then loop over those events, and print the result.

I can understand why you may want to use RxJava for some situations, but, that last part makes no sense to me... Let me show you your code:


//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 -&gt; {
        System.out.println("Received " + item + " at " + Thread.currentThread().getName());
    });



where consumer is:


static &lt;T&gt; Observable&lt;T&gt; consumer(BlockingQueue&lt;T&gt; queue, int count) {
    return Observable.&lt;T&gt;create(subscriber -&gt; {
        for (int i = 0; i &lt; count; i++) {
            try {
                T text = queue.take();
                subscriber.onNext(text);
            }
            catch (InterruptedException ex) {
                subscriber.onError(ex);
            }
        }
        subscriber.onCompleted();
    }).observeOn(Schedulers.immediate());
}



What does that code do? It looks for three strings on a queue, and prints them.

for (int i = 0; i &lt; 3; i++) {
    System.out.println("Received " + queue.take() + " at " + Thread.currentThread().getName());
}


What's wrong with that (apart from the 3 constant)? ^^^^

Having figured out all the loops you are going through, I struggled to understand the advantage that RxJava is giving you, especially when Java has already got the "right tool" in the java.util.concurrent.* toolbox.

Let me explain, the CompletionService concept says: Submit a bunch of jobs to the service, and the service will tell you when they complete. It has a submit(...) method to add jobs to the service, and a take(...) method which will wait until the next-completed task, and return it.

Creating a Thread Factory that matches the names you have in RxJava quickly, I have this code:

private static final AtomicInteger threadId = new AtomicInteger(0);

private static Thread threadFactory(Runnable r) {
    Thread t = new Thread(r, "RxComputationThreadPool-" + threadId.incrementAndGet());
    t.setDaemon(true);
    return t;
}

private static ExecutorService pool = Executors.newCachedThreadPool(Reactive::threadFactory);


That code creates an ExecutorService which has threads named things like your code's results.

Now, with that code available, I can have a really simple problem space:

public static void main(String[] args) throws InterruptedException, ExecutionException {
    List&lt;Callable&lt;String&gt;&gt; tasks = Arrays.asList(
            () -&gt; "One: " + Thread.currentThread().getName(),
            () -&gt; "Two: " + Thread.currentThread().getName(),
            () -&gt; "Three: " + Thread.currentThread().getName()
        );

    int count = tasks.size();

    CompletionService&lt;String&gt; completor = new ExecutorCompletionService&lt;&gt;(pool);

    tasks.stream().forEach(completor::submit);
    for (int i = 0; i &lt; count; i++) {
        System.out.println("Received: " + completor.take().get() + " at " + Thread.currentThread().getName());
    }

}


Note that the CompletionService returns a completed Future&lt;String&gt;, not the actual String, so you have to get() the string from the Future&lt;String&gt;.

I have put this code up on to ideone so you can see it in action.

Note, the solution I provided has no need for RxJava, and that's because, if you want to execute a bunch of things asynchronously, and retrieve the results in a single thread, then perhaps the observable pattern is not the right one....