（与java inorder一起提供的反应式处理和反应式反应器：MWIO.3）

Question

问题陈述：

在块中执行I/O。一旦有一个块可用，就开始处理，同时在后台读取更多的块（但提前读取的块不超过X个）。在接收块时并行处理它们。按照读取顺序，即按照正在读取的块的原始顺序，使用每个已处理的块

我所做的：

我已经建立了一个MWE类来模拟这种情况，它在一定程度上起作用：

• “预回迁”部分似乎不像我预期的那样工作，“生成器”模拟IO，在“处理”之前生成任意多个项目，需要更多元素，具体取决于我设置的时间延迟
• 最终消耗量不符合要求（预计，但我还不知道该怎么办）

伪Rx代码解释了我想要实现的目标：

Flux.fromFile(path, some-function-to-define-chunk)
   // done with Flux.generate in MWE below

 .prefetchOnIoThread(x-count: int)
   // at this point we try to maintain a buffer filled with x-count pre-read chunks

 .parallelMapOrdered(n-threads: int, limit-process-ahead: int)
   // n-threads: are constantly trying to drain the x-count buffer, doing some transformation
   // limit-process-ahead: as the operation results are needed in order, if we encounter an
   // input element that takes a while to process, we don't want the pipeline to run too far
   // ahead of this problematic element (to not overflow the buffers and use too much memory)

 .consume(TMapped v)

反应堆的当前尝试（MWE）：

依赖关系：实现'io。项目反应堆：反应堆堆芯：3.3.5。发布“

import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.ParallelFlux;
import reactor.core.scheduler.Schedulers;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.atomic.AtomicInteger;

public class Tmp {
  static final SimpleDateFormat fmt = new SimpleDateFormat("HH:mm:ss.SSS");
  static long millisRead = 1; // time taken to "read" a chunk
  static long millisProcess = 100; // time take to "process" a chunk

  public static void main(String[] args) {
    log("Before flux construct");

    // Step 1: Generate / IO
    Flux<Integer> f = Flux.generate( // imitate IO
        AtomicInteger::new,
        (atomicInteger, synchronousSink) -> {
          sleepQuietly(millisRead);
          Integer next = atomicInteger.getAndIncrement();
          if (next > 50) {
            synchronousSink.complete();
            log("Emitting complete");
          } else {
            log("Emitting next : %d", next);
            synchronousSink.next(next);
          }
          return atomicInteger;
        },
        atomicInteger -> log("State consumer called: pos=%s", atomicInteger.get()));

    f = f.publishOn(Schedulers.elastic());
    f = f.subscribeOn(Schedulers.elastic());

    ParallelFlux<Integer> pf = f.parallel(2, 2);
    pf = pf.runOn(Schedulers.elastic(), 2);


    // Step 2: transform in parallel
    pf = pf.map(i -> {           // imitate processing steps
      log("Processing begin: %d", i);
      sleepQuietly(millisProcess); // 10x the time it takes to create an input for this operation
      log("Processing done : %d", i);
      return 1000 + i;
    });

    // Step 3: use transformed data, preferably in order of generation
    Disposable sub = pf.sequential(3).subscribe(
        next -> log(String.format("Finally got: %d", next)),
        err -> err.printStackTrace(),
        () -> log("Complete!"));

    while (!sub.isDisposed()) {
      log("Waiting pipeline completion...");
      sleepQuietly(500);
    }

    log("Main done");
  }

  public static void log(String message) {
    Thread t = Thread.currentThread();
    Date d = new Date();
    System.out.printf("[%s] @ [%s]: %s\n", t.getName(), fmt.format(d), message);
  }

  public static void log(String format, Object... args) {
    log(String.format(format, args));
  }

  public static void sleepQuietly(long millis) {
    try {
      Thread.sleep(millis);
    } catch (InterruptedException e) {
      throw new IllegalStateException();
    }
  }
}

Answer 1

考虑到缺乏答案，我将发布我的想法

final int threads = 2;
final int prefetch = 3;

Flux<Integer> gen = Flux.generate(AtomicInteger::new, (ai, sink) -> {
  int i = ai.incrementAndGet();
  if (i > 10) {
    sink.complete();
  } else {
    sink.next(i);
  }
  return ai;
});

gen.parallel(threads, prefetch)             // switch to parallel processing after genrator
    .runOn(Schedulers.parallel(), prefetch) // if you don't do this, it won't run in parallel
    .map(i -> i + 1000)                     // this is done in parallel
    .ordered(Integer::compareTo)            // you can do just .sequential(), which is faster
    .subscribeOn(Schedulers.elastic())      // generator will run on this scheduler (once subscribed)
    .subscribe(i -> {
      System.out.printf("Transformed integer: " + i); // do something with generated and processed item
    });