Làm thế nào để quan sátBên trong chuỗi gọi trong java rx?

Question

là có anyway để nói java rx sử dụng các chủ đề hiện tại trong các chức năng quan sát? Tôi đang viết mã cho syncadapter android và tôi muốn kết quả được quan sát trong thread adapter đồng bộ và không nằm trong chủ đề chính.

Một ví dụ cuộc gọi mạng với Retrofit + RX Java trông giống như thế:

MyRetrofitApi.getInstance().getObjects() 
.subscribeOn(Schedulers.io()) 
.observeOn(<current_thread>) 
.subscribe(new Subscriber<Object>() { 
    //do stuff on the sync adapter thread 

} 

Tôi đã cố gắng sử dụng sử dụng

... 
.observeOn(AndroidSchedulers.handlerThread(new Handler(Looper.myLooper()))) 
... 

đó là cùng một cách android rx tạo ra lịch trình cho các chủ đề chính nhưng không hoạt động ngay khi tôi thay thế Looper.myLooper() cho Looper.getMainLooper().

Tôi có thể sử dụng Schedulers.newThread() nhưng là mã đồng bộ phức tạp với nhiều cuộc gọi máy chủ. Tôi sẽ liên tục tạo chuỗi mới chỉ để kích hoạt cuộc gọi mạng mới tạo lại chủ đề mới để khởi chạy nhiều cuộc gọi mạng hơn . Có cách nào để làm việc này không? Hoặc là cách tiếp cận của tôi chính nó hoàn toàn sai?

Answer 1

Ồ, tôi chỉ thấy điều này trong wiki tại địa chỉ: https://github.com/ReactiveX/RxAndroid#observing-on-arbitrary-threads

new Thread(new Runnable() { 
    @Override 
    public void run() { 
     final Handler handler = new Handler(); // bound to this thread 
     Observable.just("one", "two", "three", "four", "five") 
       .subscribeOn(Schedulers.newThread()) 
       .observeOn(HandlerScheduler.from(handler)) 
       .subscribe(/* an Observer */) 

     // perform work, ... 
    } 
}, "custom-thread-1").start(); 

Tôi nghĩ rằng điều này sẽ làm việc cho trường hợp của bạn, quá - ngoại trừ việc tạo ra một chủ đề mới, tất nhiên ... Vì vậy chỉ:

final Handler handler = new Handler(); // bound to this thread 
MyRetrofitApi.getInstance().getObjects() 
    .subscribeOn(Schedulers.io()) 
    .observeOn(HandlerScheduler.from(handler)) 
    .subscribe(new Subscriber<Object>() { 
     //do stuff on the sync adapter thread 

    } 

Answer 2

hãy thử sử dụng Schedulers.immediate()

MyRetrofitApi.getInstance().getObjects() 
.subscribeOn(Schedulers.io()) 
.observeOn(Schedulers.immediate()) 
.subscribe(new Subscriber<Object>() { 
    //do stuff on the sync adapter thread 

} 

mô tả của nó nói: Creates and returns a Scheduler that executes work immediately on the current thread.

LƯU Ý:
tôi nghĩ rằng nó không quan trọng để giữ tất cả các công việc trên chủ đề của SyncAdapter bởi vì nó đã được sử dụng một sợi khác nhau