MACRO NEPAL
Taming Asynchronous Data: A Guide to Reactive Streams in Java
awiskar acharya
November 4, 2025
2:20 am

In the era of microservices, big data, and real-time applications, traditional synchronous and blocking programming models often fall short. Systems need to handle massive streams of data efficiently while remaining responsive under heavy load. This is where Reactive Streams comes in—a formal specification that provides a standard for asynchronous stream processing with non-blocking back pressure.

This article explores the Reactive Streams specification, its core components, and how it enables the development of resilient, responsive, and elastic applications in Java.

The Problem: Asynchronous Data Flow Challenges

Before Reactive Streams, handling asynchronous data streams between producers and consumers presented several challenges:

  1. Memory Management: A fast producer could overwhelm a slow consumer, causing OutOfMemoryError exceptions.
  2. Complex Error Handling: Propagating errors across asynchronous boundaries was difficult and error-prone.
  3. Resource Control: There was no standard way for consumers to signal how much data they could handle.
  4. Composition Difficulty: Combining multiple asynchronous operations often led to "callback hell."

The critical missing piece was back pressure—the ability for consumers to control the rate at which they receive data from producers.

What is Reactive Streams?

Reactive Streams is an initiative to provide a standard for asynchronous stream processing with non-blocking back pressure. It defines a minimal set of interfaces, methods, and protocols that describe the necessary operations and entities to achieve asynchronous data streams with back pressure.

Key Characteristics:

  • Asynchronous: Operations don't block threads while waiting for data
  • Non-blocking: Threads can perform other work while I/O operations complete
  • Back pressure: Consumers can control the flow from producers
  • Composable: Operations can be chained together declaratively

The Four Core Interfaces

The Reactive Streams specification defines just four interfaces in the org.reactivestreams package:

1. Publisher

public interface Publisher<T> {
void subscribe(Subscriber<? super T> s);
}
  • Produces a stream of data
  • Accepts a Subscriber that will consume the data
  • Each Publisher can have multiple Subscribers

2. Subscriber

public interface Subscriber<T> {
void onSubscribe(Subscription s);
void onNext(T t);
void onError(Throwable t);
void onComplete();
}
  • Consumes the data stream
  • Receives notifications through four callback methods
  • Controls flow using the Subscription object

3. Subscription

public interface Subscription {
void request(long n);
void cancel();
}
  • Represents the one-to-one lifecycle between a Publisher and Subscriber
  • Controls the flow of data through request(n)
  • Allows cancellation of the stream

4. Processor

public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
}
  • Acts as both a Subscriber and a Publisher
  • Used for transforming data between stages in a stream pipeline

The Subscription Lifecycle Protocol

The interaction between these components follows a strict protocol:

Publisher --(onSubscribe)--> Subscriber
Subscriber --(request(n))--> Publisher  
Publisher --(onNext×n)--> Subscriber
Publisher --(onComplete/onError)--> Subscriber

Step-by-Step Lifecycle:

  1. Subscription: A Subscriber subscribes to a Publisher
  2. Activation: The Publisher calls onSubscribe(Subscription) on the Subscriber
  3. Demand Signaling: The Subscriber calls subscription.request(n) to request data
  4. Data Flow: The Publisher calls onNext(T) up to n times
  5. Completion: The Publisher calls onComplete() when no more data is available
  6. Error Handling: The Publisher calls onError(Throwable) if an error occurs
  7. Cancellation: The Subscriber can call subscription.cancel() at any time

Manual Implementation Example

Here's a simple manual implementation to demonstrate the protocol:

import org.reactivestreams.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;
// A simple Publisher that emits numbers from 1 to N
class NumberPublisher implements Publisher<Long> {
private final long count;
private final ExecutorService executor = Executors.newFixedThreadPool(2);
public NumberPublisher(long count) {
this.count = count;
}
@Override
public void subscribe(Subscriber<? super Long> subscriber) {
NumberSubscription subscription = new NumberSubscription(subscriber, count, executor);
subscriber.onSubscribe(subscription);
}
}
// The Subscription implementation
class NumberSubscription implements Subscription {
private final Subscriber<? super Long> subscriber;
private final long total;
private final ExecutorService executor;
private final AtomicLong pendingRequests = new AtomicLong(0);
private long currentValue = 1;
public NumberSubscription(Subscriber<? super Long> subscriber, long total, ExecutorService executor) {
this.subscriber = subscriber;
this.total = total;
this.executor = executor;
}
@Override
public void request(long n) {
if (n <= 0) {
subscriber.onError(new IllegalArgumentException("Request must be positive"));
return;
}
pendingRequests.addAndGet(n);
executor.execute(this::processItems);
}
@Override
public void cancel() {
pendingRequests.set(-1); // Signal cancellation
}
private void processItems() {
long emitted = 0;
while (emitted < pendingRequests.get() && currentValue <= total) {
if (pendingRequests.get() == -1) return; // Cancelled
subscriber.onNext(currentValue++);
emitted++;
if (currentValue > total) {
subscriber.onComplete();
return;
}
}
pendingRequests.addAndGet(-emitted);
}
}
// A simple Subscriber implementation
class SimpleSubscriber implements Subscriber<Long> {
private Subscription subscription;
private final String name;
public SimpleSubscriber(String name) {
this.name = name;
}
@Override
public void onSubscribe(Subscription s) {
this.subscription = s;
System.out.println(name + ": Subscribed");
// Request first 3 items
subscription.request(3);
}
@Override
public void onNext(Long item) {
System.out.println(name + ": Received " + item);
// Simulate processing time
try { Thread.sleep(100); } catch (InterruptedException e) {}
// Request one more item after processing each
subscription.request(1);
}
@Override
public void onError(Throwable t) {
System.out.println(name + ": Error - " + t.getMessage());
}
@Override
public void onComplete() {
System.out.println(name + ": Completed!");
}
}
// Usage example
public class ReactiveStreamsDemo {
public static void main(String[] args) throws InterruptedException {
Publisher<Long> publisher = new NumberPublisher(10);
Subscriber<Long> subscriber = new SimpleSubscriber("Subscriber-1");
publisher.subscribe(subscriber);
Thread.sleep(2000); // Allow time for processing
}
}

Output:

Subscriber-1: Subscribed
Subscriber-1: Received 1
Subscriber-1: Received 2
Subscriber-1: Received 3
Subscriber-1: Received 4
Subscriber-1: Received 5
... (continues to 10)
Subscriber-1: Completed!

Reactive Streams in Practice: Popular Implementations

While you can implement the interfaces directly, most developers use established Reactive Streams implementations:

1. Project Reactor (Spring WebFlux default)

Flux.range(1, 10)
.delayElements(Duration.ofMillis(100))
.subscribe(item -> System.out.println("Received: " + item));

2. RxJava

Observable.range(1, 10)
.zipWith(Observable.interval(100, TimeUnit.MILLISECONDS), (i, t) -> i)
.subscribe(item -> System.out.println("Received: " + item));

3. Akka Streams

Source.range(1, 10)
.throttle(1, Duration.ofMillis(100))
.runForeach(i -> System.out.println("Received: " + i), materializer);

4. Java 9+ Flow API
The Reactive Streams interfaces were incorporated into Java 9 as the java.util.concurrent.Flow class, which contains the same four interfaces.

Key Benefits of Reactive Streams

  1. Back Pressure: Prevents fast producers from overwhelming slow consumers
  2. Resource Efficiency: Non-blocking operations enable high concurrency with fewer threads
  3. Error Resilience: Built-in error propagation and recovery mechanisms
  4. Composition: Declarative API for building complex data processing pipelines
  5. Interoperability: Different implementations can work together seamlessly

When to Use Reactive Streams

  • High-Throughput Systems: Applications processing large volumes of streaming data
  • Microservices Architectures: Communication between services with varying processing capabilities
  • Real-Time Applications: Chat systems, live data feeds, stock tickers
  • Network-Intensive Applications: Proxy servers, API gateways, message brokers

When to Consider Alternatives

  • Simple CRUD Applications: Traditional blocking approaches may be sufficient
  • Synchronous Workflows: When operations must execute in strict sequence
  • Team Expertise: If the team lacks experience with reactive programming concepts

Conclusion

The Reactive Streams specification provides a crucial foundation for building modern, scalable Java applications. By standardizing asynchronous stream processing with non-blocking back pressure, it solves fundamental problems in distributed systems and high-load environments.

While the learning curve can be steep, understanding Reactive Streams is essential for Java developers working on:

  • Cloud-native applications
  • Real-time data processing systems
  • High-performance microservices
  • Responsive web applications

The specification's adoption into Java 9+ and its implementation in major frameworks like Spring WebFlux and Akka demonstrate its importance in the future of Java development. By mastering Reactive Streams, developers can build systems that are truly responsive, resilient, and elastic—the core principles of reactive systems.

Category : JAVA

Leave a Reply

Your email address will not be published. Required fields are marked *


Macro Nepal Helper