java - Rxjava2 + Retrofit2 + 安卓。进行数百次网络调用的最佳方式

Question

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

我正在使用 RXjava2 和 retrofit2。我有我的单元测试与一个电话一起工作。但是我需要进行 N 次网络调用。

我想避免的是让 observable 调用队列中的下一个项目。我正处于需要实现我的可运行对象的地步。我已经看到了一些关于 map 的信息，但我还没有看到有人将它用作队列。我还想避免一个项目失败并报告所有项目都失败，就像 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();
    }

}

我的 Retrofit 东西

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.

---

编辑。不再尝试 Lambdas。即使在我的 mac 上设置了路径，javahome 指向 1.8 等。我无法让它工作。如果这是一个较新的项目，我会更加努力。然而，由于这是一个由尝试使用 android 的 Web 开发人员编写的继承 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);