RxJava和Retrofit2:NetworkOnMainThreadException

Question

我意识到我在MainThread上使用了subscribeOn()/ observeOn().我可以传递给subscribeOn()的选项有哪些？我可以传递给observeOn()的选项有哪些？

12-17 21:36:09.154 20550-20550/rx.test D/MainActivity2: [onCreate]
12-17 21:36:09.231 20550-20550/rx.test D/MainActivity2: starting up observable...
12-17 21:36:09.256 20550-20550/rx.test D/MainActivity2: [onError] 
12-17 21:36:09.256 20550-20550/rx.test W/System.err: android.os.NetworkOnMainThreadException

GovService.java

import java.util.List;
import retrofit.Call;
import retrofit.http.GET;
import rx.Observable;

public interface GovService {
    @GET("/txt2lrn/sat/index_1.json")
    Observable<MyTest> getOneTestRx();
}

MyTest.java

public class MyTest {
    private String name, url;
    private int num;

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getUrl() {
        return url;
    }

    public void setUrl(String url) {
        this.url = url;
    }

    public int getNum() {
        return num;
    }

    public void setNum(int num) {
        this.num = num;
    }

    @Override
    public String toString() {
        return "Name: " + this.name + ", num: " + this.num + ", url: " + this.url;
    }
}

MainActivity2.java

import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.support.v7.widget.DefaultItemAnimator;
import android.support.v7.widget.LinearLayoutManager;
import android.support.v7.widget.RecyclerView;
import android.support.v7.widget.Toolbar;
import android.util.Log;

import retrofit.GsonConverterFactory;
import retrofit.Retrofit;
import retrofit.RxJavaCallAdapterFactory;
import rx.Observable;
import rx.Subscriber;
import rx.android.schedulers.AndroidSchedulers;
import rx.schedulers.Schedulers;

public class MainActivity2 extends AppCompatActivity {
    private final String TAG = getClass().getSimpleName();

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        Log.d(TAG, "[onCreate]");
        setContentView(R.layout.activity_main);
        Toolbar toolbar = (Toolbar) findViewById(R.id.toolbar);
        setSupportActionBar(toolbar);
        RecyclerView mRV = (RecyclerView) findViewById(R.id.rv);
        mRV.setLayoutManager(new LinearLayoutManager(this));// setup LayoutManager
        mRV.setItemAnimator(new DefaultItemAnimator());// setup ItemAnimator

        // setup retrofit
        Retrofit retrofit = new Retrofit.Builder()
                .baseUrl("http://goanuj.freeshell.org")
                .addConverterFactory(GsonConverterFactory.create())
                .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
                .build();
        GovService service = retrofit.create(GovService.class);

        Log.d(TAG, "starting up observable...");
        Observable<MyTest> o = service.getOneTestRx();
        o.subscribeOn(Schedulers.io());
        o.observeOn(AndroidSchedulers.mainThread());
        o.subscribe(new Subscriber<MyTest>() {
            @Override
            public void onCompleted() {
                Log.d(TAG, "[onCompleted] ");
            }

            @Override
            public void onError(Throwable t) {
                Log.d(TAG, "[onError] ");
                t.printStackTrace();
            }

            @Override
            public void onNext(MyTest m) {
                Log.d(TAG, "[onNext] " + m.toString());
            }
        });
    }
}

Answer 1

将代码的最后一部分重写为:

service.getOneTestRx()
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Subscriber<MyTest>() {
        @Override
        public void onCompleted() {
            Log.d(TAG, "[onCompleted] ");
        }

        @Override
        public void onError(Throwable t) {
            Log.d(TAG, "[onError] ");
            t.printStackTrace();
        }

        @Override
        public void onNext(MyTest m) {
            Log.d(TAG, "[onNext] " + m.toString());
        }
    });

来自@akarnokd的重要提示:

值得一提的是,需要将调用链接到此处,因为Observable不是构建器模式(您修改现有对象的设置)

Answer 2

你应该打电话Observable.unsubscribeOn(Schedulers.io()),retrofit在http请求结束时取消订阅.

在RxJavaCallAdapterFactory中retrofit-rxjava-adapter

它是这样的行动.

if (!subscriber.isUnsubscribed()) {
    subscriber.onCompleted();
}

但是,当subscriber是SafeSubscriber,它会调用unsubscribe最后.

我在我的应用程序中遇到此问题.

完整代码:

o.subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .unsubscribeOn(Schedulers.io());