Reactive Streams - 使用超时批处理

Question

我正在寻找替换本土的日志处理库,它看起来非常接近ReactiveStreams io.projectreactor.目标是减少我们维护的代码,并利用社区添加的任何新功能(瞄准操作员融合).

首先,我需要使用stdio并将多行日志条目合并到流向管道的文本blob中.在Filebeat文档的多行日志条目章节中详细解释了用例(除了我们希望它在进程中).

到目前为止,我的代码是:

BufferedReader input = new BufferedReader(new InputStreamReader(System.in));
Flux<String> lines = Flux.generate(sink -> rethrow(() -> { while (true) sink.next(input.readLine()); }));
Flux<String> logRecordsStr = lines.concatMap(new LogRecordJoiner());
Flux<LogRecord> logRecords = logRecordsStr.map(new LogRecordMapper());
logRecords.doOnEach(r -> System.out.printf("%s payload: %d chars\n", r.timestamp, r.payload.length()))
          .subscribe();          

这会在检测到新的日志标头时关注多行合并,但在现有库中,我们还会在超时后刷新累积的行(即如果在5秒内没有收到文本,则刷新记录).

在Reactor中对此进行建模的正确方法是什么？我是否需要编写自己的运算符,还是可以自定义任何现有运算符？

任何有关在Project Reactor或RxJava中实现此用例的相关示例和文档的指针都将非常受欢迎.

Answer 1

这取决于您如何识别每个缓冲区的开始和结束，因此以下 RxJava 2 代码旨在作为有关使用主源的值来打开和关闭缓冲区门的提示：

TestScheduler scheduler = new TestScheduler();
PublishProcessor<String> pp = PublishProcessor.create();

Function<Flowable<String>, Flowable<List<String>>> f = o -> 
        o.buffer(o.filter(v -> v.contains("Start")), 
                 v -> Flowable.merge(o.filter(w -> w.contains("End")), 
                                     Flowable.timer(5, TimeUnit.MINUTES, scheduler))); 

pp.publish(f)
.subscribe(System.out::println);

pp.onNext("Start");
pp.onNext("A");
pp.onNext("B");
pp.onNext("End");

pp.onNext("Start");
pp.onNext("C");

scheduler.advanceTimeBy(5, TimeUnit.MINUTES);

pp.onNext("Start");
pp.onNext("D");
pp.onNext("End");
pp.onComplete();

印刷：

[Start, A, B, End]
[Start, C]
[Start, D, End]

它的工作原理是共享源publish，允许重用来自上游的相同值，而无需同时运行多个源副本。开盘是通过检测线路上的“Start”字符串来控制的。关闭由“End”字符串的检测或宽限期后触发的计时器控制。

编辑：

如果“开始”也是下一批的指示符，则可以将“结束”检查替换为“开始”并更改缓冲区的内容，因为它将在前一个缓冲区中包含新标头，否则：

pp.publish(f)
.doOnNext(v -> {
    int s = v.size();
    if (s > 1 && v.get(s - 1).contains("Start")) {
        v.remove(s - 1);
    }
})
.subscribe(System.out::println);