Project Reactor: buffer with parallel execution

user_709289 0
Viewed 665

I need to copy date from one source (in parallel) to another with batches.

I did this:

 Flux.generate((SynchronousSink<String> sink) -> {
                    try {
                        String val = dataSource.getNextItem();
                        if (val == null) {
                            sink.complete();
                            return;
                        }
                        sink.next(val);

                    } catch (InterruptedException e) {
                        sink.error(e);
                    }
                })
                .parallel(4)
                .runOn(Schedulers.parallel())
                .doOnNext(dataTarget::write)
                .sequential()
                .blockLast();

class dataSource{
  public Item getNextItem(){ 
    //...
  }
}

class dataTarget{
  public void write(List<Item> items){ 
    //...
  }
}

It receives data in parallel, but writes one at a time.

I need to collect them in batches (like by 10 items) and then write the batch.

How can I do that?

UPDATE:

The main idea that the source is the messaging system (i.e. rabbitmq or nats) that's suitable to efficiently send messages one by one, but the target is the database which is more efficient on inserting a batch.

So the final result should be like — I receive messages in parallel until buffer is not filled up, then I write all the buffer into database by one shot.

It's easy to do in regular java, but in case of streams — I don't get how to do it. How to buffer the data and how to pause the reader till the writer is not ready to get next part.

project-reactor reactive-programming java
3 Answers

All you need is Flux#buffer(int maxSize) operator:

Flux.generate((SynchronousSink<String> sink) -> {
        try {
            String val = dataSource.getNextItem();
            if (val == null) {
                sink.complete();
                return;
            }
            sink.next(val);

        } catch (InterruptedException e) {
            sink.error(e);
        }
    })
    .buffer(10) //Flux<List<String>>
    .flatMap(dataTarget::write)
    .blockLast();

class DataTarget{
    public Mono<Void> write(List<String> items){
         return reactiveDbClient.insert(items);
    }
}

Here, buffer collects items into multiple List's of 10 items(batches). You do not need to use parallel scheduler. The flatmap will run these operations asynchronously. See Understanding Reactive’s .flatMap() Operator.

user_89092350

You need to do your heavy work in individual Publisher-s which will be materialized in flatMap() in parallel. Like this

Flux.generate((SynchronousSink<String> sink) -> {
    try {
        String val = dataSource.getNextItem();
        if (val == null) {
            sink.complete();
            return;
        }
        sink.next(val);

    } catch (InterruptedException e) {
        sink.error(e);
    }
})
.parallel(4)
.runOn(Schedulers.parallel())
.flatMap(item -> Mono.fromCallable(() -> dataTarget.write(item)))
.sequential()
.blockLast();
user_2268950

Best approach (from algorithmic view) is to have ringbuffer and use microbatching technique. Writes to ringbuffer is done from rabbitmq, one-by-one (or multiple in parallel). Reading thread (single only) would get all messages at once (presented at a time of batch start), insert them into database and do it again... All at once means single message (if there is only one), or bunch of them (if they have been accumulated while duration of last insert was long enough to).

This technique is used also in jdbc (if I remember correctly) and can be implemented easily using lmax disruptor library in java.

Sample project (using ractor /Flux/ and System.out.println) can be found on https://github.com/luvarqpp/reactorBatch

Core code:

    final Flux<String> stringFlux = Flux.interval(Duration.ofMillis(1)).map(x -> "Msg number " + x);

    final Flux<List<String>> stringFluxMicrobatched = stringFlux
            .bufferTimeout(100, Duration.ofNanos(1));

    stringFluxMicrobatched.subscribe(strings -> {
        // Batch insert into DB
        System.out.print("Inserting in batch " + strings.size() + " strings.");
        try {
            // Inserting into db is simulated by 10 to 40 ms sleep here...
            Thread.sleep(rnd.nextInt(30) + 10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(" ... Done");
    });

Please feel welcome to edit and improve this post with name of technique and references. This is community wiki...

user_111526830
Related