使用 RxJava 将无限的分组事件流写入旋转文件

Question

我正在尝试实现以下行为：

• 定期轮询/生成事件流（持续时间短，例如 1 秒）
• 然后根据某些内部特征对事件进行分组。
• 每组事件立即写入匹配文件（这是我要保持的行为的关键）
• 预计文件将在后续事件中重复用于匹配组（具有相同的密钥），直到它们被密封/轮换
• 如果持续时间较长（例如 5 秒），文件将被密封/轮换，并在使用其他订阅者时采取行动

我编写了以下示例代码来实现上述行为：

    private static final Integer EVENTS = 3;
    private static final Long SHORTER = 1L;
    private static final Long LONGER = 5L;
    private static final Long SLEEP = 100000L;

    public static void main(final String[] args) throws Exception {

        val files = new DualHashBidiMap<Integer, File>();

        Observable.just(EVENTS)
                .flatMap(num -> Observable.fromIterable(ThreadLocalRandom.current().ints(num).boxed().collect(Collectors.toList())))
                .groupBy(num -> Math.abs(num % 2))
                .repeatWhen(completed -> completed.delay(SHORTER, TimeUnit.SECONDS))
                .map(group -> {
                    val file = files.computeIfAbsent(group.getKey(), Unchecked.function(key -> File.createTempFile(String.format("%03d-", key), ".txt")));
                    group.map(Object::toString).toList().subscribe(lines -> FileUtils.writeLines(file, StandardCharsets.UTF_8.name(), lines, true));
                    return file;
                })
                .buffer(LONGER, TimeUnit.SECONDS)
                .flatMap(Observable::fromIterable)
                .distinct(File::getName)
                .doOnNext(files::removeValue)
                .doOnNext(file -> System.out.println("File - '" + file + "', Lines - " + FileUtils.readLines(file, StandardCharsets.UTF_8)))
                .subscribe();
        Thread.sleep(SLEEP);
    }

虽然它按预期工作（暂时搁置地图访问的线程安全问题，我使用来自commons-collections4 的双向地图只是为了演示），我想知道是否有办法实现纯 RX 形式的上述功能，不依赖外部地图访问？

请注意，在创建组时立即写入文件至关重要，这意味着我们必须使文件在生成的事件组范围之外存在

提前致谢。

Answer 1

有趣的问题.. 我可能是错的，但我认为您无法避免在管道中的某个地方出现 Map 或 Files。

我认为我的解决方案可以进一步清理，但它似乎完成了以下工作：

• 无需双向映射
• 无需致电Map.remove(...)

我建议你将Files 的Map 写成一个独特的Observable，以较慢的间隔发出一个全新的Map：

    Observable<HashMap<Integer, File>> fileObservable = Observable.fromCallable(
                () -> new HashMap<Integer, File>() )
            .repeatWhen( completed -> completed.delay( LONGER, TimeUnit.SECONDS ));

那么在你的活动Observable，你可以调用.withLatestFrom( fileObservable, ( group, files ) -> {...} )（注意：这个块还是不完整的）：

    Observable.just( EVENTS )
        .flatMap( num -> Observable.fromIterable(
                ThreadLocalRandom.current().ints( num ).boxed().collect( Collectors.toList() )))
        .groupBy( num -> Math.abs( num % 2 ))
        .repeatWhen( completed -> completed.delay( SHORTER, TimeUnit.SECONDS ))
        .withLatestFrom( fileObservable, ( group, files ) -> {

            File file = files.computeIfAbsent(
                    group.getKey(),
                    Unchecked.function( key -> File.createTempFile( String.format( "%03d-", key ), ".txt" )));

            group.map( Object::toString ).toList()
                .subscribe( lines -> FileUtils.writeLines(file, StandardCharsets.UTF_8.name(), lines, true ));

            return files;
        } )

到目前为止一切顺利，您将在活动旁边获得最新的Files。接下来你必须处理Files。我认为您可以使用distinctUntilChanged() 做到这一点。它应该非常有效，因为它会在后台调用HashMap.equals(...)，并且Map 对象的身份大部分时间都不会改变。 HashMap.equals(...) 首先检查相同的身份。

由于此时您真的有兴趣处理上一个 发出的Files 集合而不是当前，因此您可以使用.scan(( prev, current ) -> {...} ) 运算符。有了它，这是上面完整的代码块：

    Observable.just( EVENTS )
        .flatMap( num -> Observable.fromIterable(
                ThreadLocalRandom.current().ints( num ).boxed().collect( Collectors.toList() )))
        .groupBy( num -> Math.abs( num % 2 ))
        .repeatWhen( completed -> completed.delay( SHORTER, TimeUnit.SECONDS ))
        .withLatestFrom( fileObservable, ( group, files ) -> {

            File file = files.computeIfAbsent(
                    group.getKey(),
                    Unchecked.function( key -> File.createTempFile( String.format( "%03d-", key ), ".txt" )));

            group.map( Object::toString ).toList()
                .subscribe( lines -> FileUtils.writeLines(file, StandardCharsets.UTF_8.name(), lines, true ));

            return files;
        } )
        .distinctUntilChanged()
        .scan(( prev, current ) -> {

            Observable.fromIterable( prev.entrySet() )
                .map( Entry::getValue )
                .subscribe( file -> System.out.println( "File - '" + file + "', Lines - " +
                                FileUtils.readLines( file, StandardCharsets.UTF_8 )));

            return current;
        } )
        .subscribe();

    Thread.sleep( SLEEP );

比原来的解决方案要长一点，但可能会解决几个问题。