函数式编程--响应式编程 ---android应用例子

RxJava implements this operator as create.

It is good practice to check the observer’s isUnsubscribed state from within the function you pass tocreate so that your Observable can stop emitting items or doing expensive calculations when there is no longer an interested observer.

创建的这个流，会产生三个值(事件)，值可以用户自己定义。

Observable.create(xxxx)，返回一个流

　　

XXXX方法，是用户自定义的。要实现一个自定义的call方法。

Observable.OnSubscribe<?>，？指定了流产生的事件的值类型，一个事件包含了值的信息。

在流里面，检查外部是否有观察者(observer)，如果没有就不用产生事件了。这样可以避免，在没有观察者处理产生的事件时，依然在发出事件。

 

在下述代码里面，流它做的事情是，用红色代码表示，产生4个onNext事件；然后，产生一个onCompleted事件。

然后，观察者，订阅流发出的事件，对流发出的事件做响应，用蓝色代码表示。 

下述代码的事件流:

--onNext---onNext---onNext---onNext---onCompleted--

订阅者对事件进行处理

Observable.create(new Observable.OnSubscribe<Integer>() {
    @Override
    public void call(Subscriber<? super Integer> observer) {
        try {
            if (!observer.isUnsubscribed()) {
                for (int i = 1; i < 5; i++) {
                    observer.onNext(i);
                }
                observer.onCompleted();
            }
        } catch (Exception e) {
            observer.onError(e);
        }
    }
 } ).subscribe(new Subscriber<Integer>() {
        @Override
        public void onNext(Integer item) {
            System.out.println("Next: " + item);
        }

        @Override
        public void onError(Throwable error) {
            System.err.println("Error: " + error.getMessage());
        }

        @Override
        public void onCompleted() {
            System.out.println("Sequence complete.");
        }
    });

结果：

Next: 1
Next: 2
Next: 3
Next: 4
Sequence complete.

Scheduler

If you want to introduce multithreading into your cascade of Observable operators, you can do so by instructing those operators (or particular Observables) to operate on particular Schedulers.

 

Scheduler用来定义，流中产生值的操作，在哪个线程工作；订阅者，在哪个线程工作。

 

Some ReactiveX Observable operators have variants that take a Scheduler as a parameter. These instruct the operator to do some or all of its work on a particular Scheduler.

可以指定流的操作是在哪个线程中执行的，通过传入一个Scheduler参数。

 

You can make an Observable act on a particular Scheduler by means of the ObserveOn or SubscribeOn operators. ObserveOn instructs an Observable to call its observer’s onNext, onError, and onCompletedmethods on a particular Scheduler; SubscribeOn takes this a step further and instructs the Observable to do all of its processing (including the sending of items and notifications to observers) on a particular Scheduler.

 

你可以让一个流，在某个Scheduler中执行，通过给流的方法ObserveOn，SubscribeOn提供参数Scheduler。

ObserveOn方法，指定流(Observable)在哪个线程中调用观察者(observer)的onNext,onError,onCompleted方法。

SubscribeOn指定流的所有处理，生成事件和值的执行操作，是在哪个线程中执行的。

 

Varieties of Scheduler 

You obtain a Scheduler from the factory methods described in the Schedulers class. The following table shows the varieties of Scheduler that are available to you by means of these methods in RxJava:

可以通过Scheduler的工厂方法，取得一个scheduler实例。

根据使用场景，选择合适的Scheduler实例。

 

Scheduler	purpose
Schedulers.computation( )	meant for computational work such as event-loops and callback processing; do not use this scheduler for I/O (use Schedulers.io( )instead); the number of threads, by default, is equal to the number of processors
Schedulers.from(executor)	uses the specified Executor as a Scheduler
Schedulers.immediate( )	schedules work to begin immediately in the current thread
Schedulers.io( )	meant for I/O-bound work such as asynchronous performance of blocking I/O, this scheduler is backed by a thread-pool that will grow as needed; for ordinary computational work, switch toSchedulers.computation( )
Schedulers.newThread( )	creates a new thread for each unit of work
Schedulers.trampoline( )	queues work to begin on the current thread after any already-queued work

---------------------------------------

完整例子代码

 

1.回调方法执行先后顺序

------->onCreateView------>onActivityCreated---->onDestroy

 

2.执行界面元素注入，初始化界面操作。

 

3.然后，监听点击事件。每次点击的时候，就专门生成一个事件流(Observable)，并且生成一个观察者Observer专门订阅事件流发出的事件。

Observer处理发出的事件是在UI线程中进行，事件流调用方法observeOn(AndroidSchedulers.mainThread())来达到。

事件流，产生事件和执行后台操作，则是在一个非线程中执行，调用方法subscribeOn(Schedulers.io())。

 

4.界面销毁的时候

则不再订阅事件流。

 

下述例子，一个事件流的过程:

首先_doSomeLongOperation_thatBlocksCurrentThread执行该操作，然后，依次发出

--onNext--onCompleted--

 

相应的，订阅者，则对发出的事件进行处理。

public class ConcurrencyWithSchedulersDemoFragment extends BaseFragment {

    @InjectView(R.id.progress_operation_running) ProgressBar _progress;
    @InjectView(R.id.list_threading_log) ListView _logsList;

    private LogAdapter _adapter;
    private List<String> _logs;
    private Subscription _subscription;

    @Override
    public void onDestroy() {
        super.onDestroy();
        if (_subscription != null) {
            _subscription.unsubscribe();
        }
    }

    @Override
    public void onActivityCreated(@Nullable Bundle savedInstanceState) {
        super.onActivityCreated(savedInstanceState);
        _setupLogger();
    }

    @Override
    public View onCreateView(LayoutInflater inflater,
                             @Nullable ViewGroup container,
                             @Nullable Bundle savedInstanceState) {
        View layout = inflater.inflate(R.layout.fragment_concurrency_schedulers, container, false);
        ButterKnife.inject(this, layout);
        return layout;
    }

    @OnClick(R.id.btn_start_operation)
    public void startLongOperation() {

        _progress.setVisibility(View.VISIBLE);
        _log("Button Clicked");

        _subscription = AppObservable.bindFragment(this, _getObservable())      // Observable,相当于是stream，流
              .subscribeOn(Schedulers.io())
              .observeOn(AndroidSchedulers.mainThread())
              .subscribe(_getObserver());                             // Observer
    }

    private Observable<Boolean> _getObservable() {
        return Observable.create(new Observable.OnSubscribe<Boolean>() {

            @Override
            public void call(Subscriber<? super Boolean> observer) {

                if(!observer.isUnsubscribed())
                {
                    _log("Within Observable");

                    _doSomeLongOperation_thatBlocksCurrentThread();
                    observer.onNext(true);
                    observer.onCompleted();
                }

            }
        });
    }

    /**
     * Observer that handles the result List<Integer> from Observable
     * through the 3 important actions:
     *
     * 1. onCompleted
     * 2. onError
     * 3. onNext
     */
    private Observer<Boolean> _getObserver() {
        return new Observer<Boolean>() {

            @Override
            public void onCompleted() {
                _log("On complete");
                _progress.setVisibility(View.INVISIBLE);
            }

            @Override
            public void onError(Throwable e) {
                Timber.e(e, "Error in RxJava Demo concurrency");
                _log(String.format("Boo Error %s", e.getMessage()));
                _progress.setVisibility(View.INVISIBLE);
            }

            @Override
            public void onNext(Boolean aBoolean) {
                _log(String.format("onNext with return value "%b"", aBoolean));
            }
        };
    }

    // -----------------------------------------------------------------------------------
    // Method that help wiring up the example (irrelevant to RxJava)

    private void _doSomeLongOperation_thatBlocksCurrentThread() {
        _log("performing long operation");

        try {
            Thread.sleep(3000);

        } catch (InterruptedException e) {
            Timber.d("Operation was interrupted");
        }
    }

    private void _log(String logMsg) {

        if (_isCurrentlyOnMainThread()) {
            _logs.add(0, logMsg + " (main thread) ");
            _adapter.clear();
            _adapter.addAll(_logs);
        } else {
            _logs.add(0, logMsg + " (NOT main thread) ");

            // You can only do below stuff on main thread.
            new Handler(Looper.getMainLooper()).post(new Runnable() {

                @Override
                public void run() {
                    _adapter.clear();
                    _adapter.addAll(_logs);
                }
            });
        }
    }

    private void _setupLogger() {
        _logs = new ArrayList<String>();
        _adapter = new LogAdapter(getActivity(), new ArrayList<String>());
        _logsList.setAdapter(_adapter);
    }

    private boolean _isCurrentlyOnMainThread() {
        return Looper.myLooper() == Looper.getMainLooper();
    }

    private class LogAdapter
          extends ArrayAdapter<String> {

        public LogAdapter(Context context, List<String> logs) {
            super(context, R.layout.item_log, R.id.item_log, logs);
        }
    }
}

　　

参考资料:http://reactivex.io/documentation/operators/create.html
       https://github.com/ReactiveX/RxJava/wiki/The-RxJava-Android-Module