4 ms·

Have you got any links, or can you explain the optimisations that are possible in RxJava? I've done some vert.x and creating new Futures all the time is a bit

Have you got any links, or can you explain the optimisations that are possible in RxJava? I've done some vert.x and creating new Futures all the time is a bit cumbersome, but I understand it. With RxJava it's another layer on top, and you are even more removed. The whole non-blocking experience in java feels like there is a lot of plumbing. I know it's fast for IO reasons, but it really feels like the JVM/CPU has to jump through a lot of glue code to make it work.

The_rationalist 6y ago
can you explain the optimisations that are possible in RxJava? Here's the fundamental problem that reactive programming solve: Here's a kotlin function from https://kotlinlang.org/docs/reference/coroutines/flow.html#suspending-functions https://kotlinlang.org/docs/reference/coroutines/flow.html#s... suspend fun foo(): List<Int> { delay(1000) // pretend we are doing something asynchronous here return listOf(1, 2, 3) } This function is not reactive but it is at least async. suspend ~= async keyword in js or C# As you can see its return type is not colored: List<Int> In inferior languages you would have a Promise<List<Int>> and need to check yourself for its resolution before accessing the value. (Promise == CompletableFuture) But this is not what matters here This example function is will be executed asynchronously, it will do processing on a list then return the list. It means that it will need to wait for all elements of the list to have been processed each before being able to use the result. The big idea is that with reactive streams we can process list elements asynchronously, individually https://kotlinlang.org/docs/reference/coroutines/flow.html#flows https://kotlinlang.org/docs/reference/coroutines/flow.html#f... fun foo(): Flow<Int> = flow { // flow builder for (i in 1..3) { delay(100) // pretend we are doing something useful here emit(i) // emit next value } } A flow is just an asynchronous list, an array of CompletableFutures if you want. Each time you emit() a value from the flow, the processed element will be received at call site, on demand Beyond this base idea, reactive programming add on to of it many useful concepts like: Cancellation primitive, automatic multithreaded dispatching, lazy or eager evaluation, many things and the ability to choose a specific scheduler (a blocking io friendly one for example) Reactive streams are fantastic ideas but API wise even with reactor it needed a lot of boilerplate and had incidental complexity. Kotlin flow bring the best of both world and is to me the state of the art asynchronous programming model, transcendanting goroutines.
dima_vm 6y ago
I'm neutral on RxJava, but honestly don't know how to handle backpressure without it. Would be glad to learn.
- Ironlink 6y ago
  My naive response is that if the socket buffer you're trying to write to is full, there's backpressure. What am I missing here?
  - mac01021 6y ago
    It can be unacceptable sometimes and other times just wasteful to block a whole host OS thread just because a read or a write on one socket out of 1000s is not ready. We solve this problem with APIs for asynchronous or nonblocking IO. But such APIs must be cleverly designed if they are to permit you to propagate backpressure from the downstream end of your program's dataflow to the upstream end. And handle errors in a sane way, etc.
  - isbvhodnvemrwvn 6y ago
    You assume that the thread pool which takes requests is the very same one that gets bogged down. It might very well be something down the line in the processing pipeline, and unless you poll these stages or they signal their capacity through the pipeline, you can't know that. You're wasting resources when your application is already in a state where it knows it won't be able to handle the request. Eventually the memory taken by the partially processed requests is going to exceed what you can take in (unless you cap the number of concurrently processed requests, which is also an inelastic backpressure of sorts) and the service will crash. What you mentioned is decent for inelastic blocking synchronous processing (you can have at most X concurrent requests, because that's how many threads for processing you've configured based on performance tests and production monitoring), but you can relatively easy fill in an internal queue somewhere if it's async.
unscaled 6y ago
RxJava or Reactive Streams won't solve the issue of ceremony needed for dealing with async I/O. You're correct that it only makes it worse. The real advantage the reactive model offers is a more powerful abstraction for data flow than futures. I found it useful than straight coroutines or async/await in cases where I had to process complex data flows or combine periodically polled data from multiple sources, but in well over 99% of the use cases for an async REST API it's an overkill. Removing the ceremony around asynchronous I/O, on the other hand, requires language support, so your best option right now is to just use Kotlin coroutines. Or wait for Project Loom, but I assume it's at least 2 years away from a stable release.
- The_rationalist 6y ago
  Kotlin flow allows you to have the best of both worlds: Suspending function that have obscoleted the concept of promises and flow when you want to deal with fine grained async list processing

The_rationalist 6y ago

can you explain the optimisations that are possible in RxJava? Here's the fundamental problem that reactive programming solve: Here's a kotlin function from https://kotlinlang.org/docs/reference/coroutines/flow.html#suspending-functions https://kotlinlang.org/docs/reference/coroutines/flow.html#s... suspend fun foo(): List<Int> { delay(1000) // pretend we are doing something asynchronous here return listOf(1, 2, 3) } This function is not reactive but it is at least async. suspend ~= async keyword in js or C# As you can see its return type is not colored: List<Int> In inferior languages you would have a Promise<List<Int>> and need to check yourself for its resolution before accessing the value. (Promise == CompletableFuture) But this is not what matters here This example function is will be executed asynchronously, it will do processing on a list then return the list. It means that it will need to wait for all elements of the list to have been processed each before being able to use the result. The big idea is that with reactive streams we can process list elements asynchronously, individually https://kotlinlang.org/docs/reference/coroutines/flow.html#flows https://kotlinlang.org/docs/reference/coroutines/flow.html#f... fun foo(): Flow<Int> = flow { // flow builder for (i in 1..3) { delay(100) // pretend we are doing something useful here emit(i) // emit next value } } A flow is just an asynchronous list, an array of CompletableFutures if you want. Each time you emit() a value from the flow, the processed element will be received at call site, on demand Beyond this base idea, reactive programming add on to of it many useful concepts like: Cancellation primitive, automatic multithreaded dispatching, lazy or eager evaluation, many things and the ability to choose a specific scheduler (a blocking io friendly one for example) Reactive streams are fantastic ideas but API wise even with reactor it needed a lot of boilerplate and had incidental complexity. Kotlin flow bring the best of both world and is to me the state of the art asynchronous programming model, transcendanting goroutines.

dima_vm 6y ago

I'm neutral on RxJava, but honestly don't know how to handle backpressure without it. Would be glad to learn.

Ironlink 6y ago
My naive response is that if the socket buffer you're trying to write to is full, there's backpressure. What am I missing here?
- mac01021 6y ago
  It can be unacceptable sometimes and other times just wasteful to block a whole host OS thread just because a read or a write on one socket out of 1000s is not ready. We solve this problem with APIs for asynchronous or nonblocking IO. But such APIs must be cleverly designed if they are to permit you to propagate backpressure from the downstream end of your program's dataflow to the upstream end. And handle errors in a sane way, etc.
- isbvhodnvemrwvn 6y ago
  You assume that the thread pool which takes requests is the very same one that gets bogged down. It might very well be something down the line in the processing pipeline, and unless you poll these stages or they signal their capacity through the pipeline, you can't know that. You're wasting resources when your application is already in a state where it knows it won't be able to handle the request. Eventually the memory taken by the partially processed requests is going to exceed what you can take in (unless you cap the number of concurrently processed requests, which is also an inelastic backpressure of sorts) and the service will crash. What you mentioned is decent for inelastic blocking synchronous processing (you can have at most X concurrent requests, because that's how many threads for processing you've configured based on performance tests and production monitoring), but you can relatively easy fill in an internal queue somewhere if it's async.

unscaled 6y ago

RxJava or Reactive Streams won't solve the issue of ceremony needed for dealing with async I/O. You're correct that it only makes it worse. The real advantage the reactive model offers is a more powerful abstraction for data flow than futures. I found it useful than straight coroutines or async/await in cases where I had to process complex data flows or combine periodically polled data from multiple sources, but in well over 99% of the use cases for an async REST API it's an overkill. Removing the ceremony around asynchronous I/O, on the other hand, requires language support, so your best option right now is to just use Kotlin coroutines. Or wait for Project Loom, but I assume it's at least 2 years away from a stable release.

The_rationalist 6y ago
Kotlin flow allows you to have the best of both worlds: Suspending function that have obscoleted the concept of promises and flow when you want to deal with fine grained async list processing