All generic questions around Reactor. For advanced questions you can also try #reactor-core and #reactor-netty
I am trying to read all accounts from a mongo collection (a LOT like 1Million and this is mandatory)
than i need to do some I/O bound operations on each account and than Update them in the mongoCollection again.
Can someone help me with the best approach (maybe some demo code impl) to do this? Im struggling with the performance.
Also we might want to scale verically, so go from 32Cpu cores to 64. I assume since we have I/O bound proccessing scaling vertically will not help much or am i wrong? Is there also some other impl to take also this scaling in consideration? Thanks a lot in advance
Hi,
I am struggling to understand the parallel behavior.
@Test
public void test() throws InterruptedException {
Flux.range(1, 1_000)
.subscribeOn(Schedulers.parallel())
.subscribe((a) -> {
System.out.println(Thread.currentThread().getName() + " " + a);
});
Thread.sleep(10_000_000);
}Output
parallel-1 1
parallel-1 2
parallel-1 3
...
parallel-1 998
parallel-1 999
parallel-1 1000I was expecting the subscribe function to execute in many threads depending on my CPU.
Instead they all execute on parallel-1
Doing some wrong I suspect?
Thanks
1 reply
Hi, what is the recommended concurrency in the case of flatMap? I bumped to this question which says the default value is 256
https://stackoverflow.com/questions/62558749/project-reactor-backpressure-issue
My question is this on which scenarios we need to explicitly handle concurrency in flatMap
13 replies
Hi, what is the recommended concurrency in the case of flatMap? I bumped to this question which says the default value is 256
https://stackoverflow.com/questions/62558749/project-reactor-backpressure-issueMy question is this on which scenarios we need to explicitly handle concurrency in flatMap
@OlegDokuka any suggestions?
senderOptions = senderOptions
.producerProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "SampleTxn");
KafkaSender.create(senderOptions)
.sendTransactionally(source.map(r -> Flux.fromIterable(transform(r))))
.concatMap(r -> r)
.doOnError(e-> log.error("Send failed, terminating.", e))
.doOnNext(r -> log.debug("Send completed {}", r.correlationMetadata());
Bumping my question
Hi its me again, Sorry to bother everyone
I am trying to read all accounts from a mongo collection (a LOT like 1Million and this is mandatory)
than i need to do some I/O bound operations on each account (sum all values inside a big array, some other arithmetic operations like * multiplication and Math.pow) and than Update them in the mongoCollection again.
Can someone help me with the best approach (maybe some demo code impl) to do this? Im struggling with the performance.
Also we might want to scale verically, so go from 32Cpu cores to 64. I assume since we have I/O bound proccessing scaling vertically will not help much or am i wrong? Is there also some other impl to take also this scaling in consideration? Thanks a lot in advance
hi there. I'd appreciate some advice: I have a flux of timestamped events (from kafka), and I need to collate them into event-time windows, with a configured accumulation timeout. the events aren't strictly ordered, but I can tolerate windows closing prematurely: stragglers that arrive after their previously-open window has closed can just be allocated into a new, separate window for the same time period.
I seek an appropriate strategy for achieving this without leaking resources or degrading performance as the cardinality of event-time windows grows without bound.
at first glance, something like the following seems like it might work:
events.groupBy(event -> event.timestamp().truncatedTo(windowDuration))
.flatMap(group -> group.window(windowAccumulationTimeout))however, I'm concerned about the implications of groupBy for that unbounded stream of groups. I suspect this might be problematic because each group will remain in the the groupBy state forever, even after the window-timeout expires for the last straggling event in each window.
am I wrong to be concerned about this? is there a better way to do what I need?
7 replies
Hey guys
I trust you all are thriving.
For reusability concerns, I run my validation on the service layer, which returns Mono.error( constraintViolationException )...
So that my web handlers merely forward the unmarshalled domain to the service layer.
So far, so great.
But how do I advise (AOP) my web handlers so that it returns HTTP 422 with the formatted constraint violations ?
WebExchangeBindException only handle exceptions thrown synchronously.
My AOP advice trigger and error b/c :
- my web handler return
Mono<DataType> - but my advice return a
ResponseEntity
And if I wrap my response entity (from the advice) into a Mono<ResponseEntity>, I an HTTP 200 OK with the response entity serialized :(
What is the equivalent operator/API to do the same work using Reactor?
Hi there!
I have a question. Why it's recommended to run blocking code out of Mono.defer() or mono.flatMap() in Mono.fromRunnable() or Mono.fromCallable()?
I mean, if I look at sources of Mono.fromRunnable(), actually it does like the same as Mono.defer():
public void subscribe(CoreSubscriber<? super T> actual) {
MonoRunnableEagerSubscription s = new MonoRunnableEagerSubscription();
actual.onSubscribe(s);
if (s.isCancelled()) {
return;
}
try {
run.run();
actual.onComplete();
} catch (Throwable ex) {
actual.onError(Operators.onOperatorError(ex, actual.currentContext()));
}
}The blocking code runs in subscribe() method. Supplier in Mono.defer() does the same?
Does it matter if I switch thread through .subscribeOn() operator?
Also MonoSubscribeOn cause me for thought. It does worker.schedule(() -> s.request(n)); on parent, when subscribe happens. But I can't find what happens, why request() scheduled on worker, not subscribe()?
Thanks
Hi all,
I have function that deletes user both from database and keyclaock. Sometimes it happen that user gets deleted from database but remains in Keyclaock. Dose anyone have an idea what could be a problem? Only happens from time to time but do not know why.
return _companyRepository.findById(companyId))
.flatMap(companyDocument -> {
final List<Employees> employees = companyDocument.getEmployees();
if (employees != null && !employees.isEmpty())
{
final boolean isRemoved =
employees.removeIf(employees -> employees.getId().equals(employeeId));
if (isRemoved)
{
companyDocument.setEmployees(employees);
return _companyRepository.save(companyDocument).thenReturn(employeeId);
}
}
}
return Mono.just(employeeId);
})
.flatMap(employeeId -> _keycloakService.deleteUser(employeeRealm, employeeId))
.onErrorResume(e -> Mono.empty());Hi, I am using SSE in the Spring web flux application. I have created an SSE endpoint that gets data from Redis pub-sub and Redis-pub gets data from an External WebSocket connection.
My problem is that when there is no data from Websocket then as soon as I try to connect with my SSE endpoint the connection terminates due to the unavailability of data. My hunch is that I need to pass comments heartbeat if data is not coming. how can I do that?
@PostConstruct
private fun realtimeSinkInitializer() {
/**
* onBackpressureBuffer by default cancel the subscription when last client closes, we made auto cancel false
*/
realtimeSink = Sinks.many().multicast().onBackpressureBuffer(1000, false)
reactiveMsgListenerContainer.receive(weatherTopic)
.onBackpressureBuffer(1000)
.map(ReactiveSubscription.Message<String, String>::getMessage)
.map { msg ->
objectMapper.deserialize<List<WeatherDto>>(msg)
}
.subscribe(
{ weatherData: List<WeatherDto> ->
realtimeSink.tryEmitNext(weatherData)
},
{ err ->
logger.error(err) { "Error occurred redis subscription channel" }
}
)
}
fun getRealtimeWeather(): Flux<ServerSentEvent<Map<String, WeatherDto>>> {
return weatherSink.asFlux()
.map { e -> ServerSentEvent.builder(e).build() }.onErrorResume { e ->
logger.error(e) { "Error getting realtime weather data" }
Flux.error(InternalServerException())
}
}
@EventListener
public void handle(TheEvent event) {
reactiveEventService.handle(event).subscribe();
}Second...
Even if you do, it doesn't really make the whole thing reactor-compliant
TheEvent as they come in
handle( TheEvent ) method from your except pushes events and do not subscribe to it
```
public DiscountCommand create(Mono<String> insertId, DiscountCommand discountCommand) {
return insertId.map(item -> {
return new DiscountCommand(
item,
discountCommand.discountCode(),
discountCommand.discountType(),
);
}).block(Duration.of(1000, ChronoUnit.MILLIS));
}If the insertId is null, it throws an null pointer exception and application goes to infinite. How can I handle error
2./ You can use if/else OR Optional
Optional.ofNullable( insertId )
.map( __ -> insertId.map( item -> new /* ... */ ) )
.orElseGet( Mono::empty )
.block(/* duration */);Something like that (untested)
9 replies
```
Sinks.Many<String> inputSink = Sinks.many().multicast().onBackpressureBuffer();
public Mono<String> process(String uid) {
inputSink.emitNext(uid, (signalType, emitResult) -> {
return false;
});
return Mono.just(uid);
}
I am subscribing to inputSink and process item and emitting outputSink
Sinks.Many<String> outputSink = Sinks.many().unicast().onBackpressureBuffer();
public Flux<String> getProcessedItems() {
return outputSink.asFlux();
}
Hi folks. I'm trying to use Sinks.many().unicast().onBackpressureBuffer() for the first time and I have unexpected result. Here is how I use it
// on Thread B
flushingSink.tryEmit(...)
flushingSink.asFlux()
.subscribeOn(THREAD-POOL-A)
.map {
// doing work here
// executed on Thread B, instead of THREAD-POOL-A
}I see that flushingSink.tryEmit is blocking AND queued messages are processed on calling thread. I thought they would be queued asynchronously. What am I doing wrong ?
Thanks!!
3 replies
Does anybody know if passing Mono/Flux around in the project is right way of doing chaining? It seem like I'm losing performance on all these reactor wrappers. In profiler I see call-stack like this one https://snipboard.io/fXT6KW.jpg
I see calls as deep as 150 nested Reactor invocations.
1 reply
Basically I'm trying to do the same - I want to process some work in parallel, always preserving original order per group (even if group changes thread later).
6 replies
I want to read a message and then execute a http request. If http request 2хх, ack message, otherwise - nack. But the problem is
flatMap returns Mono<ResponseEntity> and I can’t pass to map original amp message.RabbitFlux.createReceiver().consumeManualAck(QUEUE_NAME, consumeOptions)
.flatMap(amqpMessage -> {
String url = new String(amqpMessage.getBody());
return webclient.get().uri(url).retrieve().toBodilessEntity();
})
.map(responseEntity -> {
if (responseEntity.getStatusCode().is2xxSuccessful()) {
// how to pass here amqpMessage?
amqpMessage.ack();
} else {
amqpMessage.nack();
}
return amqpMessage;
}).subscribe();
var sourceFlux= Flux.range(1,10);
var hotflux1=sourceFlux.publish().refCount(2);
var firstFlux=hotflux1.map(i->i*2).log("First");
//this is a io blocking call
var secondFlux=hotflux1
.publishOn(Schedulers.boundedElastic()).map(i->i*5).log("Second");
secondFlux.subscribe();
//
var firstMono=Flux.just(300).log("First Mono");
var concatedFlux=Flux.concat(firstFlux,firstMono);
concatedFlux.subscribe();
I am using an approach similar to this to lift a reactor context into ThreadLocal for use by a grpc client interceptor.
https://stackoverflow.com/questions/47501349/using-hooks-and-a-lift-to-push-context-into-threadlocal
I am getting some stale values due to some bugs in the way I am using it. My main concern with this approach is its dependence on the onNext call to lift a fresh context into the ThreadLocal. If used incorrectly - namely not doing a contextWrite on a stream, it can lead to leaked values from other threads.
One way I thought of to add more safety would be to clear the ThreadLocal in the onComplete or onError of the subscriber. From my initial inspection it appears this works because, so far, the onComplete and onError is invoked on the same thread as onNext.
Question: Is it guaranteed that onComplete and onError is called on the same thread as onNext for a given subscriber like this?
1 reply
public static Flux<ByteBuffer> chunker(Flux<ByteBuffer> data, int chunkSize) {
return Flux.defer(() -> {
AtomicLong countDown = new AtomicLong(chunkSize);
return data.bufferUntil(buf -> {
if (countDown.addAndGet(-buf.remaining()) <= 0L) {
countDown.set(chunkSize);
return true;
}
return false;
}).concatMap(bufList -> {
int size = bufList.stream().mapToInt(ByteBuffer::remaining).sum();
return Mono.just(bufList.stream().reduce(ByteBuffer.allocate(size), ByteBuffer::put));
});
});
}