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

從@akarnokd

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

重要提示：

Answer 1

你的代碼的最後一部分改寫爲

值得一提的是一個需要鏈中的通話這裏，因爲 可觀察是不生成器模式（在此修改現有對象的設置 ）

Answer 2

您應該致電Observable.unsubscribeOn(Schedulers.io()),retrofit將在http請求結束時取消訂閱。

在retrofit-rxjava-adapter

它的操作是這樣RxJavaCallAdapterFactory。

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

但當subscriber是SafeSubscriber，它會最後調用unsubscribe。

我在我的應用程序中遇到此問題。

全碼：

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