1. Introduction
The final put up offered the elemental ideas of the reactive method, together with a specification and its implementation. Now it’s time to present the place the reactive method shines, together with these prospects which are typically ignored.
2. Reactor docs
One of many nice options of reactive libraries is the graphical descriptions of the operators. They’re referred to as marble diagrams and, in most circumstances, signify the supply flux, the motion the operator performs, and the ensuing flux. For instance, right here is the .map() marble diagram. Such graphic representations enable builders to shortly grasp the essential precept of an operator, with a few exceptions.
3. Flux/Mono states
Earlier than going additional, it’s essential to make clear the essential performance omitted within the earlier put up. As the reader can keep in mind, there are three strategies within the Subscriber interface:
public static interface Subscriber<T> {
public void onSubscribe(Subscription subscription);
public void onNext(T merchandise);
public void onError(Throwable throwable);
public void onComplete();
}
They signify callbacks for indicators {that a} writer might ship downstream. Aside from the common factor requested by a subscriber, the writer can signalize about an error that occurred whereas offering the weather, in addition to concerning the completion (assuming that the writer is finite). The Venture Reactor presents a number of operators primarily based on these indicators. They range from aspect impact ones, like doOnError, to behavioral, like switchIfEmpty and onErrorResume. This a part of the API offers the power to deal with distinctive conditions analogous to the attempt/catch block within the crucial programming mannequin.
4. Operator .publishOn()
So as to carry out a thread switching through .subscribeOn() a Writer must be created to be subscribed. However what if we need to change a thread for the operator that doesn’t create a nested writer, like .map() or .buffer()? That’s the place the .publishOn() comes into the image: it permits to modify a thread for all operators downstream:
Flux.vary(1, 10)
.map(i -> {
System.out.printf("[%s] Mapping %sn", Thread.currentThread().getName(), i);
return i * 2;
})
.publishOn(Schedulers.parallel())
.filter(i -> {
System.out.printf("[%s] Filtering %sn", Thread.currentThread().getName(), i);
return i >= 10;
})
.subscribeOn(Schedulers.boundedElastic())
.subscribe();
// Output:
// ---
// [boundedElastic-1] Mapping 1
// [boundedElastic-1] Mapping 2
// [parallel-1] Filtering 2
// [parallel-1] Filtering 4
// [boundedElastic-1] Mapping 3
// [boundedElastic-1] Mapping 4
// [parallel-1] Filtering 6
// [boundedElastic-1] Mapping 5
// [parallel-1] Filtering 8
// [boundedElastic-1] Mapping 6
// [parallel-1] Filtering 10
// [parallel-1] Filtering 12
// [boundedElastic-1] Mapping 7
// [boundedElastic-1] Mapping 8
// [parallel-1] Filtering 14
// [boundedElastic-1] Mapping 9
// [boundedElastic-1] Mapping 10
// [parallel-1] Filtering 16
// [parallel-1] Filtering 18
// [parallel-1] Filtering 20
As could be seen, .map() is executed on the scheduler configured for the entire chain through .subscribeOn(), whereas for .filter() the scheduler was switched. It is extremely vital to notice that .subscribeOn() impacts your complete chain, so its placement doesn’t matter whereas .publishOn() impacts solely the downstream.
5. Greatest use circumstances
With an important features coated, we’re able to dive into follow. The Venture Reactor API is designed to course of a flux of knowledge in a practical model asynchronously. Thus, naturally, it is finest fitted to duties that contain a flux (or stream) of knowledge and require asynchronous processing, reminiscent of described additional.
5.1 Streaming IO
IO is the commonest purpose when the reactive method is taken into account: the essential concept is to dump the blocking IO to one other thread pool whereas holding “employee threads” all the time busy with precise processing. This method works particularly effectively when IO is carried out on a stream of knowledge, like a file or results of a database, and with some kind of reactive driver plugged in. An ideal instance can be the r2dbc driver mixed with ORM that helps reactive, like jOOQ.
Let’s display it on the acquainted downside of parallel names counting from earlier posts. To do that, let’s transfer the names right into a file and offload studying to the devoted thread pool:
public class IoReactiveExampleMain {
non-public static remaining Scheduler IO = Schedulers.newParallel("IO");
public static void important(String[] args) throws Exception {
int batchSize = 1000;
var finalCounts = Flux.fromStream(Recordsdata.strains(Paths.get("src/important/assets/names.txt")))
// Break up to batches
.buffer(batchSize)
.subscribeOn(IO)
.doOnNext(__ -> System.out.printf("[%s] batch is providedn", Thread.currentThread().getName()))
// Combination intermediate counts asynchronously
.flatMap(IoReactiveExampleMain::processBatch)
.cut back(new HashMap<>(), IoReactiveExampleMain::mergeIntermediateCount)
.flatMapIterable(HashMap::entrySet);
String mostFrequentName = MathFlux.max(finalCounts, Map.Entry.comparingByValue())
.map(Map.Entry::getKey)
.block();
System.out.printf("Essentially the most frequent title is %sn", mostFrequentName);
}
non-public static HashMap<String, Lengthy> mergeIntermediateCount(HashMap<String, Lengthy> acc,
Map<String, Lengthy> intermediateResult) {
intermediateResult.forEach((title, intermediateCount) -> acc.merge(title, intermediateCount,
Lengthy::sum));
return acc;
}
non-public static Mono<Map<String, Lengthy>> processBatch(Checklist<String> batch) {
return Flux.fromIterable(batch)
.groupBy(Operate.id())
.flatMap(group -> group.rely().map(rely -> Tuples.of(group.key(), rely)))
.collectMap(Tuple2::getT1, Tuple2::getT2)
.doOnSubscribe(__ -> System.out.printf("[%s] Processing batch... n", Thread.currentThread().getName()))
.subscribeOn(Schedulers.boundedElastic());
}
}
Discover how seamlessly it’s built-in with the Stream API: a writer created through Flux#fromStream() offers new names from the file lazily, solely when requested through the .flatMap(). This ensures that reminiscence gained’t be overflowed if batch processing is slower than file studying. Additionally, the Flux#fromStream() closes the underlying stream when it’s completed, which is critical when utilizing the Recordsdata#strains().
Now, let’s implement the identical through the Loom API:
public class IoLoomExampleMain {
public static void important(String[] args) throws Exception {
remaining var thread = Thread.ofVirtual().begin(IoLoomExampleMain::begin);
thread.be part of();
}
non-public static void begin() {
attempt (var batchScope = new BatchScope()) {
int batchSize = 1000;
attempt (var strains = Recordsdata.strains(Paths.get("src/important/assets/names.txt"))) {
remaining var iterator = strains.iterator();
attempt (var fileReadingScope = new StructuredTaskScope.ShutdownOnFailure()) {
whereas (iterator.hasNext()) {
remaining var batchFuture = fileReadingScope.fork(() -> prepareBatch(batchSize, iterator));
fileReadingScope.be part of();
batchScope.fork(prepareBatchProcessing(batchFuture.resultNow()));
}
}
}
batchScope.be part of();
System.out.println("Essentially the most frequent title is " + batchScope.mostFrequentName());
} catch (Exception e) {
throw new RuntimeException(e);
}
}
non-public static ArrayList<String> prepareBatch(int batchSize, Iterator<String> iterator) {
System.out.printf("[%s][virtual=%s] Getting ready batch... n", LocalDateTime.now(), Thread.currentThread().isVirtual());
ArrayList<String> batch = new ArrayList<>(batchSize);
whereas (iterator.hasNext() && batch.dimension() < batchSize) {
batch.add(iterator.subsequent());
}
return batch;
}
non-public static Callable<Map<String, Lengthy>> prepareBatchProcessing(Checklist<String> batch) {
return () -> {
Map<String, Lengthy> localCounts = new ConcurrentHashMap<>();
System.out.printf("[%s][virtual=%s] Processing batch... n", LocalDateTime.now(), Thread.currentThread().isVirtual());
for (String title : batch) {
localCounts.compute(title, (n, c) -> c == null ? 1L : c + 1);
}
return localCounts;
};
}
non-public static class BatchScope extends StructuredTaskScope<Map<String, Lengthy>> {
non-public remaining ConcurrentHashMap<String, Lengthy> outcome = new ConcurrentHashMap<>();
@Override
protected void handleComplete(Future<Map<String, Lengthy>> future) {
remaining var intermediateResult = future.resultNow();
for (var stringLongEntry : intermediateResult.entrySet()) {
outcome.compute(stringLongEntry.getKey(), (n, c) -> updateCount(stringLongEntry.getValue(), c));
}
}
non-public lengthy updateCount(Lengthy newCount, Lengthy existingCount) {
return existingCount == null ? newCount : existingCount + newCount;
}
public String mostFrequentName() {
return outcome.entrySet()
.stream()
.max(Map.Entry.comparingByValue())
.get()
.getKey();
}
}
}
Whereas the Structured Concurrency API is within the incubating stage and finest practices are usually not but shaped, that’s, conceptually, the supposed approach to do concurrency in future variations of java.
Whereas it could appear extra acquainted and simple, it requires builders to carry out lots of boilerplate crucial actions (like batch preparation, for this process). Additionally, since there is no such thing as a approach to sign the backpressure within the Structured Concurrency API, the offered resolution will overflow reminiscence when batch processing is sluggish. Builders must implement the signaling system themselves to keep away from that.
5.2 Occasions
The subscription mannequin of the reactive method makes it very acceptable for duties associated to occasions, like numerous listeners and occasion buses. To realize this, the Venture Reactor offers just a few methods to create a Flux programmatically through a Sink abstraction. Right here is an instance of making a Flux that wraps the theoretical listener interface:
interface MyEventListener<T> {
void onDataChunk(Checklist<T> chunk);
void processComplete();
}
Flux<String> bridge = Flux.create(sink -> {
myEventProcessor.register(
new MyEventListener<String>() {
public void onDataChunk(Checklist<String> chunk) {
for (String s : chunk) {
sink.subsequent(s);
}
}
public void processComplete() {
sink.full();
}
});
});
That is described in additional element in the documentation. It must be famous that with this wrapper, it’s now potential to subscribe to the bridge and use all accessible Reactor API to buffer/filter/rework/carry out any vital logic with incoming occasions.
The earlier instance described a wrapper for the exterior occasion supply that supposed to have a single subscription. With Reactor, this can be very simple to raise this to full-blown bus performance with broadcasting help. That is achieved through the Sinks API primarily based on the Processor interface from the Reactive Streams Specification. With the Sinks API implementation of a bus could be achieved as follows:
public class ReactorEventBusImpl<T> {
non-public remaining Sinks.Many<T> sink = Sinks.many().replay().restrict(5);
non-public remaining Flux<T> eventFlux = sink.asFlux()
.publishOn(Schedulers.newParallel("bus"));
public void publish(T occasion) {
sink.emitNext(occasion, Sinks.EmitFailureHandler.busyLooping(Period.ofMinutes(1)));
}
public Flux<T> obtain() {
return eventFlux;
}
}
On this explicit instance, the bus permits many subscribers and replays as much as 5 parts for brand spanking new ones:
ReactorEventBusImpl<Integer> bus = new ReactorEventBusImpl<>();
remaining var firstSubscription = bus.obtain()
.subscribeOn(Schedulers.boundedElastic())
.doOnNext(i1 -> System.out.printf("[%s][%s] acquired %sn", 0, Thread.currentThread().getName(), i1))
.subscribe();
Thread.sleep(Period.ofSeconds(1));
System.out.println("Sending 1..10");
IntStream.vary(1, 11)
.forEach(bus::publish);
Thread.sleep(Period.ofSeconds(1));
remaining var secondSubscription = bus.obtain()
.subscribeOn(Schedulers.boundedElastic())
.doOnNext(i -> System.out.printf("[%s][%s] acquired %sn", 1, Thread.currentThread().getName(), i))
.subscribe();
Thread.sleep(Period.ofSeconds(1));
System.out.println("Sending 11..20");
IntStream.vary(11, 21)
.forEach(bus::publish);
// Output:
// ---
// Sending 1..10
// [0][bus-1] acquired 1
// [0][bus-1] acquired 2
// [0][bus-1] acquired 3
// [0][bus-1] acquired 4
// [0][bus-1] acquired 5
// [0][bus-1] acquired 6
// [0][bus-1] acquired 7
// [0][bus-1] acquired 8
// [0][bus-1] acquired 9
// [0][bus-1] acquired 10
// [1][bus-2] acquired 6
// [1][bus-2] acquired 7
// [1][bus-2] acquired 8
// [1][bus-2] acquired 9
// [1][bus-2] acquired 10
// Sending 11..20
// [0][bus-1] acquired 11
// [0][bus-1] acquired 12
// [0][bus-1] acquired 13
// [0][bus-1] acquired 14
// [0][bus-1] acquired 15
// [0][bus-1] acquired 16
// [0][bus-1] acquired 17
// [0][bus-1] acquired 18
// [0][bus-1] acquired 19
// [0][bus-1] acquired 20
// [1][bus-2] acquired 11
// [1][bus-2] acquired 12
// [1][bus-2] acquired 13
// [1][bus-2] acquired 14
// [1][bus-2] acquired 15
// [1][bus-2] acquired 16
// [1][bus-2] acquired 17
// [1][bus-2] acquired 18
// [1][bus-2] acquired 19
// [1][bus-2] acquired 20
The offered implementation could be simply reconfigured to a special conduct through the Sinks API.
The Venture Loom might assist with this use case both, however because it was not designed particularly for it, builders would need to implement the bus primarily based on a concurrent Queue from scratch, together with the administration of a number of subscribers, and many others. Due to that, it doesn’t make a lot sense to check the Tasks Loom and Reactor for this explicit use case.
5.3 Resiliency
One (in all probability, surprising) final result of the subscription mannequin is how simple it’s to retry execution on an error: it’s sufficient to easily resubscribe to a writer! Due to this trait, the Venture Reactor offers a wealthy retry facility out-of-box. The only kind is a .retry() operator that merely resubscribes to the writer in case of error indefinitely or configured variety of instances:
Flux.simply(1, 2, 3, 4, 5, 6)
.doOnNext(i -> {
if (i != 3) {
System.out.printf("[%s] Received %sn", Thread.currentThread().getName(), i);
} else {
System.out.printf("[%s] Received unlawful %sn", Thread.currentThread().getName(), i);
throw new IllegalStateException("Received unlawful 3");
}
})
.retry(2)
.subscribeOn(Schedulers.boundedElastic())
.blockLast();
// Output:
// ---
// [boundedElastic-1] Received 1
// [boundedElastic-1] Received 2
// [boundedElastic-1] Received unlawful 3
// [boundedElastic-1] Received 1
// [boundedElastic-1] Received 2
// [boundedElastic-1] Received unlawful 3
// [boundedElastic-1] Received 1
// [boundedElastic-1] Received 2
// [boundedElastic-1] Received unlawful 3
// Exception in thread "important" java.lang.IllegalStateException: Received unlawful 3
// at javacodegeeks.reactor.examples.important/org.javacodegeeks.examples.PresentTimeExampleMain.lambda$important$0(PresentTimeExampleMain.java:24)
// ...
For extra superior use circumstances, the Venture Reactor offers the .retryWhen() operator and Retry class that works in conjunction. The Retry API permits advantageous configuration of retries. For instance, with exponential backoff, with callbacks invoked earlier than and after retry, ought to the error counter be reset on a profitable retry try or not, and much more:
public class RetryWhenMain {
public static remaining RetryBackoffSpec RETRY_SPEC =
Retry.backoff(3, Period.ofSeconds(2))
.doBeforeRetry(retrySignal -> System.out.printf("[%s][%s] Error, earlier than retryn", now(), Thread.currentThread().getName()))
.doAfterRetry(retrySignal -> System.out.printf("[%s][%s] Error, after retryn", now(), Thread.currentThread().getName()))
.scheduler(Schedulers.boundedElastic())
.transientErrors(false);
public static void important(String[] args) {
Flux.simply(1, 2, 3, 4, 5, 6)
.doOnNext(i -> {
if (i != 3) {
System.out.printf("[%s][%s] Received %sn", now(), Thread.currentThread().getName(), i);
} else {
System.out.printf("[%s][%s] Received unlawful %sn", now(), Thread.currentThread().getName(), i);
throw new IllegalStateException("Received unlawful 3");
}
})
.retryWhen(RETRY_SPEC)
.subscribeOn(Schedulers.parallel())
.blockLast();
}
}
// Output:
// ---
// [2022-11-02T19:04:00.316492874][parallel-1] Received 1
// [2022-11-02T19:04:00.317204359][parallel-1] Received 2
// [2022-11-02T19:04:00.317347538][parallel-1] Received unlawful 3
// [2022-11-02T19:04:00.322003149][parallel-1] Error, earlier than retry
// [2022-11-02T19:04:03.176562622][boundedElastic-1] Error, after retry
// [2022-11-02T19:04:03.177192567][boundedElastic-1] Received 1
// [2022-11-02T19:04:03.177669363][boundedElastic-1] Received 2
// [2022-11-02T19:04:03.178068239][boundedElastic-1] Received unlawful 3
// [2022-11-02T19:04:03.178973002][boundedElastic-1] Error, earlier than retry
// [2022-11-02T19:04:07.523104743][boundedElastic-2] Error, after retry
// [2022-11-02T19:04:07.523776491][boundedElastic-2] Received 1
// [2022-11-02T19:04:07.524500027][boundedElastic-2] Received 2
// [2022-11-02T19:04:07.524941269][boundedElastic-2] Received unlawful 3
// [2022-11-02T19:04:07.525736592][boundedElastic-2] Error, earlier than retry
// [2022-11-02T19:04:15.969594038][boundedElastic-1] Error, after retry
// [2022-11-02T19:04:15.970121536][boundedElastic-1] Received 1
// [2022-11-02T19:04:15.970579856][boundedElastic-1] Received 2
// [2022-11-02T19:04:15.971036717][boundedElastic-1] Received unlawful 3
// Exception in thread "important" reactor.core.Exceptions$RetryExhaustedException: Retries exhausted: 3/3
// ...
// Suppressed: java.lang.Exception: #block terminated with an error
// at reactor.core@3.4.23/reactor.core.writer.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:99)
// at reactor.core@3.4.23/reactor.core.writer.Flux.blockLast(Flux.java:2645)
// at javacodegeeks.reactor.examples.important/org.javacodegeeks.examples.v3.retry.RetryWhenMain.important(RetryWhenMain.java:33)
// Brought on by: java.lang.IllegalStateException: Received unlawful 3
// at javacodegeeks.reactor.examples.important/org.javacodegeeks.examples.v3.retry.RetryWhenMain.lambda$important$2(RetryWhenMain.java:28)
// ...
Different use circumstances associated to resiliency, like fee limiting, are offered by the resilience4j library, which is built-in with the Reactor very effectively.
5.4 Price mentioning
Offered examples are simply the tip of the iceberg which is the Venture Reactor API. Throughout its evolution, it accrued operators for nearly each conceivable process. To focus on just a few typically ignored however extraordinarily helpful examples:
5.4.1 Operators .utilizing*()
Meant for use for closable assets. Analogous to the try-with-resources performance.
5.4.2 Operators .swtich*()
Enable switching to a different writer on an outlined situation: when the upstream is empty, or on the primary factor. This method another move could also be simply offered.
5.4.3 Operators .increase*()
Operators that enable the recursive provision of parts. This could be a lifesaver for integration with paginated APIs, for instance.
5.4.5 Augmenting operators
This set of operators permits to assign an index for parts of flux or to offer elapsed time from the second of subscription for every factor.
5.4.6 Observability operators
These operators enable to log indicators in a reactive chain or acquire metrics utilizing Micrometer.
5.4.7 Reactive context
In Java purposes it’s a widespread sample to make use of ThreadLocals to retailer a state, however, since reactive code is executed throughout a number of asynchronous boundaries, they aren’t match for this function. As a substitute, Reactor offers the Context API that enables assigning a state tied to the Flux, not thread.
5.4.8 Operators .rework*()
In case no operators present the required performance, it’s potential to implement a customized operator and apply it to a Flux through .transformDeffered() operator. The resilience4j library is applied utilizing this method.
5.4.9 Operators information
So as to describe the offered performance in a understandable method, the docs have a “Which operator do I would like” part, describing virtually each use case coated by API. Right here builders can get a grasp of what’s provided and whether or not is it appropriate for the duty at hand.
6. Conclusion
This weblog put up demonstrated use circumstances the place the reactive method fits higher than the Loom and Structured Concurrency, in the meanwhile of writing. It additionally highlighted some typically ignored performance of the API. The one main a part of API which was not talked about is sizzling vs chilly publishers, which readers are urged to discover on their very own, in addition to the remainder of the accessible performance.
The following put up will display the other: the place it doesn’t make sense to make use of the reactive method in gentle of the Venture Loom, in addition to the downsides and caveats of it normally, and on the JVM platform, specifically.
