You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
Copy file name to clipboardExpand all lines: modules/howtos/pages/concurrent-async-apis.adoc
+59-16Lines changed: 59 additions & 16 deletions
Display the source diff
Display the rich diff
Original file line number
Diff line number
Diff line change
@@ -5,8 +5,23 @@
5
5
[abstract]
6
6
{description} This page outlines the different options with their drawbacks and benefits.
7
7
8
+
The main concurrency options offered by Java offer different optimizations for different use cases.
9
+
On this page we give a brief analysis of which may be best for your application, with links to further documentation to help you find the best solution.
10
+
11
+
12
+
== Virtual Threads
13
+
14
+
For high-throughput concurrent applications, https://docs.oracle.com/en/java/javase/21/core/virtual-threads.html[Java Virtual Threads^] can scale throughput, particularly for tasks which spend much of their time blocked.
15
+
16
+
Couchbase fully supports and recommends virtual threads and structured concurrency with the Java SDK, as the modern threading solution.
17
+
Virtual threads can help with scaling of applications using the blocking API, and also the asynchronous API.
18
+
19
+
8
20
== Reactive Programming with Reactor
9
-
You want to consider an asynchronous, reactive API if the blocking API does not suit your needs anymore. There are plenty of reasons why this might be the case, like more effective resource utilization, non-blocking error handling or batching together various operations. We recommend using the reactive API over the `CompletableFuture` counterpart because it provides all the bells and whistles you need to build scalable asynchronous stacks.
21
+
22
+
You want to consider an asynchronous, reactive API if the blocking API does not suit your needs anymore.
23
+
There are plenty of reasons why this might be the case, like more effective resource utilization, non-blocking error handling or batching together various operations.
24
+
We recommend using the reactive API over the `CompletableFuture` counterpart because it provides all the bells and whistles you need to build scalable asynchronous stacks.
10
25
11
26
Each blocking API provides access to its reactive counterpart through the `reactive()` accessor methods:
12
27
@@ -15,7 +30,9 @@ Each blocking API provides access to its reactive counterpart through the `react
The reactive API uses the https://projectreactor.io/[Project Reactor] library as the underlying implementation, so it exposes its `Mono` and `Flux` types accordingly. As a rule of thumb, if the blocking API returns a type `T` the reactive counterpart returns `Mono<T>` if one (or no) results is expected or in some cases `Flux<T>` if there are more than one expected. We *highly* recommend that you make yourself familar with the https://projectreactor.io/docs/core/release/reference/[reactor documentation] to understand its fundamentals and also unlock its full potential.
33
+
The reactive API uses the https://projectreactor.io/[Project Reactor] library as the underlying implementation, so it exposes its `Mono` and `Flux` types accordingly.
34
+
As a rule of thumb, if the blocking API returns a type `T` the reactive counterpart returns `Mono<T>` if one (or no) results is expected or in some cases `Flux<T>` if there are more than one expected.
35
+
We *highly* recommend that you make yourself familar with the https://projectreactor.io/docs/core/release/reference/[reactor documentation^] to understand its fundamentals and also unlock its full potential.
19
36
20
37
The following example fetches a document and prints out the `GetResult` once it has been loaded (or the exception if failed):
21
38
@@ -24,7 +41,8 @@ The following example fetches a document and prints out the `GetResult` once it
It is important to understand that reactive types are lazy, which means that they are only executed when a consumer subscribes to them. So a code like this won't even be executed at all:
44
+
It is important to understand that reactive types are lazy, which means that they are only executed when a consumer subscribes to them.
45
+
So a code like this won't even be executed at all:
28
46
29
47
[source,java]
30
48
----
@@ -42,10 +60,17 @@ You will come across the `Flux` type in APIs like query where there is one or mo
The `QueryResult` itself is wrapped in a `Mono`, but the class itself carries a `Flux<T>` of rows where `T` is a type of choice you can convert it to (in this example we simply convert it into `JsonObject`). The `flatMap` operator allows to map the stream or rows into the previous stream of the original result. If you have more question on how this works, check out the documentation https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Flux.html#flatMap-java.util.function.Function-[here].
63
+
The `QueryResult` itself is wrapped in a `Mono`, but the class itself carries a `Flux<T>` of rows where `T` is a type of choice you can convert it to (in this example we simply convert it into `JsonObject`).
64
+
The `flatMap` operator allows to map the stream or rows into the previous stream of the original result.
65
+
If you have more question on how this works, check out the documentation https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Flux.html#flatMap-java.util.function.Function-[here^].
66
+
46
67
47
68
== Low Level Asynchronous API with CompletableFutures
48
-
Both the blocking API and the reactive one are built on a lower level foundation using the https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html[CompletableFuture] type. It is built into the JDK starting from version 1.8 and while it is not as powerful as its reactive counterpart it does provide even better performance. In simplified terms, the `core-io` layer is responsible for mapping a `Request` to a `CompletableFuture<Response>`. The blocking API waits until the future completes on the caller thread while the reactive API wraps it into a `Mono`.
69
+
70
+
Both the blocking API and the reactive one are built on a lower level foundation using the https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html[CompletableFuture^] type.
71
+
It is built into the JDK starting from version 1.8 and while it is not as powerful as its reactive counterpart it does provide even better performance.
72
+
In simplified terms, the `core-io` layer is responsible for mapping a `Request` to a `CompletableFuture<Response>`.
73
+
The blocking API waits until the future completes on the caller thread while the reactive API wraps it into a `Mono`.
49
74
50
75
You can access this API by using the `async()` accessor methods both on the blocking and reactive counterparts:
51
76
@@ -54,21 +79,29 @@ You can access this API by using the `async()` accessor methods both on the bloc
We recommend using this API only if you are either writing integration code for higher level concurrency mechanisms or you really need the last drop of performance. In all other cases, the blocking API (for simplicity) or the reactive API (for richness in operators) is likely the better choice.
82
+
We recommend using this API only if you are either writing integration code for higher level concurrency mechanisms or you really need the last drop of performance.
83
+
In all other cases, the blocking API (for simplicity) or the reactive API (for richness in operators) is likely the better choice.
84
+
58
85
59
86
== Batching
87
+
60
88
The SDK itself does not provide explicit APIs for batching, because using the reactive mechanisms it allows you to build batching code applied to your use case much better than a generic implementation could in the first place.
61
89
62
-
While it can be done with the async API as well, we recommend using the reactive API so you can use async retry and fallback mechanisms that are supplied out of the box. The most simplistic bulk fetch (without error handling or anything) looks like this:
90
+
While it can be done with the async API as well, we recommend using the reactive API so you can use async retry and fallback mechanisms that are supplied out of the box.
91
+
The most simplistic bulk fetch (without error handling or anything) looks like this:
This code grabs a list of keys to fetch and passes them to `ReactiveCollection#get(String)`. Since this is happening asynchronously, the results will return in whatever order they come back from the server cluster. The `block()` at the end waits until all results have been collected. Of course the blocking part at the end is optional, but it shows that you can mix and match reactive and blocking code to on the one hand benefit from simplicity, but always go one layer below for the more powerful concepts if needed.
98
+
This code grabs a list of keys to fetch and passes them to `ReactiveCollection#get(String)`.
99
+
Since this is happening asynchronously, the results will return in whatever order they come back from the server cluster.
100
+
The `block()` at the end waits until all results have been collected.
101
+
Of course the blocking part at the end is optional, but it shows that you can mix and match reactive and blocking code to on the one hand benefit from simplicity, but always go one layer below for the more powerful concepts if needed.
70
102
71
-
While being simple, the code as shown has one big downside: individual errors for each document will fail the whole stream (this is how the `Flux` semantics are specified). In some cases this might be what you want, but most of the time you either want to ignore individual failures or mark them as failed.
103
+
While being simple, the code as shown has one big downside: individual errors for each document will fail the whole stream (this is how the `Flux` semantics are specified).
104
+
In some cases this might be what you want, but most of the time you either want to ignore individual failures or mark them as failed.
72
105
73
106
Here is how you can ignore individual errors:
74
107
@@ -77,9 +110,12 @@ Here is how you can ignore individual errors:
The `.onErrorResume(e -> Mono.empty()))` returns an empty `Mono` regardless of the error. Since you have the exception in scope, you can also decide based on the actual error if you want to ignore it or propagate/fallback to a different reactive computation.
113
+
The `.onErrorResume(e -> Mono.empty()))` returns an empty `Mono` regardless of the error.
114
+
Since you have the exception in scope, you can also decide based on the actual error if you want to ignore it or propagate/fallback to a different reactive computation.
81
115
82
-
If you want to separate out failures from completions, one way would be to use side effects. This is not as clean as with pure functional programming but does the job as well. Make sure to use concurrent data structures for proper thread safety:
116
+
If you want to separate out failures from completions, one way would be to use side effects.
117
+
This is not as clean as with pure functional programming but does the job as well.
118
+
Make sure to use concurrent data structures for proper thread safety:
If the result succeeds the side-effect method `doOnNext` is used to store it into the `successfulResults` and if the operation fails we are utilizing the same operator as before (`onErrorResume`) to store it in the `erroredResults` map -- but then also to ignore it for the overall sequence.
90
126
91
-
Finally, it is also possible to retry individual failures before giving up. The built-in retry mechanisms help with this:
127
+
Finally, it is also possible to retry individual failures before giving up.
It is recommended to check out the `retry` and `retryBackoff` methods for their configuration options and overloads. Of course, all the operators shown here can be combined to achieve exactly the semantics you need. Finally, for even advanced retry policies you can utilize the retry functionality in the https://projectreactor.io/docs/extra/release/api/reactor/retry/Retry.html[reactor-extra] package.
135
+
It is recommended to check out the `retry` and `retryBackoff` methods for their configuration options and overloads.
136
+
Of course, all the operators shown here can be combined to achieve exactly the semantics you need.
137
+
Finally, for even advanced retry policies you can utilize the retry functionality in the https://projectreactor.io/docs/extra/release/api/reactor/retry/Retry.html[reactor-extra^] package.
138
+
99
139
100
140
== Reactive Streams Integration
101
141
102
-
https://www.reactive-streams.org/[Reactive Streams] is an initiative to provide a standard for asynchronous stream processing with non-blocking back pressure. The reactor library the SDK depends on has out-of-the-box support for this interoperability specification, so with minimal hurdles you can combine it with other reactive libraries. This is especially helpful if your application stack is built on https://github.com/ReactiveX/RxJava[RxJava].
142
+
https://www.reactive-streams.org/[Reactive Streams^] is an initiative to provide a standard for asynchronous stream processing with non-blocking back pressure.
143
+
The reactor library the SDK depends on has out-of-the-box support for this interoperability specification, so with minimal hurdles you can combine it with other reactive libraries.
144
+
This is especially helpful if your application stack is built on https://github.com/ReactiveX/RxJava[RxJava].
103
145
104
146
The easiest way you can do this is by including the https://projectreactor.io/docs/adapter/release/api/[Reactor Adapter] library:
105
147
@@ -117,11 +159,12 @@ The easiest way you can do this is by including the https://projectreactor.io/do
117
159
</dependency>
118
160
----
119
161
120
-
Then, you can use the various conversion methods to convert back and forth between the rx and reactor types. The following snippet takes a `Mono<GetResult>` from the SDK and converts it into the RxJava `Single<GetResult>` equivalent.
162
+
Then, you can use the various conversion methods to convert back and forth between the rx and reactor types.
163
+
The following snippet takes a `Mono<GetResult>` from the SDK and converts it into the RxJava `Single<GetResult>` equivalent.
The same strategy can be used to convert to https://akka.io/[Akka], but if you are working in the scala world we recommend using our first-class Scala SDK directly instead!
170
+
The same strategy can be used to convert to https://akka.io/[Akka^], but if you are working in the scala world we recommend using our first-class Scala SDK directly instead!
0 commit comments