我正在使用改造来访问返回 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 从 myData[] json 数组下载第一个 json 对象时被调用，并继续直到 myData[]下载。这样，我的 UI 在用户交互方面看起来响应速度更快，效果更好。

谁能帮忙解决这个问题？

最佳答案

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

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

示例:

1. 声明改造接口(interface):

   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

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

关于android - 使用 Retrofit 和 RxJava 从 Android 中的 JSON 数组中一一获取 JSON 对象，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/39330983/