Java Pills

🕐 Reading time: 2 minutes

It's been a first season of articles that entertained me and taught me a lot; after today's appointment, Java Pills will take a summer break and return in September!

Let's close the season with the second part of pill #17, focusing on .share(), to handle multiple concurrent subscribers effectively.

📌 Cold stream vs hot shared

When several subscribers subscribe to the same Flux at almost the same moment, the default behaviour – being cold – is to re-run the logic for each one. With .share(), instead, the Flux becomes hot:

• the first subscription activates the flow
• the following ones hook onto the same shared stream
• the source logic runs only once (as long as there is at least one active subscriber)

📑 The test (yes, it's a bit convoluted)

@Test
void test_buildFlux_withShare() throws InterruptedException {
    final AtomicInteger subscriptionCount = new AtomicInteger();
    //create a cold Flux that emits "A", "B", "C" and increments the counter on each new subscription
    final Flux<String> sharedFlux = buildFlux(subscriptionCount)
            .delayElements(Duration.ofMillis(100)) //simulate delay to keep the Flux active
            .share();                              //share the source among concurrent subscribers

    final List<String> result1 = new ArrayList<>();
    final List<String> result2 = new ArrayList<>();
    final CountDownLatch latch = new CountDownLatch(2); //to wait for both subscriptions to finish

    subscribeAsync(sharedFlux, result1, latch);
    subscribeAsync(sharedFlux, result2, latch);

    assertTrue(latch.await(2, TimeUnit.SECONDS)); //wait for both subscriptions to complete
    checkResults(result1, result2);
    //expect <1> because the subscription is shared among concurrent subscribers
    assertEquals(1, subscriptionCount.get());
}

//start both subscriber threads (they will subscribe almost simultaneously) in separate threads
private void subscribeAsync(final Flux<String> sharedFlux, final List<String> result, final CountDownLatch latch) {
    new Thread(() -> sharedFlux.collectList()
            .doOnNext(result::addAll)               //collect the emitted values
            .doFinally(signal -> latch.countDown()) //signal when done
            .block())                               //block until the Flux completes
        .start();
}

What happens in the test:

• the Flux emits the values "A", "B", "C" and (should) increment a counter every time a new subscription starts
• the two threads, however, subscribe almost in parallel
• without .share(), subscriptionCount would be 2
• with .share(), subscriptionCount is 1: the source runs only once, and the data (starting from the subscription) is sent to both subscribers

📌 When is it worth using share?

• simultaneous subscribers: when several threads or clients subscribe in parallel (or almost) to the same flow
• shared processing: to avoid duplicate computations on frequently changing data (e.g. microservice calls, heavy queries, Kafka streams)

⚠️ Limits and things to watch out for

• the key point: .share() is not a cache, and it's not a replay. It's an alias for .publish().refCount(1)
• each emitted element is broadcast only to the subscribers active at that moment
• if a subscriber subscribes after the flow has started, it misses the elements already emitted
• if all the subscribers disconnect, the flow stops: on the next subscription it will start over from scratch
• so: .share() only works if the subscribers are "live" at the same time

See you in September with the next pill! ☕