使用场景:

当从服务器请求的数据比较庞大时，这个时候，需要对数据进行分批处理，然后以当前的数据刷新界面

效果图:

image

原理分析图:

image

完整代码如下:

mDataInfo = new DataInfo();
List<Integer> objects = new ArrayList<>();
for (int i = 0; i < 4000; i++) {
    objects.add(i);
}
mDataInfo.l1 = objects;
objects = new ArrayList<>();
for (int i = 0; i < 8000; i++) {
    objects.add(i);
}
mDataInfo.l2 = objects;
objects = new ArrayList<>();
for (int i = 0; i < 12000; i++) {
    objects.add(i);
}

Observable.concat(Observable.just(mDataInfo.l1), Observable.just(mDataInfo.l2), Observable
        .just(mDataInfo.l3)).subscribeOn(Schedulers.io()).
        concatMap(new Function<List<Integer>, Observable<NewPageInfo>>() {
            @Override
            public Observable<NewPageInfo> apply(List<Integer> list1) throws Exception {
                LoggerUtils.loge("concatMap list1.size = " + list1.size());
                List<NewPageInfo> list = new ArrayList<>();
                int pageSize = list1.size() / 1000;
                int start = -1, end = 0;
                List<Integer> tmp = new ArrayList<>(list1);
                NewPageInfo newPageInfo = null;
                String tagStr = "";
                if (list1.size() == mDataInfo.l1.size()) {
                    tagStr = "l1";
                } else if (list1.size() == mDataInfo.l2.size()) {
                    tagStr = "l2";
                } else if (list1.size() == mDataInfo.l3.size()) {
                    tagStr = "l3";
                }
                for (int i = 0; i < pageSize; i++) {
                    // 0 : 0
                    // 1 1000
                    start = i * 1000;
                    end = (i + 1) * 1000 - 1;
                    LoggerUtils.loge("s : " + start + " , e : " + end);
                    List<Integer> subList = tmp.subList(start, end);
                    newPageInfo = new NewPageInfo();
                    newPageInfo.tag = tagStr;
                    newPageInfo.datas = subList;
                    newPageInfo.page = i + "";
                    list.add(newPageInfo);
                }
                ThreadUtils.doSleep(1500);
                return Observable.fromIterable(list);
            }
        })
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Consumer<NewPageInfo>() {
            @Override
            public void accept(NewPageInfo newPageInfo) throws Exception {
                List<Integer> datas = newPageInfo.datas;
                int size = datas.size();
                String msg = "count : " + newPageInfo.page + " , dataSize : " + size + " " +
                        ", " + "scope : start : " + datas.get(0) + " , end : " + datas
                        .get(size
                                - 1);
                LoggerUtils.loge(msg);
                UiUtils.postDelayed(new Runnable() {
                    @Override
                    public void run() {
                        if ("l1".equalsIgnoreCase(newPageInfo.tag)) {
                            line1.append(msg + "\r\n");
                        } else if ("l2".equalsIgnoreCase(newPageInfo.tag)) {
                            line2.append(msg + "\r\n");
                        } else if ("l3".equalsIgnoreCase(newPageInfo.tag)) {
                            line3.append(msg + "\r\n");
                        }
                    }
                }, 800);
            }
        });

场景二

以上的操作，对象为空了，直接就会抛出空指针的异常，流程没办法往下走了。

改进后的操作如下

image

示例代码

private void initData() {
    MPChartInfo mpChartInfo = new MPChartInfo();
    mDataInfo = new DataInfo();
    List<Integer> objects = new ArrayList<>();
    for (int i = 0; i < 1000; i++) {
        objects.add(i);
    }
    mDataInfo.name = "l1";
    mDataInfo.l1 = objects;
    mpChartInfo.V52 = mDataInfo;

    mDataInfo = new DataInfo();
    objects = new ArrayList<>();
    for (int i = 0; i < 2000; i++) {
        objects.add(i);
    }
    mDataInfo.l1 = objects;
    objects = new ArrayList<>();
    for (int i = 0; i < 3000; i++) {
        objects.add(i);
    }
    mDataInfo.l2 = objects;
    mDataInfo.name = "l2";
    mpChartInfo.V54 = mDataInfo;

    mDataInfo = new DataInfo();
    objects = new ArrayList<>();
    for (int i = 0; i < 3000; i++) {
        objects.add(i);
    }
    mDataInfo.l1 = objects;
    objects = new ArrayList<>();
    for (int i = 0; i < 4000; i++) {
        objects.add(i);
    }
    mDataInfo.l2 = objects;
    objects = new ArrayList<>();
    for (int i = 0; i < 5000; i++) {
        objects.add(i);
    }
    mDataInfo.name = "l3";
    mDataInfo.l3 = objects;
    mpChartInfo.V58 = mDataInfo;
    showUI(mpChartInfo);
} 



/**
 * 标识传递,空数据的处理
 */
private void ll_02_03(MPChartInfo mpChartInfo) {
    mV52 = mpChartInfo.getV52();
    mV54 = mpChartInfo.getV54();
    mV54 = null;
    // 对象为空时就使用空的Observable
    Observable empty = Observable.empty();
    mV58 = mpChartInfo.getV58();
    mV58 = null;
    Observable.concat(mV52 != null ? Observable.just(mV52) : empty, mV54 != null ? Observable
            .just(mV54) : empty, mV58 != null ? Observable
            .just(mV58) : empty)
            .subscribeOn(Schedulers.io()).
            concatMap(new Function<DataInfo, Observable<Map<String, List<Integer>>>>() {
                @Override
                public Observable<Map<String, List<Integer>>> apply(DataInfo dataInfo) throws
                        Exception {
                    LoggerUtils.loge("concatMap apply = " + dataInfo.name);
                    ThreadUtils.doSleep(300);
                    List<Integer> l1 = dataInfo.l1;
                    List<Integer> l2 = dataInfo.l2;
                    List<Integer> l3 = dataInfo.l3;
                    // 定义标识，标识一定要往下传递，要区分是刷新哪个控件使用 
                    Map<String, List<Integer>> map = new LinkedHashMap<>();
                    map.put(dataInfo.name, l1);
                    // 对象为空或集合为空时，使用空的Observable
                    Observable<Map<String, List<Integer>>> source1 = l1 != null && l1.size()
                            > 0 ?
                            Observable.just(map) : empty;

                    map = new LinkedHashMap<>();
                    map.put(dataInfo.name, l2);

                    Observable<Map<String, List<Integer>>> source2 = l2 != null && l2.size()
                            > 0 ? Observable
                            .just(map) : empty;

                    map = new LinkedHashMap<>();
                    map.put(dataInfo.name, l3);

                    Observable<Map<String, List<Integer>>> source3 = l3 != null && l3.size()
                            > 0 ?
                            Observable.just(map) :
                            empty;
                    Observable<Map<String, List<Integer>>> concat = Observable.concat
                            (source1, source2,
                                    source3);
                    return concat;
                }
            })
            .subscribeOn(Schedulers.io())
            .concatMap(new Function<Map<String, List<Integer>>, Observable<NewPageInfo>>() {
                @Override
                public Observable<NewPageInfo> apply(Map<String, List<Integer>> list1) throws
                        Exception {
                    Iterator<Map.Entry<String, List<Integer>>> iterator = list1.entrySet()
                            .iterator();
                    Map.Entry<String, List<Integer>> entry = iterator.next();

                    List<Integer> entryValue = entry.getValue();
                    String key = entry.getKey();
                    LoggerUtils.loge("concatMap apply tag = " + key + " , list1.size " +
                            "= " + entryValue.size());
                    List<NewPageInfo> list = new ArrayList<>();
                    int pageSize = entryValue.size() / 1000;
                    int start = -1, end = 0;
                    List<Integer> tmp = new ArrayList<>(entryValue);
                    NewPageInfo newPageInfo = null;
                    String tagStr = key;
                    for (int i = 0; i < pageSize; i++) {
                        // 0 : 0
                        // 1 1000
                        start = i * 1000;
                        end = (i + 1) * 1000 - 1;
//                            LoggerUtils.loge("s : " + start + " , e : " + end);
                        List<Integer> subList = tmp.subList(start, end);
                        newPageInfo = new NewPageInfo();
                        newPageInfo.tag = tagStr;
                        newPageInfo.datas = subList;
                        newPageInfo.page = i + "";
                        list.add(newPageInfo);
                    }
                    ThreadUtils.doSleep(500);
                    return Observable.fromIterable(list);
                }
            }).observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Consumer<NewPageInfo>() {
                @Override
                public void accept(NewPageInfo newPageInfo) throws Exception {
                    List<Integer> datas = newPageInfo.datas;
                    int size = datas.size();
                    String msg = "count : " + newPageInfo.page + " , dataSize : " + size + " " +
                            ", " + "scope : start : " + datas.get(0) + " , end : " + datas
                            .get(size
                                    - 1);
                    LoggerUtils.loge("consumer name = " + newPageInfo.tag + " , dataSize : "
                            + size);
                    UiUtils.postDelayed(new Runnable() {
                        @Override
                        public void run() {
                            // 根据标签刷新控件
                            if ("l1".equalsIgnoreCase(newPageInfo.tag)) {
                                line1.append(msg + "\r\n");
                            } else if ("l2".equalsIgnoreCase(newPageInfo.tag)) {
                                line2.append(msg + "\r\n");
                            } else if ("l3".equalsIgnoreCase(newPageInfo.tag)) {
                                line3.append(msg + "\r\n");
                            }
                        }
                    }, 400);
                }
            });
}