如何在Java中的反应堆通量中缓冲和组元素

Question

给定每个对象都有ID的无限通量，我如何使用磁通量来创建由ID属性分组的更新的缓冲列表（保持最后一个发射值）？ 感谢

示例

    Obj(ID=A, V=1)
    Obj(ID=A, V=2)
    Obj(ID=B, V=3) 
    --- buffer -> I want to subscribe with a list of [Obj(ID=A, V=2), Obj(ID=B, V=3)]
    Obj(ID=A, V=1)
    Obj(ID=B, V=4)
    Obj(ID=B, V=6)
    Obj(ID=A, V=2)
    --- buffer -> I want to subscribe with a list of [Obj(ID=B, V=6), Obj(ID=A, V=2)]
    Obj(ID=B, V=1)
    --- buffer -> I want to subscribe with a list of [Obj(ID=B, V=1)]

类似以下内容是完美的，但似乎等待我的测试中的通量结束而不是缓冲。

flux
    .buffer(Duration.ofMillis(2000))
    .groupBy(Obj::getId)
    .flatMap(GroupedFlux::getLast)
    .collectToList()
    .subscribe(this::printList);

它可以与缓冲区和自定义逻辑进行分组

    public static void main(String[] args) {
        flux.buffer(Duration.ofMillis(2000)).subscribe(this::groupList);
    }

    private void groupList(List<T> ts) {
        Collection<T> values = ts.stream()
                .collect(Collectors.toMap(T::getId, Function.identity(), (k, v) -> v))
                .values();
        System.out.println(values);
    }

Answer 1

缓冲区将发射list＆lt; t＆gt;，因此您可以使用非反应性Java进行分组。例如，Java流像您的示例一样。假设您的过程功能是反应性的，您可以继续流动

flux
    .buffer(Duration.ofMillis(2000))
    .map(list -> list.stream().collect(Collectors.toMap(Entry::getId, Function.identity(), (k, v) -> v)))
    .flatMapIterable(Map::values)
    .flatMap(obj -> process(obj));

Answer 2

具有类似的用例。在AWS S3上拥有模拟数据。数据以JSON记录存储在512的块中。需要下载可变数量的这些块并将其转换为单个JSON数组。变压器信号表明它不再需要通过返回转换记录的数量来进行记录 - 当不需要进一步的记录时，这将为零。以下是两个代码片段。第一个使用 flux.window ，而后者使用 flux.buffer 。后者的执行时间约为前者的速度十分之一，但下载的数据比绝对必要的数据要多。

flux.window 代码

    Flux<Flux<S3Object>> s3Objects = Flux.fromIterable(filteredList).window(windowSize);
    Integer cumulativeSize = s3Objects.concatMap(innerS3Objects -> {
        Flux <File> downloadedFiles = innerS3Objects.publishOn(Schedulers.boundedElastic()).concatMap(s3Object -> downloadS3Object(BUCKET_NAME, s3Object.key()));
        return downloadedFiles.publishOn(Schedulers.single()).concatMap(downloadedFile -> {
            return transformMockarooData(downloadedFile, generator, targetHourCount);
        }).takeUntil(transformedRecordCount -> (transformedRecordCount == 0)).reduce((i,j) -> i + j);               
    }).takeUntil(transformedRecordCount -> (transformedRecordCount == 0)).reduce((i,j) -> i + j).block();

flux.buffer 代码

    Flux<List<S3Object>> s3Objects = Flux.fromIterable(filteredList).buffer(windowSize);
    Integer cumulativeSize = s3Objects.concatMap(innerS3Objects -> {
        List <Mono<File>> toDownloadFiles = innerS3Objects.stream().map(s3Object -> downloadS3Object(BUCKET_NAME, s3Object.key())).collect(Collectors.toList());
        Flux <File> downloadedFiles = Flux.merge(toDownloadFiles);
        return downloadedFiles.publishOn(Schedulers.single()).concatMap(downloadedFile -> {
            return transformMockarooData(downloadedFile, generator, targetHourCount);
        }).takeUntil(transformedRecordCount -> (transformedRecordCount == 0)).reduce((i,j) -> i + j);               
    }).takeUntil(transformedRecordCount -> (transformedRecordCount == 0)).reduce((i,j) -> i + j).block();

Answer 3

我能够通过反应性分组来实现它

flux.window(Duration.ofMillis(2000))
    .flatMap(window -> window.groupBy(Entry::getId)
        .flatMap(GroupedFlux::last)
        .collectList()
    )
    .subscribe(this::printList);