Rx是响应式编程的意思, 本质是观察者模式, 是以观察者(Observer)和订阅者(Subscriber)为基础的异步响应方式. 在Android编程时, 经常会使用后台线程, 那么就可以使用这种方式. 目前的异步编程方式都会导致一些问题, 如

(1) Asynctasks can easily lead to memory leaks

(2) CursorLoaders with a ContentProvider require a large amount of configuration and boilerplate code to setup.

(3) Services are intended for longer running background tasks and not fast-finishing operations, such as making a network call or loading content from a database.

RxAndroid来源于RxJava, 在RxJava的基础上扩展了一些Android的功能 来处理异步的任务处理,它可以使代码更加优雅，代码的可维护性更强。

下面简单介绍Rx_Android的几种实用方法

1.create方式(1)建立

/**
 * 使用create方式
 */
public static void createObserable(){
    Observable<String>observable=Observable.create(new Observable.OnSubscribe<String>(){

        @Override
        public void call(Subscriber<? super String> subscriber) {
             if(!subscriber.isUnsubscribed()){
                 subscriber.onNext("hello");
                 subscriber.onNext("world");
                 subscriber.onNext(downloadJson());
                 subscriber.onCompleted();
             }
        }
    });

    Subscriber<String>showsub=new Subscriber<String>() {
        @Override
        public void onCompleted() {
            Log.e(TAG,"onCompleted");
        }

        @Override
        public void onError(Throwable e) {
            Log.e(TAG,e.getMessage());
        }

        @Override
        public void onNext(String s) {
            Log.e(TAG,"result->>"+s);
        }
    };
    //关联被观察者
    observable.subscribe(showsub);
}

/**
 * 调用下载方法
 * @return
 */
public static String downloadJson(){
    return "json success!";
}

cretae 方式(2)建立

/**
 * create的第2中方法
 */
public static void createPrint(){
    Observable.create(new Observable.OnSubscribe<Integer>() {
        @Override
        public void call(Subscriber<? super Integer> subscriber) {
            if(!subscriber.isUnsubscribed()){
                for(int i=0;i<10;i++){
                    subscriber.onNext(i);
                }
                subscriber.onCompleted();
            }
        }
    }).subscribe(new Subscriber<Integer>() {
        @Override
        public void onCompleted() {
            Log.i(TAG,"onCompleted");
        }

        @Override
        public void onError(Throwable e) {
            Log.i(TAG,e.getMessage());
        }

        @Override
        public void onNext(Integer integer) {
            Log.i(TAG,"result-->>"+integer);
        }
    });
}

原文链接：http://www.apkbus.com/blog-813041-62859.html