Android中使用Retrofit和RxJava从JSON数组中一一获取JSON对象

我正在使用 Retrofit 来访问返回 json 数组的网络 api。我正在使用以下代码这样做-

Observable<MyJson[]> response = (Observable<MyJson[]>)mNetworkService.getReadyObserverable(mNetworkService.getNetworkServiceApi().getMyDataJsons(), MyJson.class, true, useCache);
        mAirlineSubscription = response.subscribe(new Observer<MyJson[]>() {
            @Override
            public void onCompleted() {
                Log.d(TAG, "getData completed..");
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError: "  + e.getLocalizedMessage());
            }

            @Override
            public void onNext(MyJson[] myJsonData) {
                //I want here json data one by one
            }

但我的问题是 json 数据数组被完全下载，然后才onNext被叫。在这里我想要onNext当 Retrofit 下载第一个 json 对象时应该被调用myData[]json 数组并继续，直到我的所有 json 对象myData[]得到下载。这样，我的 UI 看起来响应更快，用户交互也更好。

任何人都可以帮忙解决这个问题吗？

尽可能快地一一接收元素？
不想等待所有内容都下载完毕？

解决办法是：用注解声明retrofit接口的方法@Streaming并使用Observable<ResponseBody>作为返回值。然后，通过使用flatMap()， 转变ResponseBody到 POJO 系列（Observable<TYPE>).

Example:

1. 声明改造接口：

   public interface HugeJsonApi {
   
   String SERVICE_ENDPOINT = "https://raw.githubusercontent.com";
   
   @Streaming
   @GET("/zemirco/sf-city-lots-json/master/citylots.json")
   Observable<ResponseBody> get();
   }

2. 像这样使用它：

   public void playHugeJsonSample() {
   
       HugeJsonApi hugeJsonApi = RestUtils.createService(HugeJsonApi.class, HugeJsonApi.SERVICE_ENDPOINT);
   
       Handler handler = new Handler(Looper.getMainLooper());
       hugeJsonApi.get()
               .flatMap(responseBody -> convertObjectsStream(responseBody, gson, Feature.class))
               .subscribeOn(Schedulers.io())
               .observeOn(AndroidSchedulers.mainThread())
               .subscribe(new Subscriber<Feature>() {
   
                   @Override
                   public void onStart() {
                       super.onStart();
                       request(1);
                   }
   
                   @Override
                   public void onNext(Feature feature) {
                       Log.i(TAG, gson.toJson(feature));
                       counter[0]++;
                       request(1);
                   }
   
                   @Override
                   public void onCompleted() {
                       Log.i(TAG, "onCompleted() called. Fetched elements:" + counter[0]);
                   }
   
                   @Override
                   public void onError(Throwable e) {
                       Log.e(TAG, "something went wrong", e);
                   }
               });
   }
   
   @NonNull
   private static <TYPE> Observable<TYPE> convertObjectsStream(ResponseBody responseBody, Gson gson, Class<TYPE> clazz) {
       Type type = TypeToken.get(clazz).getType();
       return Observable.create(SyncOnSubscribe.<JsonReader, TYPE>createStateful(
               // initialize the reader
               () -> {
                   try {
                       JsonReader reader = gson.newJsonReader(responseBody.charStream());
                       reader.beginObject();
                       return reader;
                   } catch (IOException e) {
                       e.printStackTrace();
                       RxJavaHooks.onError(e);
                   }
                   return null;
               },
               // read elements one by one
               (reader, observer) -> {
   
                   if (reader == null) {
                       observer.onCompleted();
                       return null;
                   }
   
                   try {
                       if (reader.hasNext()) {
                           TYPE t = gson.fromJson(reader, type);
                           observer.onNext(t);
                       }
                       else {
                           observer.onCompleted();
                       }
   
                   } catch (IOException e) {
                       e.printStackTrace();
                       observer.onError(e);
                   }
   
                   return reader;
               },
               // close the reader
               reader -> {
                   if (reader != null) {
                       try {
                           reader.close();
                       } catch (Exception e) {
                           e.printStackTrace();
                           RxJavaHooks.onError(e);
                       }
                   }
               }
   
       ));
   }

这是可行的示例：

https://github.com/allco/RetrofitAndRxJava https://github.com/allco/RetrofitAndRxJava

它需要 180+Mb Json 并将其解析为真正的流。