3

在使用 RxJava 编写数据同步作业时，我发现了一个我无法解释的奇怪行为。我对 RxJava 很陌生，希望能得到帮助。

简而言之，我的工作非常简单，我有一个元素 ID 列表，我调用一个 Web 服务来按 ID 获取每个元素，进行一些处理并进行多次调用以将数据推送到数据库。数据加载比数据存储快，所以我遇到了 OutOfMemory 错误。

我的代码看起来很像“失败”测试，但随后做了一些测试，我意识到删除该行：

flatMap(dt -> Observable.just(dt))

让它起作用。失败的测试输出清楚地表明未消耗的项目堆积起来，这会导致 OutOfMemory。工作测试输出显示生产者将始终等待消费者，因此这永远不会导致 OutOfMemory。

public static class DataStore {
    public Integer myVal;
    public byte[] myBigData;

    public DataStore(Integer myVal) {
        this.myVal = myVal;
        this.myBigData = new byte[1000000];
    }
}

@Test
public void working() {
    int MAX_CONCURRENT_LOAD = 1;
    int MAX_CONCURRENT_STORE = 2;

    AtomicInteger nbUnconsumed = new AtomicInteger(0);

    List<Integer> ids = IntStream.range(0, 1000).boxed().collect(Collectors.toList());
    Observable.from(ids)
        .flatMap(this::produce, MAX_CONCURRENT_LOAD)
        .doOnNext(s -> logger.info("+1 Total unconsumed values: " + nbUnconsumed.incrementAndGet()))
        .flatMap(this::consume, MAX_CONCURRENT_STORE)
        .doOnNext(s -> logger.info("-1 Total unconsumed values: " + nbUnconsumed.decrementAndGet()))
        .toBlocking().forEach(s -> {});

    logger.info("Finished");
}

@Test
public void failing() {
    int MAX_CONCURRENT_LOAD = 1;
    int MAX_CONCURRENT_STORE = 2;

    AtomicInteger nbUnconsumed = new AtomicInteger(0);

    List<Integer> ids = IntStream.range(0, 1000).boxed().collect(Collectors.toList());
    Observable.from(ids)
        .flatMap(this::produce, MAX_CONCURRENT_LOAD)
        .doOnNext(s -> logger.info("+1 Total unconsumed values: " + nbUnconsumed.incrementAndGet()))
        .flatMap(dt -> Observable.just(dt))
        .flatMap(this::consume, MAX_CONCURRENT_STORE)
        .doOnNext(s -> logger.info("-1 Total unconsumed values: " + nbUnconsumed.decrementAndGet()))
        .toBlocking().forEach(s -> {});

    logger.info("Finished");
}

private Observable<DataStore> produce(final int value) {
    return Observable.<DataStore>create(s -> {
        try {
            if (!s.isUnsubscribed()) {
                Thread.sleep(200); //Here I synchronous call WS to retrieve data
                s.onNext(new DataStore(value));
                s.onCompleted();
            }
        } catch (Exception e) {
            s.onError(e);
        }
    }).subscribeOn(Schedulers.io());
}

private Observable<Boolean> consume(DataStore value) {
    return Observable.<Boolean>create(s -> {
        try {
            if (!s.isUnsubscribed()) {
                Thread.sleep(1000); //Here I synchronous call DB to store data
                s.onNext(true);
                s.onCompleted();
            }
        } catch (Exception e) {
            s.onNext(false);
            s.onCompleted();
        }
    }).subscribeOn(Schedulers.io());
}

这种行为背后的解释是什么？我如何在不删除 Observable.just(dt)) 的情况下解决我的失败测试，​​在我的真实情况下是 Observable.from(someListOfItme)

4

1 回答 1

5

flatMap默认情况下合并无限数量的源，并且通过应用没有 maxConcurrent 参数的特定 lambda，您基本上可以不受限制地现在可以全速运行的上游，压倒其他运营商的内部缓冲区。

于 2016-02-10T18:11:03.057 回答