时间戳

首页 > 解决方案 > RxJava 分组排序数据

java - RxJava 分组排序数据

问题描述

我有一个非常大的数据集合，原始数据已经按照关键的内容进行了排序。例如，我有一个 CSV 文件，第一列将成为分组的关键。

A,x,x,x
A,x,y,x
A,z,y,y
C,x,s,d
C,t,d,s
B,a,s,a
E,x,x,x
E,t,r,y

这些行被转换为对象并放入一个列表并使用 RxJava 进行流式传输Flowable。因为这个 CSV 会很大（大到会导致应用程序崩溃），有没有一种好方法可以将这些对象转换为地图条目，如下所示：

{ 'A': [[x,x,x],[x,y,x],[z,y,y]] }

标签： javarx-javareactive-programming

解决方案

我现在正在使用FlowableTransformers.partialCollect。一个例子是

Flowable.fromPublisher(FlowableTransformers.partialCollect(
        (Consumer<PartialCollectEmitter<LineData, Integer, 
        ListBuilder, ListDataModel>>) emitter -> {
            // Get or initialize collecting object
            ListBuilder lb = emitter.getAccumulator();
            if (lb == null) {
                lb = new ListBuilder();
                emitter.setAccumulator(lb);
            }

            if (emitter.demand() != 0) {
                boolean d = emitter.isComplete();
                if (emitter.size() != 0 && !d) {
                    LineData data = emitter.getItem(0);
                    emitter.dropItems(1);

                    // add returns the finished model if the prefix changes
                    ListDataModel model = lb.add(data);

                    if (model != null) {
                        emitter.next(model);
                    }
                } else if (d) {
                    if (!lb.isEmpty()) {
                        // clear returns the last model
                        emitter.next(lb.clear());
                    }
                    emitter.complete();
                    return;
                }
            }
            emitter.setIndex(0);
        }, Functions.emptyConsumer(), settings.getReadBufferSize() + 1).apply(
                Flowable.fromIterable(file.getFileNameList())
                        .concatMap(
                                fileName -> reader
                                        .getLineData(fileName)
                                        .buffer(settings.getReadBufferSize()))
                        .flatMap(Flowable::fromIterable)))

推荐阅读