Rxjava和Retrofit结合使用大量请求时候出现OOM的问题

在使用RxJava+Retrofit的过程中,出现了OOM的问题,报错日志如下:

java.lang.OutOfMemoryError: pthread_create (1040KB stack) failed: Try
again
at java.lang.Thread.nativeCreate(Native Method)
at java.lang.Thread.start(Thread.java:733)
at java.util.concurrent.ThreadPoolExecutor.addWorker(ThreadPoolExecutor.java:970)
at java.util.concurrent.ThreadPoolExecutor.ensurePrestart(ThreadPoolExecutor.java:1611)
at java.util.concurrent.ScheduledThreadPoolExecutor.delayedExecute(ScheduledThreadPoolExecutor.java:342)
at java.util.concurrent.ScheduledThreadPoolExecutor.schedule(ScheduledThreadPoolExecutor.java:579)
at java.util.concurrent.ScheduledThreadPoolExecutor.submit(ScheduledThreadPoolExecutor.java:680)
at io.reactivex.internal.schedulers.NewThreadWorker.scheduleActual(NewThreadWorker.java:145)
at io.reactivex.internal.schedulers.IoScheduler$EventLoopWorker.schedule(IoScheduler.java:239)
at io.reactivex.Scheduler.scheduleDirect(Scheduler.java:203)
at io.reactivex.Scheduler.scheduleDirect(Scheduler.java:179)
at io.reactivex.internal.operators.observable.ObservableSubscribeOn.subscribeActual(ObservableSubscribeOn.java:36)
at io.reactivex.Observable.subscribe(Observable.java:12267)
at io.reactivex.internal.operators.observable.ObservableMap.subscribeActual(ObservableMap.java:32)
at io.reactivex.Observable.subscribe(Observable.java:12267)
at io.reactivex.internal.operators.observable.ObservableObserveOn.subscribeActual(ObservableObserveOn.java:45)
at io.reactivex.Observable.subscribe(Observable.java:12267)

原因是在创建大量请求时候会创建大量的线程, 在手机上由于手机对进程有线程数量的限制导致了闪退.

第一种解决方案

可以查看:
https://github.com/ReactiveX/RxAndroid/issues/387

最佳答案是:

JakeWharton commented on 9 Aug 2017

The problem is that Schedulers.io() uses a cached thread pool without
a limit and thus is trying to create 1500 threads. You should consider
using a Scheduler that has a fixed limit of threads, or using RxJava
2.x’s parallel() operator to parallelize the operation to a fixed number of workers.

If you’re using raw Retrofit by default it uses OkHttp’s dispatcher
which limits the threads to something like 64 (with a max of 5 per
host). That’s why you aren’t seeing it fail.

If you use createAsync() when creating the RxJava2CallAdapterFactory
it will create fully-async Observable instances that don’t require a
subscribeOn and which use OkHttp’s Dispatcher just like Retrofit would
otherwise. Then you only need observeOn to move back to the main
thread, and you avoid all additional thread creation.

主要意思就是: 如果只使用retrofit是不会出现这个问题的, 由于rxjava自身没有对线程数量做限制所以可能会出现这个问题.

这个解决方法主要有两点:
首先是在创建retrofit的时候使用 createAsync()

if (retrofit == null) {
            retrofit = new Retrofit.Builder()
                    .baseUrl(baseurl)       //设置远程地址
                    .addConverterFactory(GsonConverterFactory.create())
                    .addCallAdapterFactory(RxJava2CallAdapterFactory.createAsync())
                    .client(OkHttpHelper.getInstance().getOkHttpClient())
                    .build();
}

此时将创建完全异步的可见实例，这些实例不需要订阅，并且使用OkHttp的Dispatcher，就像Retrofit一样。
不需要订阅就表示不需要调用 .subscribeOn(getScheduler())

Observable<T> result = responseBody
//      .subscribeOn(Schedulers.io())    //如果使用RxJava2CallAdapterFactory.createAsync(),则不需要这一句
        .map(new Function<ResponseBody, T>() {
            @Override
            public T apply(ResponseBody responseBody) throws Exception {
                String string = responseBody.string();
                return new Gson().fromJson(string, clazz);
            }
        })
        .observeOn(AndroidSchedulers.mainThread());

最后的结果:

anilmaddala commented on 9 Aug 2017

Thanks @JakeWharton tried the createAsync() and removed subscribeOn to
Scheduler.io and I am not seeing the crash. Thanks.

I also posted the question in StackOverflow. If you can, please post
your answer there and I will mark it as a chosen answer.

这种解决方法有两点, 首先使用 RxJava2CallAdapterFactory.createAsync() 来自动创建异步, 然后不使用 .subscribeOn(Schedulers.io()) ,这样rxjava的线程数量就会做出和retrofit一样的限制, 因为retrofit不出问题, 则rxjava不出问题.

一个重要的后续

by 2020.8
最近,在升级项目架构的时候,也升级了rxjava3, 于是将retrofit的adapter也进行了升级,进行了如下引用

api "com.squareup.retrofit2:adapter-rxjava3:2.9.0"

同时,在创建retrofit的时候

//使用RxJava2版本
if (retrofit == null) {
    retrofit = new Retrofit.Builder()
                    .baseUrl(baseurl)      
                    .addConverterFactory(GsonConverterFactory.create())
                    .addCallAdapterFactory(RxJava2CallAdapterFactory.createAsync())
                    .client(OkHttpHelper.getInstance().getOkHttpClient())
                    .build();
}

升级到

//使用RxJava3版本
if (retrofit == null) {
    retrofit = new Retrofit.Builder()
                    .baseUrl(baseurl)      
                    .addConverterFactory(GsonConverterFactory.create())
                    .addCallAdapterFactory(RxJava3CallAdapterFactory.createSynchronous())
                    .client(OkHttpHelper.getInstance().getOkHttpClient())
                    .build();
}

发现网络请求不到数据了, 查看日志发现原因是: 在主线程请求了网络数据

比较蒙,经过排查, 原因是 RxJava3CallAdapterFactory 和 RxJava2CallAdapterFactory的调用方法是相反的,通过查看源码可以知道:

//RxJava2CallAdapterFactory
/**
 * 返回一个创建不操作任何调度器的同步可观察对象的实例
 * Returns an instance which creates synchronous observables that do not operate on any scheduler
 * by default.
 */
public static RxJava2CallAdapterFactory create() {
  return new RxJava2CallAdapterFactory(null, false);
}


//返回一个创建异步可观察对象的实例
/** Returns an instance which creates asynchronous observables. */
public static RxJava2CallAdapterFactory createAsync() {
  return new RxJava2CallAdapterFactory(null, true);
}

而在RxJava3CallAdapterFactory 中

//RxJava3CallAdapterFactory 
/**
 * 返回一个运行在后台线程上的异步可观察对象的实例
 * Returns an instance which creates asynchronous observables that run on a background thread by
 * default. Applying {@code subscribeOn(..)} has no effect on instances created by the returned
 * factory.
 */
public static RxJava3CallAdapterFactory create() {
  return new RxJava3CallAdapterFactory(null, true);
}

/**
 * 返回一个创建不操作任何调度器的同步可观察对象的实例
 * Returns an instance which creates synchronous observables that do not operate on any scheduler
 * by default. Applying {@code subscribeOn(..)} will change the scheduler on which the HTTP calls
 * are made.
 */
public static RxJava3CallAdapterFactory createSynchronous() {
  return new RxJava3CallAdapterFactory(null, false);
}

也就是说, 可能是因为RxJava2CallAdapterFactory 中默认没有对rxjava在使用

.subscribeOn(Schedulers.io())

方法的时候做出限制,导致很多人提交了 issues, 所以在RxJava3CallAdapterFactory 中默认对此作了限制,当然你自己也可以扩展,使用

.addCallAdapterFactory(RxJava3CallAdapterFactory.createSynchronous())

来自定义, 就像最初使用RxJava2 一样.
总之,RxJava3CallAdapterFactory对此默认作了限制,现在,可以放心的使用 create() 方法即可, 同时在RxJava中无需再使用

.subscribeOn(Schedulers.io())

end.

第二种解决方案

对rxjava的 Scheduler 进行自定义限制线程数量

Observable<T> result = responseBody
        .subscribeOn(getScheduler())
        .map(new Function<ResponseBody, T>() {
            @Override
            public T apply(ResponseBody responseBody) throws Exception {
                String string = responseBody.string();
                return new Gson().fromJson(string, clazz);
            }
        })
        .observeOn(AndroidSchedulers.mainThread());

private static Scheduler mScheduler;

private static Scheduler getScheduler() {
    if (mScheduler == null) {
        synchronized (HttpLoaderUtils.class) {
            if (mScheduler == null) {
                mScheduler = Schedulers.from(Executors.newFixedThreadPool(15));
            }
        }
    }
    return mScheduler;
}

自行对Scheduler的数量进行限制.
可以查看:
Android 栈内存溢出bug

总结

两种方法的解决核心都是对rxjava自身的线程数量的限制.

第一种方案通过Retrofit配置adapter对Rxjava产生影响,使其线程的调度用Retrofit本身提供的线程,其中有一句:

/**
 * 返回一个运行在后台线程上的异步可观察对象的实例
 * Returns an instance which creates asynchronous observables that run on a background thread by
 * default. Applying {@code subscribeOn(..)} has no effect on instances created by the returned
 * factory.
 */
public static RxJava3CallAdapterFactory create() {
  return new RxJava3CallAdapterFactory(null, true);
}

也就是说用了这个以后即使在RxJava中调用 subscribeOn(…) 方法也不会影响返回的实例.

第二种方案是通过RxJava自身调度来限制其所使用的线程数量,因为 .subscribeOn(Schedulers.io())方法不对线程数量做任何限制, 所以当并发请求的接口数量过多就可能会出现OOM的情况,通过线程池来限制也是一种解决方案.