适用场景

[happydog-intj]

[ZhaoKaiQiang]

我先来开个头~

create操作符适用于创建Observable对象，使用Retrofit可以直接生成Observable，但是对于不使用Retrofit的开发者咋办呢？下面的方法，可以在任意Http请求框架下生成Observable对象

异步请求可以这样写，PS：OkHttpProxy是我对OkHttp的一个简单封装，主要把回调用Handler弄到了UI线程，所以下面的代码不需要指定观察者所在线程

public static Observable<User> getUser() {

        return Observable.create((Observable.OnSubscribe<User>) subscriber ->
                OkHttpProxy.get().url(URL_USER).enqueue(new OkCallback<User>(new OkJsonParser<User>() {
                }) {
                    @Override
                    public void onSuccess(int code, User user) {
                        subscriber.onNext(user);
                        subscriber.onCompleted();
                    }

                    @Override
                    public void onFailure(Throwable e) {
                        subscriber.onError(e);
                    }
                }));
    }

同步请求可以这样写，PS：由于是同步请求，所以需要指定subscribeOn()所在线程和observeOn所在线程，注意，这里的同步请求指的是网络请求会在发起请求的线程执行，所以我们需要指定Schedulers.io()开启工作线程

 public static Observable<User> getUserSync() {

        return Observable.create((Observable.OnSubscribe<User>) subscriber -> {
            try {
                subscriber.onNext(new OkJsonParser<User>() {
                }.parseResponse(OkHttpProxy.get().url(URL_USER).execute()));
            } catch (IOException e) {
                subscriber.onError(e);
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread());
    }

通过上面简单的封装，你就可以把现有的Http框架改造为RxJava适用的框架，从而可以体验RxJava带来的便捷

[geminiwen]

返回不同的Observable模板：
比如一个Observable和一个Observable

Observable.zip(mRxUserPresenter.mineInfo(), mRxUserPresenter.stat(), (m, s) -> {
            handleUserInfo(m);
            handleUserStats(s);
            return Observable.empty();
        })
        .subscribeOn(Schedulers.io())
        .doOnSubscribe(() -> mContentProgressBar.setVisibility(View.VISIBLE))
        .doOnTerminate(() -> mContentProgressBar.setVisibility(View.GONE))
        .subscribe(r -> {},
                   this::handleException);

我是这么用的，有没有谁有更好的方案

[androidMVP]

我也来。RxJava经过封装后可以替代EventBus，实现组件跟组件之间的通讯。

[lincleejun]

@ZhaoKaiQiang 后面的线程切换组合，感觉参考 dan lew的会更骚气。
链接在此：http://blog.danlew.net/2015/03/02/dont-break-the-chain/
＃我不是来搅局的－ －！

[geminiwen]

@lincleejun 谢谢！！！！我要的东西！！！

[BaronZ]

mark

[happydog-intj]

dont break the chain是不错的文章。
不打断RxJava的链式调用，主要就是使用两个 operator， compose和lift

[evanwang1021]

结合Retrofit简直不能更好用

[jackTang11]

求指导 是否可以通用

    Observable.create(new Observable.OnSubscribe<Drawable>() {
        @Override
        public void call(Subscriber<? super Drawable> subscriber) {
            Logger.d("call  当前所在子线程=====");
            insertData();//插入数据库的耗时操作,或者一些耗时操作
            subscriber.onNext(null);
            subscriber.onCompleted();
        }
    }).subscribeOn(Schedulers.io())
       .doOnSubscribe(new Action0() {
           @Override
           public void call() {
               dialog.show();//启动前先调用
               Logger.d("doOnSubscribe  当前所在线程=====");
           }
       })
         .subscribeOn(AndroidSchedulers.mainThread())
         .subscribe(new Observer<Drawable>() {
             @Override
             public void onCompleted() {
                 Logger.d("onCompleted  当前所在线程=====");//任务执行完成
                    Intent intent = new Intent(this, MainActivity.class);//插入完成 跳转页面
                    startActivity(intent);
             }

             @Override
             public void onError(Throwable e) {
             }

             @Override
             public void onNext(Drawable drawable) {
                 Logger.d("onNext  当前所在线程=====");
                    dialog.dismiss();
             }
         });

[geminiwen]

@jackTang11 我今天早上踏过的坑， 你在调用subscribe()的时候，千万要注意自己是在什么线程里，因为它和你的doOnSubscribe是在同一个线程执行的，也就是说，如果你subscribe()在非主线程，那么在doOnSubscribe做UI的操作就会crash...

[jackTang11]

@geminiwen 我看了 如果放在子线程 确实不行,还好我没有放在子线程

@ZhaoKaiQiang @lzyzsd @lincleejun @androidMVP
各位老大 帮忙看看 我写的代码是否有问题 ,以后我做耗时操作(比如,查询数据库,遍历对比,发送socket)就这么做,可以吗

[asker517]

@jackTang11 我觉的你这样的写法是有问题的 建议认真阅读 给 Android 开发者的 RxJava 详解 或者本工程的作者博客**大头鬼Bruce**关于RxJava的介绍以及其线程切换和Schedulers使用的相关内容吧，就目前你的代码中，最起码最后一个.subscribe()方法你所处理的事情应该切回(.observeOn(AndroidSchedulers.mainThread))UI线程去做吧 😄 I‘m a Rx Newer
ps.使用方法的问题是不是建议另外open a new issue？

[SmartDengg]

抱歉，来晚了。

@ZhaoKaiQiang 开了个好头，而且OkHttpProxy用起来酷极了。

RxJava提供了丰富的API用来创建Observable，当然最灵活的还是使用Observable.create( )方式来自定义发送事件。虽然使用方式灵活多变，但我们最好还是遵循Observable条约和Rx设计规范
，实现Observer接口的实现类，允许使用.onNext( )发送任意数量的事件，并且可选择性的添加.onCompleted( )或.onError( )操作符，但只能添加二者之一，这样表明Observable事件序列已经结束了，保证Observable事件序列的消费者，能够安全的执行清理操作。当然，一般情况下传入.subscribe( )中的subscriber会被封装成SafeSubscriber，其中成员变量done能够确保调用onCompleted( )或.onError( )后Subscriber不再接受发送自.onNext( )的事件，然后自己自动解除订阅关系。另外需要补充的是SafeSubscriber并不能保证.onNext( )发送事件的同步执行，.unsubscribe( )的调用，也不能保证链上的操作符立即停止发送事件。

以下是我给出的基本使用方式，如有建议请指出。

Observable.create(new Observable.OnSubscribe<Object>() {
    @Override public void call(Subscriber<? super Object> subscriber) {

      if (!subscriber.isUnsubscribed()) {

        subscriber.onNext(new Object());
        /*如果不需要复用Observable，可以添加完成通知，但是对于一个SafeSubscriber来说，会触发unsubscribe*/
        subscriber.onCompleted();
      }

      subscriber.add(new Subscription() {

        /*需要做同步处理，保证线程安全，建议使用AtomicIntegerFieldUpdater*/
        public volatile boolean isUnsubscribed;

        @Override public void unsubscribe() {
          isUnsubscribed = true;
          /*解除订阅时触发，处理取消网络请求，注销监听事件，取消FutureTask异步回调，释放内存等操作*/
          /*do some clear*/
        }

        @Override public boolean isUnsubscribed() {
          return isUnsubscribed;
        }
      });
    }
  }).subscribe(new Subscriber<Object>() {
    @Override public void onCompleted() {
    }
    @Override public void onError(Throwable e) {
    }
    @Override public void onNext(Object o) {
    }
  });

当然这种使用方式并不是万能的，大多数情况下还要根据具体业务逻辑以及线程要求对其进行改造。

[zsxwing]

      subscriber.add(new Subscription() {

        /*需要做同步处理，保证线程安全，建议使用AtomicIntegerFieldUpdater*/
        public volatile boolean isUnsubscribed;

        @Override public void unsubscribe() {
          isUnsubscribed = true;
          /*解除订阅时触发，处理取消网络请求，注销监听事件，取消FutureTask异步回调，释放内存等操作*/
          /*do some clear*/
        }

        @Override public boolean isUnsubscribed() {
          return isUnsubscribed;
        }
      });

这个Subscription如果两个线程同时调用unsubscribe，可能导致clear的工作执行了两遍。这段可以用Subscriptions.create来简化一下，不需要自己处理isUnsubscribed。

[zsxwing]

尽量不要自己用lift去实现新的操作。如果真的需要，建议先看一下Dávid Karnok写的Pitfalls of operator implementations系列（需翻墙），不然很难写出没有bug的操作来，特别是一些并发的bug，自己写的测试不一定能触发。