Rxjava2 + Retrofit2 + Android。进行数百次网络调用的最佳方式

Question

我有一个应用程序。我有一个大按钮，允许用户一次将所有数据同步到云端。允许他们再次发送所有数据的重新同步功能。 （300 多个条目）

我正在使用 RXjava2 和 retrofit2。我的单元测试只需要一个调用。但是我需要进行 N 个网络调用。

我想要避免的是让 observable 调用队列中的下一个项目。我正处于需要实现可运行文件的地步。我看过一些关于地图的信息，但我还没有看到有人将它用作队列。此外，我想避免一个项目失败，它会报告所有项目都失败，就像 Zip 功能一样。我应该只做跟踪队列的讨厌的管理器类吗？或者有没有更简洁的方式来发送数百件物品？

注意：解决方案不能依赖于 JAVA8/LAMBDAS。事实证明，这比合理的工作要多得多。

注意所有项目都是同一个对象。

    @Test
public void test_Upload() {
    TestSubscriber<Record> testSubscriber = new TestSubscriber<>();
    ClientSecureDataToolKit clientSecureDataToolKit = ClientSecureDataToolKit.getClientSecureDataKit();
    clientSecureDataToolKit.putUserDataToSDK(mPayloadSecureDataToolKit).subscribe(testSubscriber);

    testSubscriber.awaitTerminalEvent();
    testSubscriber.assertNoErrors();
    testSubscriber.assertValueCount(1);
    testSubscriber.assertCompleted();
}

我的帮手收集和发送我的所有物品

public class SecureDataToolKitHelper {
private final static String TAG = "SecureDataToolKitHelper";
private final static SimpleDateFormat timeStampSimpleDateFormat =
        new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");


public static void uploadAll(Context context, RuntimeExceptionDao<EventModel, UUID> eventDao) {
    List<EventModel> eventModels = eventDao.queryForAll();

    QueryBuilder<EventModel, UUID> eventsQuery = eventDao.queryBuilder();
    String[] columns = {...};

    eventsQuery.selectColumns(columns);

    try {
        List<EventModel> models;

        models = eventsQuery.orderBy("timeStamp", false).query();
        if (models == null || models.size() == 0) {
            return;
        }

        ArrayList<PayloadSecureDataToolKit> toSendList = new ArrayList<>();
        for (EventModel eventModel : models) {
            try {
                PayloadSecureDataToolKit payloadSecureDataToolKit = new PayloadSecureDataToolKit();

                if (eventModel != null) {


                  // map my items ... not shown

                    toSendList.add(payloadSecureDataToolKit);
                }
            } catch (Exception e) {
                Log.e(TAG, "Error adding payload! " + e + " ..... Skipping entry");
            }
        }

        doAllNetworkCalls(toSendList);

    } catch (SQLException e) {
        e.printStackTrace();
    }

}

我的改装用品

public class ClientSecureDataToolKit {

    private static ClientSecureDataToolKit mClientSecureDataToolKit;
    private static Retrofit mRetrofit;

    private ClientSecureDataToolKit(){
        mRetrofit = new Retrofit.Builder()
        .baseUrl(Utilities.getSecureDataToolkitURL())
        .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
        .addConverterFactory(GsonConverterFactory.create())
        .build();
    }

    public static ClientSecureDataToolKit getClientSecureDataKit(){
        if(mClientSecureDataToolKit == null){
            mClientSecureDataToolKit = new ClientSecureDataToolKit();
        }
        return mClientSecureDataToolKit;
    }

    public Observable<Record> putUserDataToSDK(PayloadSecureDataToolKit payloadSecureDataToolKit){
        InterfaceSecureDataToolKit interfaceSecureDataToolKit = mRetrofit.create(InterfaceSecureDataToolKit.class);
        Observable<Record> observable = interfaceSecureDataToolKit.putRecord(NetworkUtils.SECURE_DATA_TOOL_KIT_AUTH, payloadSecureDataToolKit);
        return observable;
    }

}

public interface InterfaceSecureDataToolKit {

@Headers({
        "Content-Type: application/json"
})

@POST("/api/create")
Observable<Record> putRecord(@Query("api_token") String api_token, @Body PayloadSecureDataToolKit payloadSecureDataToolKit);
 }

---

更新。我一直试图将此答案应用于运气不佳。今晚我已经筋疲力尽了。我正在尝试将其作为一个单元测试来实现，就像我对一个项目的原始调用所做的那样。看起来使用 lambda 可能有些问题..

public class RxJavaBatchTest {
    Context context;
    final static List<EventModel> models = new ArrayList<>();

    @Before
    public void before() throws Exception {
        context = new MockContext();
        EventModel eventModel = new EventModel();
        //manually set all my eventmodel data here.. not shown 

        eventModel.setSampleId("SAMPLE0");
        models.add(eventModel);
        eventModel.setSampleId("SAMPLE1");
        models.add(eventModel);
        eventModel.setSampleId("SAMPLE3");
        models.add(eventModel);


    }

    @Test
    public void testSetupData() {
        Assert.assertEquals(3, models.size());
    }

    @Test
    public void testBatchSDK_Upload() {


        Callable<List<EventModel> > callable = new Callable<List<EventModel> >() {

            @Override
            public List<EventModel> call() throws Exception {
                return models;
            }
        };

        Observable.fromCallable(callable)
                .flatMapIterable(models -> models)
                .flatMap(eventModel -> {
                    PayloadSecureDataToolKit payloadSecureDataToolKit = new PayloadSecureDataToolKit(eventModel);
                    return doNetworkCall(payloadSecureDataToolKit) // I assume this is just my normal network call.. I am getting incompatibility errors when I apply a testsubscriber...
                            .subscribeOn(Schedulers.io());
                }, true, 1);
    }

    private Observable<Record> doNetworkCall(PayloadSecureDataToolKit payloadSecureDataToolKit) {

        ClientSecureDataToolKit clientSecureDataToolKit = ClientSecureDataToolKit.getClientSecureDataKit();
        Observable observable = clientSecureDataToolKit.putUserDataToSDK(payloadSecureDataToolKit);//.subscribe((Observer<? super Record>) testSubscriber);
        return observable;
    }

结果是..

An exception has occurred in the compiler (1.8.0_112-release). Please file a bug against the Java compiler via the Java bug reporting page (http://bugreport.java.com) after checking the Bug Database (http://bugs.java.com) for duplicates. Include your program and the following diagnostic in your report. Thank you.
com.sun.tools.javac.code.Symbol$CompletionFailure: class file for java.lang.invoke.MethodType not found


FAILURE: Build failed with an exception.

* What went wrong:
Execution failed for task ':app:compile<MyBuildFlavorhere>UnitTestJavaWithJavac'.
> Compilation failed; see the compiler error output for details.

---

编辑。不再尝试 Lambda。即使在我的 mac 上设置了路径，javahome 指向 1.8 等，我也无法让它工作。如果这是一个较新的项目，我会更加努力。然而，由于这是一个由 web 开发人员尝试使用 android 编写的继承的 android 应用程序，所以它不是一个很好的选择。也不值得花时间让它工作。已经进入了这项任务的日子，而不是应该花费的半天。

我找不到一个好的非 lambda 平面图示例。我自己试过了，结果很乱。

Answer 1

如果我的理解正确，您想同时拨打电话吗？

所以 rx-y 的做法是这样的：

    Observable.fromCallable(() -> eventsQuery.orderBy("timeStamp", false).query())
            .flatMapIterable(models -> models)
            .flatMap(model -> {
                // map your model

                //avoid throwing exceptions in a chain, just return Observable.error(e) if you really need to
                //try to wrap your methods that throw exceptions in an Observable via Observable.fromCallable()


                return doNetworkCall(someParameter)
                        .subscribeOn(Schedulers.io());
            }, true /*because you don't want to terminate a stream if error occurs*/, maxConcurrent /* specify number of concurrent calls, typically available processors + 1 */)
            .subscribe(result -> {/* handle result */}, error -> {/* handle error */});

在您的ClientSecureDataToolKit 中将此部分移到构造函数中

    InterfaceSecureDataToolKit interfaceSecureDataToolKit = mRetrofit.create(InterfaceSecureDataToolKit.class);