zoukankan      html  css  js  c++  java
  • Rx学习

      RXjava学习资料:
      
      https://www.gitbook.com/book/yuxingxin/rxjava-essentials-cn/details
      
      如下只是学习笔记而已,后面添加实战案例,现在只是理论总结:
      
      Rxjava语言特点:
      
      1,易于并发从而更好的利用服务器的能力;
      
      2,易于有条件的异步执行;
      
      3,一种更好的方式来避免回调地狱;
      
      4,一种响应式方法。
      
      RXjava源于观察者模式:
      
      添加了如下三个缺少的功能:
      
      1,生产者在没有更多数据可用时能够发出信号通知:oncompleted()事件。
      
      2,生产者在发生错误的时候能够发出信号通知:onError()事件。
      
      3,RxJavaObservables能够组合而不是嵌套,从而避免开发者陷入回调地狱。
      
      RXjava中的四个角色:
      
      1,Observable:观察得到的,看的见得
      
      2,Observer:观察者
      
      3,Subscriber:订阅者
      
      4,Subjects:服从。
      
      Observables和Subjects是两个“生产”实体
      
      Observers和Subscribers是两个“消费”实体。
      
      Rxjava之Observable
      
      onNext(T)检索数据
      
      onError(Throwable)发现错误
      
      onCompleted()完成
      
      RXjava之热Observables和冷Observables
      
      热Observables:只要创建完就发射数据,订阅他的观察者从序列中某位置接受数据。
      
      冷Observables:一直等待,知道观察者订阅他才发射数据,可确保能收到数据序列。

    Observale创建的两种方式:

    1:create

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

    2:from:从列表,数组来创建Observable,并一个个发射数据

    List<Integer> items = new ArrayList<Integer>();
    items.add(1);
    Observable<Integer> observableString = Observable.from(items);
    Subscription subscriptionPrint = observableString.subscribe(new Observer<Integer>() {
        @Override
        public void onCompleted() {
            System.out.println("Observable completed");
        }
        @Override
        public void onError(Throwable e) {
            System.out.println("Oh,no! Something wrong happened!");
        }
        @Override
        public void onNext(Integer item) {
            System.out.println("Item is " + item);
        }
    }); 

      3,just

      just()方法可以传入一到九个参数,它们会按照传入的参数的顺序来发射它

      们。

      无理由不发射数据并结束操作:

      Observable.empty(),Observable.never(),和

      Observable.throw()

      RxJava提供四种不同的Subject:

      PublishSubject

      BehaviorSubject:向订阅者发送截止订阅前最新的一个数据对象,然后发送订阅后数据流;

      ReplaySubject:缓存所订阅的所有数据,向任意一个订阅他的观察者重发数据流;

      AsyncSubject:当Observable完成AnsySubject只会发布最后一个数据给已定订阅的每一个观察者。

      as中开发使用RXjava,RXandroid:

      gradle依赖:

    compile 'io.reactivex:rxandroid:1.1.0'
    compile 'io.reactivex:rxjava:1.1.0'

      其他推荐的as插件:

      1,lombok:getsettostringequal等注解代码

      2,butterknife:findviewbyidonclick代码等的注解代码

      3,retrolambda:java8lambda相关函数

      具体作用情百度或者谷歌进行搜索

      just主要是得到原始的observable版本,在一个新的响应式架构的基础上迁移已存在的代码,这个方法可

      能是一个有用的开始点。

    private void loadApps(AppInfo appOne, AppInfo appTwo, AppInfo appThree)
    mRecyclerView.setVisibility(View.VISIBLE);
    Observable.just(appOne, appTwo, appThree)
    .subscribe(new Observer<AppInfo>() {
        @Override
        public void onCompleted() {
            mSwipeRefreshLayout.setRefreshing(false);
            Toast.makeText(getActivity(), "Here is the list!"
        }
        @Override
        public void onError(Throwable e) {
            Toast.makeText(getActivity(), "Something went wrong!"
                           mSwipeRefreshLayout.setRefreshing(false);
        }
        @Override
        public void onNext(AppInfo appInfo) {
            mAddedApps.add(appInfo);
            mAdapter.addApplication(mAddedApps.size() - 1, appInfo);
        }
    });
    }

      repeat()重复发数据:

      repeat(3):重复发送三次数据;

    private void loadApps(AppInfo appOne, AppInfo appTwo, AppInfo appThree)
    mRecyclerView.setVisibility(View.VISIBLE);
    Observable.just(appOne, appTwo, appThree)
    .repeat(3)
    .subscribe(new Observer<AppInfo>() {
        @Override
        public void onCompleted() {
            mSwipeRefreshLayout.setRefreshing(false);
            Toast.makeText(getActivity(), "Here is the list!"
        }
        @Override
        public void onError(Throwable e) {
            Toast.makeText(getActivity(), "Something went wrong!"
                           mSwipeRefreshLayout.setRefreshing(false);
        }
        @Override
        public void onNext(AppInfo appInfo) {
            mAddedApps.add(appInfo);
            mAdapter.addApplication(mAddedApps.size() - 1, appInfo);
        }
    });
    }

    defer()声明一个Observable但是你又想推迟这个Observable的创建直到观察者订阅

    1
    private Observable<Integer> getInt() {
        return Observable.create(subscriber -> {
            if(subscriber.isUnsubscribed()) {
                return;
            }
            App.L.debug("GETINT");
            subscriber.onNext(42);
            subscriber.onCompleted();
        });
    }
    
    2
    Observable<Integer> deferred = Observable.defer(this::getInt);
    
    3
    deferred.subscribe(number -> {
        App.L.debug(String.valueOf(number));
    });

    range()
    你需要从一个指定的数字X开始发射N个数字吗?你可以用 range,下面是从10开始,发送3个数字:

    Observable.range(10, 3)
    .subscribe(new Observer<Integer>() {
        @Override
        public void onCompleted() {
            Toast.makeText(getActivity(), "Yeaaah!", Toast.LENGTH_LONG).show();
        }
        @Override
        public void onError(Throwable e) {
            Toast.makeText(getActivity(), "Something went wrong!", Toast.LENGTH_SHORT).show();
        }
        @Override
        public void onNext(Integer number) {
            Toast.makeText(getActivity(), "I say " + number, Toast.LENGTH_SHORT).show();
        }
    });

    interval()
    interval() 函数在你需要创建一个轮询程序时非常好用。

    如下每隔3秒toast下数据:

    Subscription stopMePlease = Observable.interval(3, TimeUnit.SECONDS)
    .subscribe(new Observer<Integer>() {
        @Override
        public void onCompleted() {
            Toast.makeText(getActivity(), "Yeaaah!", Toast.LENGTH_LONG).show();
        }
        @Override
        public void onError(Throwable e) {
            Toast.makeText(getActivity(), "Something went wrong!", Toast.LENGTH_SHORT).show();
        }
        @Override
        public void onNext(Integer number) {
            Toast.makeText(getActivity(), "I say " + number, Toast.LENGTH_SHORT).show();
        }
    });

      timer()

      如果你需要一个一段时间之后才发射的Observable,你可以像下面的例子使用timer():

      如下3秒后发送数据:

    Observable.timer(3, TimeUnit.SECONDS)
    .subscribe(new Observer<Long>() {
        @Override
        public void onCompleted() {
        }
        @Override
        public void onError(Throwable e) {
        }
        @Override
        public void onNext(Long number) {
            Log.d("RXJAVA", "I say " + number);
        }
    });

      如何过滤数据 Observables?
      
      RxJava让我们使用 filter() 方法来过滤我们观测序列中不想要的值,filter((appInfo) ->appInfo.getName().startsWith("C"))是只想展示c开头的数据。

    1,过滤序列:

    private void loadList(List<AppInfo> apps) {
        mRecyclerView.setVisibility(View.VISIBLE);
        Observable.from(apps)
        .filter((appInfo) ->
                appInfo.getName().startsWith("C"))
        .subscribe(new Observer<AppInfo>() {
            @Override
            public void onCompleted() {
                mSwipeRefreshLayout.setRefreshing(false);
            }
            @Override
            public void onError(Throwable e) {
                Toast.makeText(getActivity(), "Something went wrong!"
                               mSwipeRefreshLayout.setRefreshing(false);
            }
            @Override
            public void onNext(AppInfo appInfo) {
                mAddedApps.add(appInfo);
                mAdapter.addApplication(mAddedApps.size() - 1, appInfo);
            }
        });
    }

      过滤空数据:

    .filter(new Func1<AppInfo, Boolean>() {
        @Override
        public Boolean call(AppInfo appInfo) {
            return appInfo != null;
        }
    })

      获取开头或者结尾的几个数据:take() 或 takeLast()。
      
      take前几个数据;
      
      takelast后几个数据。
      
      如下是获取前三个数据:

    private void loadList(List<AppInfo> apps) {
        mRecyclerView.setVisibility(View.VISIBLE);
        Observable.from(apps)
        .take(3)
        .subscribe(new Observer<AppInfo>() {
            @Override
            public void onCompleted() {
                mSwipeRefreshLayout.setRefreshing(false);
            }
            @Override
            public void onError(Throwable e) {
                Toast.makeText(getActivity(), "Something went wrong!"
                               mSwipeRefreshLayout.setRefreshing(false);
            }
            @Override
            public void onNext(AppInfo appInfo) {
                mAddedApps.add(appInfo);
                mAdapter.addApplication(mAddedApps.size() - 1, appInfo);
            }
        });
    }

      获取后三个数据:

    Observable.from(apps)
    .takeLast(3)
    .subscribe(...);

      数据去重Distinct:

      制造重复数据:

    Observable<AppInfo> fullOfDuplicates = Observable.from(apps)
                                           .take(3)
                                           .repeat(3);

      去重:

      

    fullOfDuplicates.distinct()
    .subscribe(new Observer<AppInfo>() {
        @Override
        public void onCompleted() {
            mSwipeRefreshLayout.setRefreshing(false);
        }
        @Override
        public void onError(Throwable e) {
            Toast.makeText(getActivity(), "Something went wrong!"
                           mSwipeRefreshLayout.setRefreshing(false);
        }
        @Override
        public void onNext(AppInfo appInfo) {
            mAddedApps.add(appInfo);
            mAdapter.addApplication(mAddedApps.size() - 1, appInfo);
        }
    });
    }

     忽略掉重复的值并且在温度确实改变的时候才得到通知:ditinctuntilchanged()

    也就是说有重复数据的时候,利用ditinctuntilchanged可以去重并展示重复数据中的一个数据。

    发射第一个数据 first()

    发射最后一个数据last()

    观测序列完成,不在发射任何值得时候,firstOrDefault(),lastOrDefault()

    跳过前两个数据 skip(2)

    跳过后两个数据 skipLast(2),也就是跳过后两个的其他数据发射。

    发射指定位置的元素:elementAt(2),从零开始,第三个元素发射

    指定时间间隔由Observable发射最近一次的数值:

    如下每隔三秒,updateDisplay(currentTemperature) 更新当前温度信息

    Observable<Integer> sensor = [...]
                                 sensor.sample(30, TimeUnit.SECONDS)
    .subscribe(new Observer<Integer>() {
        @Override
        public void onCompleted() {
        }
        @Override
        public void onError(Throwable e) {
        }
        @Override
        public void onNext(Integer currentTemperature) {
            updateDisplay(currentTemperature)
        }
    });

     timeout:  timeout() 为一个Observable的限时的副本,如果在指定的时间间隔内Observable不发射值的话,它监听的原始的Observable时就会触发 onError() 函
    数,也就是限定时间的字段:

    下面代码意思就是每隔两秒刷新当前温度,如果不发射数据,就会打印:You should go check the sensor, dude。

    Subscription subscription = getCurrentTemperature()
                                .timeout(2, TimeUnit.SECONDS)
    .subscribe(new Observer<Integer>() {
        @Override
        public void onCompleted() {
        }
        @Override
        public void onError(Throwable e) {
            Log.d("RXJAVA", "You should go check the sensor, dude");
        }
        @Override
        public void onNext(Integer currentTemperature) {
            updateDisplay(currentTemperature)
        }
    });

    Debounce

    过滤掉由Observable发射的速率过快的数据;如果在一个指定的时间间隔过去了仍旧没有发射一个,那么它将发射最后的那个。

     2016年4月6日23:02:38

    转换Observables

    Map() :将函数作用于每一个发出去的元素上面去,以此创建一个新的Observable来发射转换的数据。

    如下就是将原来的apps的名称小写,并发出去。

    private void loadList(List<AppInfo> apps) {
        mRecyclerView.setVisibility(View.VISIBLE);
        Observable.from(apps)
        .map(new Func1<AppInfo, AppInfo>() {
            @Override
            public Appinfo call(AppInfo appInfo) {
                String currentName = appInfo.getName();
                String lowerCaseName = currentName.toLowerCase();
                appInfo.setName(lowerCaseName);
                return appInfo;
            }
        })
        .subscribe(new Observer<AppInfo>() {
            @Override
            public void onCompleted() {
                mSwipeRefreshLayout.setRefreshing(false);
            }
            @Override
            public void onError(Throwable e) {
                Toast.makeText(getActivity(), "Something went wrong!"
                               mSwipeRefreshLayout.setRefreshing(false);
            }
            @Override
            public void onNext(AppInfo appInfo) {
                mAddedApps.add(appInfo);
                mAdapter.addApplication(mAddedApps.size() - 1, appInfo);
            }
        });
    }

    FlatMap RxJava的 flatMap() 函数提供一种铺平序列的方式,然后合并这些Observables发射的数据,最后将合并后的结果作为最终的Observable。

    ConcatMap:解决了 flatMap() 的交叉问题,提供了一种能够把发射的值连续在一起的铺平函数,而不是合并它们。

    FlatMapIterable:flatMapInterable() 和 flatMap() 很像。仅有的本质不同是它将源数据两两结成对并生成Iterable,而不是原始数据项和生成的Observables。

    SwitchMap:switchMap() 和 flatMap() 很像,除了一点:每当源Observable发射一个新的数据项(Observable)时,它将取消订阅并停止监视之前那个数据项

    产生的Observable,并开始监视当前发射的这一个。

    Scan:scan() 函数对原始

    Observable发射的每一项数据都应用一个函数,计算出函数的结果值。

    Observable.just(1, 2, 3, 4, 5)
    .scan((sum, item) -> sum + item)
    .subscribe(new Subscriber<Integer>() {
        @Override
        public void onCompleted() {
            Log.d("RXJAVA", "Sequence completed.");
        }
        @Override
        public void onError(Throwable e) {
            Log.e("RXJAVA", "Something went south!");
        }
        @Override
        public void onNext(Integer item) {
            Log.d("RXJAVA", "item is: " + item);
        }
    });

    groupBy()  来分组元素:

    如下按照根据最近更新时间来进行分组。

    Observable<GroupedObservable<String, AppInfo>> groupedItems = Observable.from(apps)
    .groupBy(new Func1<AppInfo, String>() {
        @Override
        public String call(AppInfo appInfo) {
            SimpleDateFormat formatter = new SimpleDateFormat("MM/yyyy"
                    return formatter.format(new Date(appInfo.getLastUpdateTime()));
        }
    });
    Observable.concat(groupedItems)
    .subscribe(new Observer<AppInfo>() {
        @Override
        public void onCompleted() {
            mSwipeRefreshLayout.setRefreshing(false);
        }
        @Override
        public void onError(Throwable e) {
            Toast.makeText(getActivity(), "Something went wrong!", Toast.LENGTH_SHORT).show();
            mSwipeRefreshLayout.setRefreshing(false);
        }
        @Override
        public void onNext(AppInfo appInfo) {
            mAddedApps.add(appInfo);
            mAdapter.addApplication(mAddedApps.size() - 1, appInfo);
        }
    });

      Buffer RxJava中的 buffer() 函数将源Observable变换一个新的Observable,这个新的Observable每次发射一组列表值而不是一个一个发射。
      buffer(count = 3)一组三个的方式进行发射一个列表的obserable。
      buffer(count = 2,skip = 2)一组两个,跳过第三个来进行发射一个列表的obserable。
      buffer(timespan = 2,count = 2)每隔2秒,每组2个发射一个列表的obserable.
      Window:RxJava的 window() 函数和 buffer() 很像,但是它发射的是Observable而不是列表。
      window(count = 3)原来observable的自己,每隔obserable三个元素。
      window(skip = 3,count = 2),每隔第三个,每隔obserable中有2个元素。
      Cast:
      map的特殊版本。将原observale每一项转换为新的类型,把他变成不同class.


    组合observable

     2016年4月7日22:23:38

     Merge:

    从单词角度分析,合并,rx中将帮助你把两个甚至更多的Observables合并到他们发射的数据项里

    private void loadList(List<AppInfo> apps) {
        mRecyclerView.setVisibility(View.VISIBLE);
        List reversedApps = Lists.reverse(apps);
        Observable<AppInfo> observableApps = Observable.from(apps);
        Observable<AppInfo> observableReversedApps = Observable.from(reversedApps);
    //合并操作 Observable<AppInfo> mergedObserbable =
    Observable.merge(observableApps, observableReversedApps); mergedObserbable.subscribe(new Observer<AppInfo>() { @Override public void onCompleted() { mSwipeRefreshLayout.setRefreshing(false); Toast.makeText(getActivity(), "Here is the list!", Toast.LENGTH_LONG).show(); } @Override public void onError(Throwable e) { Toast.makeText(getActivity(), "One of the two Observable threw an error!" mSwipeRefreshLayout.setRefreshing(false); } @Override public void onNext(AppInfoappInfo) { mAddedApps.add(appInfo); mAdapter.addApplication(mAddedApps.size() - 1, appInfo); } }); }

    Zip操作:

    zip() 合并两个或者多个Observables发射出的数据项,根据指定的函数 Func* 变换它们,并发射一个新值

    private void loadList(List<AppInfo> apps) {
        mRecyclerView.setVisibility(View.VISIBLE);
        Observable<AppInfo> observableApp = Observable.from(apps);
        Observable<Long> tictoc = Observable.interval(1, TimeUnit.SECONDS);
        Observable.zip(observableApp, tictoc,
                       (AppInfo appInfo, Long time) -> updateTitle(appInfo, time))
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Observer<AppInfo>() {
            @Override
            public void onCompleted() {
                Toast.makeText(getActivity(), "Here is the list!", Toast.LENGTH_LONG).show();
            }
            @Override
            public void onError(Throwable e) {
                mSwipeRefreshLayout.setRefreshing(false);
                Toast.makeText(getActivity(), "Something went wrong!", Toast.LENGTH_SHORT).show();
            }
            @Override
            public void onNext(AppInfoappInfo) {
                if (mSwipeRefreshLayout.isRefreshing()) {
                    mSwipeRefreshLayout.setRefreshing(false);
                }
                mAddedApps.add(appInfo);
                int position = mAddedApps.size() - 1;
                mAdapter.addApplication(position, appInfo);
                mRecyclerView.smoothScrollToPosition(position);
            }
        });
    }

      Join操作:
      RxJava的 join() 函数基于时间窗口将两个Observables发射的数据结合在一起。
      如下意思:tictoc 这个Observable数据每秒只发射一个新的 Long 型整数。
      为了合并它们,我们需要指定两个 Func1 变量:
      appInfo -> Observable.timer(2, TimeUnit.SECONDS)
      time -> Observable.timer(0, TimeUnit.SECONDS)
      上面描述了两个时间窗口。下面一行描述我们如何使用 Func2 将两个发射的数据
      结合在一起。
      this::updateTitle

    private void loadList(List<AppInfo> apps) {
        mRecyclerView.setVisibility(View.VISIBLE);
        Observable<AppInfo> appsSequence =
            Observable.interval(1000, TimeUnit.MILLISECONDS)
        .map(position -> {
            return apps.get(position.intValue());
        });
        Observable<Long> tictoc = Observable.interval(1000, TimeUnit.MILLISECONDS);
        appsSequence.join(
            tictoc,
            appInfo -> Observable.timer(2, TimeUnit.SECONDS),
            time -> Observable.timer(0, TimeUnit.SECONDS),
            this::updateTitle)
        .observeOn(AndroidSchedulers.mainThread())
        .take(10)
        .subscribe(new Observer<AppInfo>() {
            @Override
            public void onCompleted() {
                Toast.makeText(getActivity(), "Here is the list!"
            }
            @Override
            public void onError(Throwable e) {
                mSwipeRefreshLayout.setRefreshing(false);
                Toast.makeText(getActivity(), "Something went wrong!"
            }
            @Override
            public void onNext(AppInfoappInfo) {
                if (mSwipeRefreshLayout.isRefreshing()) {
                    mSwipeRefreshLayout.setRefreshing(false);
                }
                mAddedApps.add(appInfo);
                int position = mAddedApps.size() - 1;
                mAdapter.addApplication(position, appInfo);
                mRecyclerView.smoothScrollToPosition(position);
            }
        });
    }

      combineLatest操作:
      zip() 作用于最近未打包的两个Observables。相反, combineLatest() 作用于最近发射的数据项:如果 Observable1 发射了A并且 Observable2 发射了B和C, combineLatest() 将会分组处理AB和AC。
      如下:
      一个是每秒钟从我们已安装的应用列表发射一个App数据,第二个是每隔1.5秒发射一个 Long 型整数。我们将他们结合起来并执行 updateTitle() 函数。

    private void loadList(List<AppInfo> apps) {
        mRecyclerView.setVisibility(View.VISIBLE);
        Observable<AppInfo> appsSequence = Observable.interval(1000, TimeUnit.MILLISECONDS)
                                           .map(position ->apps.get(position.intValue()));
        Observable<Long> tictoc = Observable.interval(1500, TimeUnit.MILLISECONDS);
        Observable.combineLatest(appsSequence, tictoc,
                                 this::updateTitle)
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Observer<AppInfo>() {
            @Override
            public void onCompleted() {
                Toast.makeText(getActivity(), "Here is the list!", Toast.LENGTH_LONG).show();
            }
            @Override
            public void onError(Throwable e) {
                mSwipeRefreshLayout.setRefreshing(false);
                Toast.makeText(getActivity(), "Something went wrong!", Toast.LENGTH_SHORT).show();
            }
            @Override
            public void onNext(AppInfoappInfo) {
                if (mSwipeRefreshLayout.isRefreshing()) {
                    mSwipeRefreshLayout.setRefreshing(false);
                }
                mAddedApps.add(appInfo);
                int position = mAddedApps.size() - 1;
                mAdapter.addApplication(position, appInfo);
                mRecyclerView.smoothScrollToPosition(position);
            }
        });
    }

    And,Then和When:

      我们有两个发射的序列, observableApp ,发射我们安装的应用列表数据, tictoc 每秒发射一个 Long 型整数。现在我们用 and() 连接源
      Observable和第二个Observable。

    private void loadList(List<AppInfo> apps) {
        mRecyclerView.setVisibility(View.VISIBLE);
        Observable<AppInfo> observableApp = Observable.from(apps);
        Observable<Long> tictoc = Observable.interval(1, TimeUnit.SECONDS);
        Pattern2<AppInfo, Long> pattern = JoinObservable.from(observableApp).and(tictoc);
        Plan0<AppInfo> plan = pattern.then(this::updateTitle);
        JoinObservable
        .when(plan)
        .toObservable()
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Observer<AppInfo>() {
            @Override
            public void onCompleted() {
                Toast.makeText(getActivity(), "Here is the list!", Toast.LENGTH_LONG).show();
            }
            @Override
            public void onError(Throwable e) {
                mSwipeRefreshLayout.setRefreshing(false);
                Toast.makeText(getActivity(), "Something went wrong!"
            }
            @Override
            public void onNext(AppInfoappInfo) {
                if (mSwipeRefreshLayout.isRefreshing()) {
                    mSwipeRefreshLayout.setRefreshing(false);
                }
                mAddedApps.add(appInfo);
                int position = mAddedApps.size() - 1;
                mAdapter.addApplication(position, appInfo);
                mRecyclerView.smoothScrollToPosition(position);
            }
        });
    }

    Switch:

      subscribe-unsubscribe 的序列里我们能够从一个Observable自动取消订阅来订阅一个新的Observable。

      将一个发射多个Observables的Observable转换成另一个单独的Observable,后者发射那些Observables最近发射的数据项。

    StartWith:

      startWith() 是 concat() 的对应部分。正如 concat() 向发射数据的Observable追加数据那样,在Observable开始发射他们的数据之前,
      startWith() 通过传递一个参数来先发射一个数据序列。

    Schedulers调度器

     处理多线程异步操作:

      RxJava提供了5种调度器:
      .io() :IO操作,增长缩减自适应线程池(线程池无限制
      .computation:计算工作默认的调度器,调度器:buffer(),debounce(),delay(),interval(),sample,skip()。
      .immediate():允许在当前线程执行你指定的工作,是timeout(),timeInterval(),timestamp()默认调度器。
      .newThread():为指定任务启动一个线程。
      .trampoline():任务入队列,按需处理队列中的任务,是repeat()和retry()默认调度器。

    如下就是简单将对图片文件的IO操作通过RX的io调度器来线程池中进行处理。

    public static void storeBitmap(Context context, Bitmap bitmap, String filename)
    Schedulers.io().createWorker().schedule(() -> {
        blockingStoreBitmap(context, bitmap, filename);
    });
    }
    private static void blockingStoreBitmap(Context context, Bitmap bitmap, String filename)
    FileOutputStream fOut = null;
    try {
        fOut = context.openFileOutput(filename, Context.MODE_PRIVATE);
        bitmap.compress(Bitmap.CompressFormat.PNG, 100, fOut);
        fOut.flush();
        fOut.close();
    } catch (Exception e) {
        throw new RuntimeException(e);
    } finally {
        try {
            if (fOut != null) {
                fOut.close();
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
    }

    在主线程中返回结果。

    onBackpressureBuffer()  方法将告诉Observable发射的数据如果比观察者消费的数据要更快的话,它必须把它们存储在缓存中并提供一个合适的时间给它们。

    getApps()
    .onBackpressureBuffer()
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Observer<AppInfo>() {
        [...]

    处理耗时的任务:

    getObservableApps(apps)
    .onBackpressureBuffer()
    .subscribeOn(Schedulers.computation())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Observer<AppInfo>() {
        [...]
  • 相关阅读:
    python使用multiprocessing进行多进程编程(1)
    python使用multiprocessing进行多进程编程(1)
    最佳日志实践
    最佳日志实践
    最佳日志实践
    以大多数人的努力程度之低,根本轮不到去拼天赋
    以大多数人的努力程度之低,根本轮不到去拼天赋
    以大多数人的努力程度之低,根本轮不到去拼天赋
    C++Socket编程总结 [转]
    C++多线程的几个重要方法解析CreateEvent / SetEvent /ResetEvent/ 等
  • 原文地址:https://www.cnblogs.com/androidsuperman/p/5353363.html
Copyright © 2011-2022 走看看