RxJava- Group、Emit 和 Zip 排序的“块”具有共同的属性？

Question

我和我的同事经常遇到挑战，我希望反应式编程能够解决它。不过，它可能需要我自己实现Operator 或Transformer。

我想获取任何Observable<T> 发出T 的项目，但我希望操作员将它们分组到T 的映射上，并将每个分组作为List<T> 发出，或者更好的是一些通用累加器，就像Java 8 流中的Collector。

但这是我认为groupBy() 无法做到的棘手部分。我想通过这个 Operator 获取两个 Observable，并假设发出的项目按该属性排序（传入的数据将从排序的 SQL 查询中发出并映射到 T 对象）。操作员将连续累积项目，直到属性更改，然后发出该组并继续下一个。这样我就可以从每个 Observable 中获取每组数据，压缩并处理这两个块，然后将它们扔掉并继续下一个。通过这种方式，我可以保持半缓冲状态并保持较低的内存使用率。

因此，如果我在 PARTITION_ID 上进行排序、分组和压缩，这在视觉上就是我想要完成的任务。

我这样做只是因为我可以有两个查询，每个查询都超过一百万条记录，并且我需要并排进行复杂的比较。我没有内存来一次从双方导入所有数据，但我可以将其范围缩小到每个排序的属性值并将其分成批次。每一批之后，GC 都会将其丢弃，操作员可以继续进行下一批。

这是我到目前为止的代码，但我有点不清楚如何进行，因为我不想在批处理完成之前发出任何东西。我该怎么做？

public final class  SortedPartitioner<T,P,C,R> implements Transformer<T,R> {

 private final Function<T,P> mappedPartitionProperty;
 private final Supplier<C> acculatorSupplier;
 private final BiConsumer<T,R> accumulator;
 private final Function<C,R> finalResult;


 private SortedPartitioner(Function<T, P> mappedPartitionProperty, Supplier<C> acculatorSupplier,
   BiConsumer<T, R> accumulator, Function<C, R> finalResult) {
  this.mappedPartitionProperty = mappedPartitionProperty;
  this.acculatorSupplier = acculatorSupplier;
  this.accumulator = accumulator;
  this.finalResult = finalResult;
 }
 public static <T,P,C,R> SortedPartitioner<T,P,C,R> of(
   Function<T,P> mappedPartitionProperty, 
   Supplier<C> accumulatorSupplier,
   BiConsumer<T,R> accumulator,
   Function<C,R> finalResult) {

  return new SortedPartitioner<>(mappedPartitionProperty, accumulatorSupplier, accumulator, finalResult);

 }
 @Override
 public Observable<R> call(Observable<T> t) {
  return null;
 }

}

Answer 1

另一个答案是使用 Maven Central 上的库并且更短。

将此依赖项添加到您的pom.xml。

<dependency>
    <groupId>com.github.davidmoten</groupId>
    <artifactId>rxjava-extras</artifactId>
    <version>0.5.13</version>
</dependency>

就具有相同partition_id 的项目分组而言：

import com.github.davidmoten.rx.Transformers;

Observable<List<Item>> grouped = items.compose(
    Transformers.toListWhile(
        (list, item) -> list.isEmpty() || list.get(0).partitionId == item.partitionId));

此方法的测试非常全面（另请参阅 Transformers.collectWhile 了解列表以外的数据结构），但您可以在 github 上自行查看源代码。

然后继续zip。

Answer 2

这是一个棘手的问题，但我也经常遇到。

诀窍是使用materialize、scan 和flatMap。 scan 累积具有相同 partitionId 和下一个不同值（如果存在）的值列表。 materialize 是必需的，因为我们需要知道源何时完成，以便我们可以发出剩余的不同值（如果存在）。 flatMap 获取列表和值，并在值存在时发出列表（我们刚刚切换到新的 partitionId）并在流完成时发出值（剩余的）。

下面是一个单元测试，它从列表1, 1, 2, 2, 2, 3 发出列表{1, 1}, {2, 2, 2}, {3}。

对于您的用例，您只需将此技术应用于两个源并将它们压缩在一起。

代码：

import static org.junit.Assert.assertEquals;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Optional;

import org.junit.Test;

import rx.Observable;

public class StateMachineExampleTest {

    @Test
    public void testForStackOverflow() {
        Observable<Integer> a = Observable.just(1, 1, 2, 2, 2, 3);
        State<Integer> initial = new State<Integer>(Collections.emptyList(), Optional.empty(),
                false);
        List<List<Integer>> lists = a.materialize()
                // accumulate lists and uses onCompleted notification to emit
                // left overs when source completes
                .scan(initial,
                        (state, notification) -> {
                            if (notification.isOnCompleted()) {
                                return new State<>(null, state.value, true);
                            } else if (notification.isOnError())
                                throw new RuntimeException(notification.getThrowable());
                            else if (state.list.size() == 0) {
                                return new State<>(Arrays.asList(notification.getValue()), Optional
                                        .empty(), false);
                            } else if (partitionId(notification.getValue()) == partitionId(state.list
                                    .get(0))) {
                                List<Integer> list = new ArrayList<>();
                                list.addAll(state.list);
                                list.add(notification.getValue());
                                return new State<>(list, Optional.empty(), false);
                            } else if (state.value.isPresent()) {
                                if (partitionId(state.value.get()) == partitionId(notification
                                        .getValue())) {
                                    return new State<>(Arrays.asList(state.value.get(),
                                            notification.getValue()), Optional.empty(), false);
                                } else {
                                    return new State<>(Arrays.asList(state.value.get()), Optional
                                            .of(notification.getValue()), false);
                                }
                            } else {
                                return new State<>(state.list,
                                        Optional.of(notification.getValue()), false);
                            }
                        })
                // emit lists from state
                .flatMap(state -> {
                    if (state.completed) {
                        if (state.value.isPresent())
                            return Observable.just(Arrays.asList(state.value.get()));
                        else
                            return Observable.empty();
                    } else if (state.value.isPresent()) {
                        return Observable.just(state.list);
                    } else {
                        return Observable.empty();
                    }
                })
                // get as a list of lists to check
                .toList().toBlocking().single();
        assertEquals(Arrays.asList(Arrays.asList(1, 1), Arrays.asList(2, 2, 2), Arrays.asList(3)),
                lists);
    }

    private static int partitionId(Integer n) {
        return n;
    }

    private static final class State<T> {
        final List<T> list;
        final Optional<T> value;
        final boolean completed;

        State(List<T> list, Optional<T> value, boolean completed) {
            this.list = list;
            this.value = value;
            this.completed = completed;
        }
    }

}

请记住，此代码被快速破解，可能有漏洞。请务必使用您改编的此代码进行完整的单元测试。

需要额外注意的是，由于我们使用支持运算符materialize、scan 和flatMap 的背压，因此生成的转换也支持背压，因此可以安全地与zip 结合使用。