RxJava y Retrofit2: NetworkOnMainThreadException

Me doy cuenta de que estoy usando subscribeOn()/observeOn() en MainThread. ¿Cuáles son el conjunto de opciones que puedo pasar a subscribeOn()? ¿Cuáles son el conjunto de opciones que puedo pasar a 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();
}

MiPrueba.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;
    }
}

Actividad principal2.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());
            }
        });
    }
}

Jan 01 '70 08:01 AG1

Vuelva a escribir la última parte de su código para:

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());
        }
    });

Nota importante de @akarnokd:

Vale la pena mencionar que es necesario encadenar las llamadas como aquí porque Observable no es el patrón de creación (donde se modifica la configuración de un objeto existente)

Dec 18 '2015 06:12 Ilya Tretyakov

Debes llamar Observable.unsubscribeOn(Schedulers.io()), retrofitse cancelará la suscripción al final de una solicitud http.

en RxJavaCallAdapterFactoryderetrofit-rxjava-adapter

actúa así.

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

Pero cuando subscribersea un , finalmente SafeSubscriberllamará .unsubscribe

Tengo este problema en mi aplicación.

Código completo:

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

Dec 18 '2015 06:12 GeminiWen