时间戳

首页 > 解决方案 > 如何使用 RxJava 遍历列表并对第一项执行初始处理

rx-java2 - 如何使用 RxJava 遍历列表并对第一项执行初始处理

问题描述

我是 RxJava 的新手，发现它对我的 Android 应用程序中的网络和数据库处理非常有用。

我有两个用例无法在 RxJava 中完全实现

用例 1

  1. 清除我的目标数据库表 Table A
  2. 从表 B 中获取包含关键字段的数据库记录列表
  3. 对于从表 B 检索到的每一行，调用远程 API 并将所有返回的数据持久化到表 A

我管理的最接近的是这段代码

final AtomicInteger id = new AtomicInteger(0);

DatabaseController.deleteAll(TableA_DO.class);

DatabaseController.fetchTable_Bs()
        .subscribeOn(Schedulers.io())
        .toObservable()
        .flatMapIterable(b -> b)
        .flatMap(b_record -> NetworkController.getTable_A_data(b_record.getKey()))
        .flatMap(network -> transformNetwork(id, network, NETWORK_B_MAPPER))
        .doOnNext(DatabaseController::persistRealmObjects)
        .doOnComplete(onComplete)
        .doOnError(onError)
        .doAfterTerminate(doAfterTerminate())
        .doOnSubscribe(compositeDisposable::add)
        .subscribe();

用例 2

  1. 清除我的目标数据库表 Table X
  2. 清除我的目标数据库表表 Y
  3. 从表 Z 中获取包含关键字段的数据库记录列表
  4. 对于从表 B 检索到的每一行，调用远程 API 并将返回的一些数据持久化到表 X 中，其余数据应持久化到表 Y

我还没有设法为用例 2 创建任何代码。

关于在这些用例中使用 RxJava，我有很多问题。

  1. 是否可以在 RxJava 中实现我的两个用例？
  2. 将所有这些步骤组合成一个 Rx“流”是“最佳实践”吗

更新

我最终得到了这个 POC 测试代码，它似乎可以工作......我不确定它是否是最佳解决方案，但是我的 API 调用返回 Single 并且我的数据库操作返回 Completable 所以我觉得这对我来说是最好的解决方案。

public class UseCaseOneA {

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

        login()
        .andThen(UseCaseOneA.deleteDatabaseTableA())
        .andThen(UseCaseOneA.deleteDatabaseTableB())
        .andThen(manufactureRecords())
            .flatMapIterable(x -> x)
            .flatMapSingle(record -> NetworkController.callApi(record.getPrimaryKey()))
            .flatMapSingle(z -> transform(z))
            .flatMapCompletable(p -> UseCaseOneA.insertDatabaseTableA(p))
            .doOnComplete(() -> System.out.println("ON COMPLETE"))
            .doFinally(() -> System.out.println("ON FINALLY"))
            .subscribe();

    }

    private static Single<List<PayloadDO>> transform(final List<RemotePayload> payloads) {
        return Single.create(new SingleOnSubscribe<List<PayloadDO>>() {
            @Override
            public void subscribe(final SingleEmitter<List<PayloadDO>> emitter) throws Exception {
                System.out.println("transform - " + payloads.size());

                final List<PayloadDO> payloadDOs = new ArrayList<>();

                for (final RemotePayload remotePayload : payloads) {
                    payloadDOs.add(new PayloadDO(remotePayload.getPayload()));
                }

                emitter.onSuccess(payloadDOs);
            }
        });
    }

    private static Observable<List<Record>> manufactureRecords() {
        final List<Record> records = new ArrayList<>();
        records.add(new Record("111-111-111"));
        records.add(new Record("222-222-222"));
        records.add(new Record("3333-3333-3333"));
        records.add(new Record("44-444-44444-44-4"));
        records.add(new Record("5555-55-55-5-55-5555-5555"));

        return Observable.just(records);
    }

    private static Completable deleteDatabaseTableA() {

        return Completable.create(new CompletableOnSubscribe() {

            @Override
            public void subscribe(final CompletableEmitter emitter) throws Exception {
                System.out.println("deleteDatabaseTableA");

                emitter.onComplete();
            }
        });
    }

    private static Completable deleteDatabaseTableB() {

        return Completable.create(new CompletableOnSubscribe() {

            @Override
            public void subscribe(final CompletableEmitter emitter) throws Exception {
                System.out.println("deleteDatabaseTableB");

                emitter.onComplete();
            }
        });
    }

    private static Completable insertDatabaseTableA(final List<PayloadDO> payloadDOs) {
        return Completable.create(new CompletableOnSubscribe() {
            @Override
            public void subscribe(final CompletableEmitter emitter) throws Exception {
                System.out.println("insertDatabaseTableA - " + payloadDOs);

                emitter.onComplete();
            }
        });
    }

    private static Completable login() {
        return Completable.complete();
    }
}

这段代码不能满足我所有的用例要求。即能够将远程有效负载记录转换为多种数据库记录类型，并将每种类型插入到自己特定的目标数据库表中。

我可以调用 Remote API 两次来获取相同的远程数据项，然后先转换为一种数据库类型，然后再转换为第二种数据库类型，但这似乎很浪费。

RxJava 中是否有一个操作数，我可以重用我的 API 调用的输出并将它们转换为另一种数据库类型？

标签： rx-java2

解决方案

您必须以某种方式自己索引项目，例如，通过外部计数：

Observable.defer(() -> {
    AtomicInteger counter = new AtomicInteger();

    return DatabaseController.fetchTable_Bs()
        .subscribeOn(Schedulers.io())
        .toObservable()
        .flatMapIterable(b -> b)
        .doOnNext(item -> {
            if (counter.getAndIncrement() == 0) {
                // this is the very first item
            } else {
                // these are the subsequent items
            }
        });
});

有defer必要将计数器与内部序列隔离，以便在必要时重复仍然有效。

推荐阅读