内存溢出错误 RxAndroid + RxJava + Retrofit2

Question

我刚接触RxAndroid和RxJava。我正在使用RxJava + Retrofit2发送GET请求。使用以下代码进行大约1500个GET请求，结果出现了内存不足错误。 但是如果只使用retrofit而不使用RxAndroid，同样的代码可以正常工作。因此，我的结论是我在RxAndroid上做错了什么。请问我错过了什么？

代码示例：

Subject<Story> mStoryEmitter = PublishSubject.create();

private void getStory(int storyID) {
    HNApi.Factory.getInstance().getStory(storyID).subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(getStoryObserver());
}

mStoryListEmitter.subscribe(new Observer<List<Integer>>() {
    @Override
    public void onSubscribe(Disposable d) {}

    @Override
    public void onNext(List<Integer> value) {
        if(mRecyclerView != null) {
            mRecyclerView.setAdapter(null);
            if(mAdapter != null) {
                mAdapter.clear();
                mAdapter = null;
            }
        }
        mAdapter = new SimpleRecyclerViewAdapter();
        mRecyclerView.setAdapter(mAdapter);

        for(Integer storyID : value) {
            getStory(storyID);
        }
    }

    @Override
    public void onError(Throwable e) {}

    @Override
    public void onComplete() {}
});

private DisposableObserver<Story> getStoryObserver() {
    DisposableObserver<Story> observer = new DisposableObserver<Story>() {
        @Override
        public void onNext(Story value) {
            mStoryEmitter.onNext(value);
            dispose();
        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onComplete() {

        }
    };
    return observer;
}

错误：

    Throwing OutOfMemoryError "Could not allocate JNI Env"
java.lang.IllegalStateException: Fatal Exception thrown on Scheduler.
    at io.reactivex.android.schedulers.HandlerScheduler$ScheduledRunnable.run(HandlerScheduler.java:111)
    at android.os.Handler.handleCallback(Handler.java:739)
    at android.os.Handler.dispatchMessage(Handler.java:95)
    at android.os.Looper.loop(Looper.java:148)
    at android.app.ActivityThread.main(ActivityThread.java:5417)
    at java.lang.reflect.Method.invoke(Native Method)
    at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:726)
    at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:616)
 Caused by: java.lang.OutOfMemoryError: Could not allocate JNI Env
    at java.lang.Thread.nativeCreate(Native Method)
    at java.lang.Thread.start(Thread.java:1063)
    at java.util.concurrent.ThreadPoolExecutor.addWorker(ThreadPoolExecutor.java:921)
    at java.util.concurrent.ThreadPoolExecutor.ensurePrestart(ThreadPoolExecutor.java:1556)
    at java.util.concurrent.ScheduledThreadPoolExecutor.delayedExecute(ScheduledThreadPoolExecutor.java:310)
    at java.util.concurrent.ScheduledThreadPoolExecutor.schedule(ScheduledThreadPoolExecutor.java:543)
    at java.util.concurrent.ScheduledThreadPoolExecutor.submit(ScheduledThreadPoolExecutor.java:642)
    at io.reactivex.internal.schedulers.NewThreadWorker.scheduleActual(NewThreadWorker.java:120)
    at io.reactivex.internal.schedulers.IoScheduler$EventLoopWorker.schedule(IoScheduler.java:221)
    at io.reactivex.Scheduler.scheduleDirect(Scheduler.java:130)
    at io.reactivex.Scheduler.scheduleDirect(Scheduler.java:109)
 AppData::create pipe(2) failed: Too many open files
    at io.reactivex.internal.operators.observable.ObservableSubscribeOn.subscribeActual(ObservableSubscribeOn.java:36)
    at io.reactivex.Observable.subscribe(Observable.java:10514)
    at io.reactivex.internal.operators.observable.ObservableObserveOn.subscribeActual(ObservableObserveOn.java:44)
    at io.reactivex.Observable.subscribe(Observable.java:10514)
    at com.example.MainActivity.getStory(MainActivity.java:100)
    at com.example.MainActivity.access$300(MainActivity.java:25)
    at com.example.MainActivity$2.onNext(MainActivity.java:67)
    at com.example.MainActivity$2.onNext(MainActivity.java:49)
    at io.reactivex.subjects.PublishSubject$PublishDisposable.onNext(PublishSubject.java:263)
    at io.reactivex.subjects.PublishSubject.onNext(PublishSubject.java:182)
    at com.example.MainActivity$5.onNext(MainActivity.java:147)
    at com.example.MainActivity$5.onNext(MainActivity.java:138)
    at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.drainNormal(ObservableObserveOn.java:198)
    at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.run(ObservableObserveOn.java:250)
    at io.reactivex.android.schedulers.HandlerScheduler$ScheduledRunnable.run(HandlerScheduler.java:109)
    ... 7 more
 AppData::create pipe(2) failed: Too many open files
 FATAL EXCEPTION: main
 Process: com.example, PID: 15857
 java.lang.IllegalStateException: Fatal Exception thrown on Scheduler.
    at io.reactivex.android.schedulers.HandlerScheduler$ScheduledRunnable.run(HandlerScheduler.java:111)
    at android.os.Handler.handleCallback(Handler.java:739)
    at android.os.Handler.dispatchMessage(Handler.java:95)
    at android.os.Looper.loop(Looper.java:148)
    at android.app.ActivityThread.main(ActivityThread.java:5417)
    at java.lang.reflect.Method.invoke(Native Method)
    at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:726)
    at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:616)
 Caused by: java.lang.OutOfMemoryError: Could not allocate JNI Env
    at java.lang.Thread.nativeCreate(Native Method)
    at java.lang.Thread.start(Thread.java:1063)
    at java.util.concurrent.ThreadPoolExecutor.addWorker(ThreadPoolExecutor.java:921)
    at java.util.concurrent.ThreadPoolExecutor.ensurePrestart(ThreadPoolExecutor.java:1556)
    at java.util.concurrent.ScheduledThreadPoolExecutor.delayedExecute(ScheduledThreadPoolExecutor.java:310)
    at java.util.concurrent.ScheduledThreadPoolExecutor.schedule(ScheduledThreadPoolExecutor.java:543)
    at java.util.concurrent.ScheduledThreadPoolExecutor.submit(ScheduledThreadPoolExecutor.java:642)
    at io.reactivex.internal.schedulers.NewThreadWorker.scheduleActual(NewThreadWorker.java:120)
    at io.reactivex.internal.schedulers.IoScheduler$EventLoopWorker.schedule(IoScheduler.java:221)
    at io.reactivex.Scheduler.scheduleDirect(Scheduler.java:130)
    at io.reactivex.Scheduler.scheduleDirect(Scheduler.java:109)
    at io.reactivex.internal.operators.observable.ObservableSubscribeOn.subscribeActual(ObservableSubscribeOn.java:36)
    at io.reactivex.Observable.subscribe(Observable.java:10514)
    at io.reactivex.internal.operators.observable.ObservableObserveOn.subscribeActual(ObservableObserveOn.java:44)
    at io.reactivex.Observable.subscribe(Observable.java:10514)
    at com.example.MainActivity.getStory(MainActivity.java:100)
    at com.example.MainActivity.access$300(MainActivity.java:25)
    at com.example.MainActivity$2.onNext(MainActivity.java:67)
    at com.example.MainActivity$2.onNext(MainActivity.java:49)
    at io.reactivex.subjects.PublishSubject$PublishDisposable.onNext(PublishSubject.java:263)
    at io.reactivex.subjects.PublishSubject.onNext(PublishSubject.java:182)
    at com.example.MainActivity$5.onNext(MainActivity.java:147)
    at com.example.MainActivity$5.onNext(MainActivity.java:138)
    at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.drainNormal(ObservableObserveOn.java:198)
    at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.run(ObservableObserveOn.java:250)
    at io.reactivex.android.schedulers.HandlerScheduler$ScheduledRunnable.run(HandlerScheduler.java:109)
    ... 7 more

- Anil Maddala

---

看起来有一些文件句柄泄漏了：AppData::create pipe(2) failed: Too many open files。请确保您正在使用最新的 Retrofit，以防它泄漏网络句柄（不太可能），否则请检查是否在响应网络数据时存在未关闭的文件访问（例如将数据保存到磁盘）。 - akarnokd

Answer 1

在RxAndroid的github上发布了同样的问题。

JakeWharton回复说，问题在于Schedulers.io()使用没有限制的缓存线程池，因此尝试创建1500个线程。您应该考虑使用具有固定线程数限制的调度程序，或使用RxJava 2.x的parallel()操作将操作并行化到固定数量的工作线程中。

如果您使用原始的Retrofit，默认情况下它使用OkHttp的分发器，该分发器将线程限制为大约64个（每个主机最多5个）。这就是您没有看到它失败的原因。

如果在创建RxJava2CallAdapterFactory时使用createAsync()，它将创建完全异步的Observable实例，不需要subscribeOn，并且像Retrofit一样使用OkHttp的Dispatcher。然后，您只需要使用observeOn返回到主线程，就可以避免所有其他线程的创建。

- Anil Maddala

---

你好 Anil，我也遇到了同样的问题，你能帮我解决一下吗？你是如何创建 createAsyc() 的呢？ - Santhi Bharath

Answer 2

将所有的Disposables添加到CompositeDisposable中，并在每个周期中进行dispose

CompositeDisposable disposable = new CompositeDisposable();
mStoryListEmitter.subscribe(new Observer<List<Integer>>() {
    @Override
    public void onSubscribe(Disposable d) {
       disposable.add(d); // adding disposable            
    }

    @Override
    public void onNext(List<Integer> value) {
        if(mRecyclerView != null) {
            mRecyclerView.setAdapter(null);
            if(mAdapter != null) {
                mAdapter.clear();
                mAdapter = null;
            }
        }
        mAdapter = new SimpleRecyclerViewAdapter();
        mRecyclerView.setAdapter(mAdapter);

        for(Integer storyID : value) {
            getStory(storyID);
        }
    }

    @Override
    public void onError(Throwable e) {}

    @Override
    public void onComplete() {
        diposable.dispose(); // <--- Disposing on complete
    }
})

;

- Rajan Kali

---

我尝试了你的代码示例，但仍然遇到了内存不足错误。 - Anil Maddala