面向函數語言程式設計, 即會將函式作為一個數據物件.下面看Android中RxJava的實現

RxJava 是一個觀察者模式的擴充套件:

RxJava 一個觀察者模式的擴充套件,
觀察者模式: 比如Button的點選事件,對設定 OnClickListener 來說， Button 是被觀察者， OnClickListener 是觀察者，二者通過 setOnClickListener() 方法產生關係。
            OnClickListener一直觀察著Button，當Button被點選，OnClickListener執行onClick事件。
            而RxJava中 Observable(可觀察的,被觀察者),Observer(觀察者),subscribe(訂閱),事件;Observable和Observer
            之間通過subscribe()方法實現訂閱關係,從而Observable可以在需要的時候發出事件來通知Observer


       以下的Demo即為幾個實現的小例子

TestActivity是RxJava幾個關鍵字的用法;  被觀察者訂閱觀察者, 被觀察者提供資料, 觀察者觀察到資料的變化, onNext()方法中得到資料, 處理資料

package com.example.mytestrxjava;

import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.view.View;

import rx.Observable;
import rx.Observer;
import rx.Subscriber;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Func1;

/*
響應式函式程式設計
RxJava 一個觀察者模式的擴充套件,
觀察者模式: 比如Button的點選事件,對設定 OnClickListener 來說， Button 是被觀察者， OnClickListener 是觀察者，二者通過 setOnClickListener() 方法產生關係。
            OnClickListener一直觀察著Button，當Button被點選，OnClickListener執行onClick事件。
            而RxJava中 Observable(可觀察的,被觀察者),Observer(觀察者),subscribe(訂閱),事件;Observable和Observer
            之間通過subscribe()方法實現訂閱關係,從而Observable可以在需要的時候發出事件來通知Observer

       以下的Demo即為幾個實現的小例子

 */
public class TestActivity extends AppCompatActivity {
    public static final String TAG = "AppCompatActivity";

    private  String[] words = {"aaaa", "bbbb", "cccc"};

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_test);

    }

    public void onClick(View view) {
        switch (view.getId()) {
            case R.id.btn_sample:
                test();
                break;
            case R.id.btn_just:
                just();
                break;
            case R.id.btn_from:
                myFrom();
                break;
            case R.id.btn_action:
                action();
                break;
            case R.id.btn_map:
                map();
                break;
            case R.id.btn_flatmap:
                flatmap();
                break;
            case R.id.btn_filter:
                filter();
                break;
        }
    }




    private void test() {
        //被觀察者
        Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("aaaa");
                subscriber.onNext("bbbb");
                subscriber.onNext("cccc");
                subscriber.onCompleted();
            }
        });
        //觀察者
        Observer<String> observer = new Observer<String>() {


            @Override
            public void onCompleted() {
                Log.e(TAG,"onCompleted");
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG,"Error" + e.toString());
            }

            @Override
            public void onNext(String s) {
                Log.e(TAG,s);
            }
        };

        //被觀察者 訂閱觀察者
        observable.subscribe(observer);
    }

    private void just() {
        //使用just 不用重寫call方法,  被觀察者
        Observable<String> observable = Observable.just("aaa2","bbb2","ccc2");
        Observer<String> observer = new Observer<String>() {
            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted");
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "Error" + e.toString());
            }

            @Override
            public void onNext(String s) {
                Log.e(TAG, s);
            }
        };
        observable.subscribe(observer);

    }

    /** 當傳入的是一個數組的時候 , 可以使用from*/
    private void myFrom() {
        Observable<String> observable = Observable.from(words);
        Observer<String> observer = new Observer<String>() {
            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted");
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "Error" + e.toString());
            }

            @Override
            public void onNext(String s) {
                Log.e(TAG, s);
            }
        };
        observable.subscribe(observer);
    }

    /** 不是訂閱一個觀察者, 而是訂閱一個單獨執行的任務 */
    private void action() {
        Action1 action1 = new Action1<String>() {
            @Override
            public void call(String s) {
                Log.e(TAG,s);
            }
        };
       Action1 action2 = new Action1<Throwable>() {
           @Override
           public void call(Throwable e) {
               Log.e(TAG, "Error" + e.toString());
           }
       };

        Action0 action3 = new Action0() {
            @Override
            public void call() {
                Log.e(TAG,"oncompleted");
            }
        };
        Observable<String> observable = Observable.from(words);
        observable.subscribe(action1);
        observable.subscribe(action1,action2);
        observable.subscribe(action1,action2,action3);
    }


    /**
     * 被觀察者進行資料轉換, 如果我們希望傳入的是String型別,而處理的型別是int型別
     */
    private void map() {
        Observable.just("aaa","bbb","ccc")
                .map(new Func1<String, Integer>() {
                    @Override
                    public Integer call(String s) {
                        return s.length();
                    }
                })
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        Log.e(TAG,"length" + integer);   //因為傳入了三個物件,  所以被呼叫了三次
                    }
                })
                ;
    }


    /**
     * flapMap()中返回的是個Observable物件,並且這個Observable物件並不是被直接傳送到了
     * Subscriber回撥方法中.
     * 例如下面的例子,可以看做是,一個String返回了一個Observable,一個Observable執行了三次Action
     */
    private void flatmap() {
        Observable.just("aaaa","bbb","cc")
                .flatMap(new Func1<String, Observable<Integer>>() {
                    @Override
                    public Observable<Integer> call(String s) {
                        Integer[] info = new Integer[3];
                        info[0] = s.length();
                        info[1] = s.hashCode();
                        info[2] = s.getBytes().length;
                        return Observable.from(info);    //有陣列可以使用from關鍵字

                    }
                }).subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer integer) {
                Log.e(TAG,integer+"");
            }
        });
    }


    /**
     * filter指定過濾
     * 比如我指定過濾字串中含有"a"的字串
     */
    private void filter() {
        //一句話鏈式程式設計解決文字,   被觀察者訂閱了觀察者
        Observable.from(words)
                .filter(new Func1<String, Boolean>() {
                    @Override
                    public Boolean call(String s) {
                        return s.contains("a");
                    }
                })
                .subscribe(new Observer<String>() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(String s) {
                        Log.e(TAG,"過濾的字串為" + s);
                    }
                });
    }





}
 

執行緒管理Scheduler使用:

在不指定執行緒的情況下， RxJava 遵循的是執行緒不變的原則，即：在哪個執行緒呼叫 subscribe()，就在哪個執行緒生產事件；在哪個執行緒生產事件，就在哪個執行緒消費事件。如果需要切換執行緒，就需要用到 Scheduler （排程器）。


Schedulers.immediate(): 直接在當前執行緒執行，相當於不指定執行緒。這是預設的 Scheduler。


Schedulers.newThread(): 總是啟用新執行緒，並在新執行緒執行操作。


Schedulers.io(): I/O 操作（讀寫檔案、讀寫資料庫、網路資訊互動等）所使用的 Scheduler。行為模式和 newThread() 差不多，區別在於 io() 的內部實現是是用一個無數量上限的執行緒池，可以重用空閒的執行緒，因此多數情況下 io() 比 newThread() 更有效率。不要把計算工作放在 io() 中，可以避免建立不必要的執行緒。


Schedulers.computation(): 計算所使用的 Scheduler。這個計算指的是 CPU 密集型計算，即不會被 I/O 等操作限制性能的操作，例如圖形的計算。這個 Scheduler 使用的固定的執行緒池，大小為 CPU 核數。不要把 I/O 操作放在 computation() 中，否則 I/O 操作的等待時間會浪費 CPU。


另外， Android 還有一個專用的AndroidSchedulers.mainThread()，它指定的操作將在 Android 主執行緒執行。

下面的Demo ImageActivity就是在網路上獲取圖片然後展示的例子

package com.example.mytestrxjava;

import android.graphics.Bitmap;
import android.graphics.BitmapFactory;
import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.widget.ImageView;
import android.widget.Toast;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.HttpURLConnection;
import java.net.URL;

import rx.Observable;
import rx.Observer;
import rx.Subscriber;
import rx.android.schedulers.AndroidSchedulers;
import rx.schedulers.Schedulers;

public class ImageActivity extends AppCompatActivity {
    private static final String TAG = "ImageActivity";
    public static String imageurl = "https://mobile.umeng.com/images/pic/home/social/img-1.png";
    private ImageView mIv;


    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_image);

        mIv = (ImageView) findViewById(R.id.iv);

        init();
    }

    private void init() {
        //被觀察者
        Observable<Bitmap> observable = Observable.create(new Observable.OnSubscribe<Bitmap>() {
            @Override
            public void call(Subscriber<? super Bitmap> subscriber) {
                byte[] data = getNetData();
                Bitmap bitmap1 = binary2Bitmap(data);
                subscriber.onNext(bitmap1);    //在IO執行緒,  且訂閱者物件將資料傳遞給觀察者
                subscriber.onCompleted();
            }
        })
                .subscribeOn(Schedulers.io())    //指定subscribe()發生在IO執行緒
                .observeOn(AndroidSchedulers.mainThread())   //指定Subscriber的回調發生在主執行緒
                ;

        //觀察者
        Observer<Bitmap> observer = new Observer<Bitmap>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {
                Toast.makeText(ImageActivity.this,e.toString(),Toast.LENGTH_SHORT).show();
            }

            @Override
            public void onNext(Bitmap bitmap) {
                //觀察者對 被觀察者給的資料進行處理
                Log.e(TAG,"bitmap:" + bitmap.toString());
                mIv.setImageBitmap(bitmap);
            }
        };
        observable.subscribe(observer);
    }

    /**
     * 將byte[]陣列資料 轉化為BItmap物件
     * @param data
     * @return
     */
    private Bitmap binary2Bitmap(byte[] data) {
        if(data != null) {
            return BitmapFactory.decodeByteArray(data,0,data.length);
        }
        return null;
    }

    /**
     * 從網路獲取圖片的資料,得到位元組資料
     * @return
     */
    private byte[] getNetData() {
        ByteArrayOutputStream boas = null;
        InputStream in = null;
        try {
            boas = new ByteArrayOutputStream();

            URL url = new URL(imageurl);  //異常抓取
                HttpURLConnection httpURLConnection = (HttpURLConnection) url.openConnection();
                httpURLConnection.setConnectTimeout(2000);
            in = httpURLConnection.getInputStream();
            byte[] buf = new byte[4 * 1024];
            int b;
            while ((b = in.read(buf)) != -1) {
                boas.write(buf,0,b);
            }
            Log.e(TAG,boas.toString());
            return boas.toByteArray();


        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }finally {                   //流的異常處理
            if(in != null) {
                try {
                    in.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }finally {
                    if(boas != null) {
                        try {
                            boas.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
        }
    }
}

至此,Rxjava的使用方法完畢