RxJava 变换操作符 map flatMap concatMap buffer

时间：2018-09-21 23:09:23 阅读：216 评论：0 收藏：0 [点我收藏+]

标签：next 完全 lov complete 模拟 ids z-index 使用回调

demo地址：https://github.com/baiqiantao/RxJavaDemo.git

常用的变换操作符

map：【数据类型转换】将被观察者发送的事件转换为另一种类型的事件
flatMap：【化解循环嵌套和接口嵌套】将被观察者发送的事件序列进行拆分 & 转换后合并成一个新的事件序列，最后再进行发送
concatMap：【有序】与 flatMap 的区别在于，拆分 & 重新合并生成的事件序列的顺序与被观察者旧序列生产的顺序一致
flatMapIterable：相当于对 flatMap 的数据进行了二次扁平化
buffer：定期从被观察者发送的事件中获取一定数量的事件并放到缓存区中，然后把这些数据集合打包发射

map

Observable.just(new Date()) // Date 类型
      .map(Date::getTime) // long 类型
      .map(time -> time + 1000 * 60 * 60)// 改变 long 类型时间的值
      .map(time -> new SimpleDateFormat("HH:mm:ss", Locale.getDefault()).format(new Date(time))) //String 类型
      .subscribe(this::log);

 
Observable.just(new Date()) // Date 类型
      .map(Date::getTime) // long 类型
      .map(time -> time + 1000 * 60 * 60)// 改变 long 类型时间的值
      .map(time -> new SimpleDateFormat("HH:mm:ss", Locale.getDefault()).format(new Date(time))) //String 类型
      .subscribe(this::log);

flatMap concatMap flatMapIterable

基础用法：化解循环嵌套

flatMap 使用一个指定的函数对原始 Observable 发射的每一项数据之行相应的变换操作，这个函数返回一个本身也发射数据的 Observable，然后 flatMap 合并这些 Observables 发射的数据，最后将合并后的结果当做它自己的数据序列发射。

Observable.just(new Person(Arrays.asList("篮球", "足球", "排球")), new Person(Arrays.asList("画画", "跳舞")))
      .map(person -> person.loves)
      .flatMap(Observable::fromIterable) //fromIterable：逐个发送集合中的元素
      .subscribe(this::log);

 
Observable.just(new Person(Arrays.asList("篮球", "足球", "排球")), new Person(Arrays.asList("画画", "跳舞")))
      .map(person -> person.loves)
      .flatMap(Observable::fromIterable) //fromIterable：逐个发送集合中的元素
      .subscribe(this::log);

篮球，22:56:43 009，true
足球，22:56:43 010，true
排球，22:56:43 010，true
画画，22:56:43 011，true
跳舞，22:56:43 012，true

 
篮球，22:56:43 009，true
足球，22:56:43 010，true
排球，22:56:43 010，true
画画，22:56:43 011，true
跳舞，22:56:43 012，true

flatMap() 执行的过程：

使用传入的事件对象创建一个 Observable 对象；
并不发送这个 Observable，而是将它激活，于是它开始发送事件；
每一个创建出来的 Observable 发送的事件，都被汇入同一个 Observable，而这个 Observable 负责将这些事件统一交给 Observer 的回调方法。

注意：如果任何一个通过这个 flatMap 操作产生的单独的 Observable 调用 onError 异常终止了，这个 Observable 自身会立即调用 onError 并终止。例如：

Observable.just(new Person(Arrays.asList("篮球", null, "排球")), new Person(Arrays.asList("画画", "跳舞")))
      .map(person -> person.loves)
      .flatMap(Observable::fromIterable) //fromIterable：逐个发送集合中的元素
      .subscribe(this::log, e -> log("onError:" + e.getMessage()), () -> log("onComplete"));

 
Observable.just(new Person(Arrays.asList("篮球", null, "排球")), new Person(Arrays.asList("画画", "跳舞")))
      .map(person -> person.loves)
      .flatMap(Observable::fromIterable) //fromIterable：逐个发送集合中的元素
      .subscribe(this::log, e -> log("onError:" + e.getMessage()), () -> log("onComplete"));

篮球，00:20:14 762，true
onError:The iterator returned a null value，00:20:14 767，true

 
篮球，00:20:14 762，true
onError:The iterator returned a null value，00:20:14 767，true

flatMap 和 concatMap

concatMap 操作符的功能和 flatMap 非常相似，只不过经过 flatMap 操作变换后，最后输出的序列有可能是交错的(flatMap最后合并结果采用的是 merge 操作符)，而 concatMap 最终输出的数据序列和原数据序列是一致的。

long start = System.currentTimeMillis();
Observable.just(Arrays.asList(1, 2, 3), Arrays.asList(4, 5))
      .flatMap(list -> Observable.fromIterable(list).delay(list.size(), TimeUnit.SECONDS))//flatMap是无序的
      .subscribe((s -> log("f:" + s)), e -> log("f"), () -> log("f耗时" + (System.currentTimeMillis() - start))); //3秒
Observable.just(Arrays.asList("A", "B", "C"), Arrays.asList("D", "E"))
      .concatMap(list -> Observable.fromIterable(list).delay(list.size(), TimeUnit.SECONDS))//concatMap是有序的
      .subscribe(s -> log("c:" + s), e -> log("c"), () -> log("c耗时" + (System.currentTimeMillis() - start))); //5秒

 
long start = System.currentTimeMillis();
Observable.just(Arrays.asList(1, 2, 3), Arrays.asList(4, 5))
      .flatMap(list -> Observable.fromIterable(list).delay(list.size(), TimeUnit.SECONDS))//flatMap是无序的
      .subscribe((s -> log("f:" + s)), e -> log("f"), () -> log("f耗时" + (System.currentTimeMillis() - start))); //3秒
Observable.just(Arrays.asList("A", "B", "C"), Arrays.asList("D", "E"))
      .concatMap(list -> Observable.fromIterable(list).delay(list.size(), TimeUnit.SECONDS))//concatMap是有序的
      .subscribe(s -> log("c:" + s), e -> log("c"), () -> log("c耗时" + (System.currentTimeMillis() - start))); //5秒

f:4，23:21:07 944，false   //flatMap后，订阅者首先接收到的事件是【4】而不是【1】
f:5，23:21:07 945，false
f:1，23:21:08 942，false
f:2，23:21:08 943，false
f:3，23:21:08 943，false
f耗时3025，23:21:08 945，false    //flatMap耗时3秒

c:A，23:21:08 949，false   //concatMap后，订阅者首先接收到的事件是【1】
c:B，23:21:08 950，false
c:C，23:21:08 950，false
c:D，23:21:10 953，false
c:E，23:21:10 953，false
c耗时5034，23:21:10 954，false    //concatMap耗时5秒

 
f:4，23:21:07 944，false   //flatMap后，订阅者首先接收到的事件是【4】而不是【1】
f:5，23:21:07 945，false
f:1，23:21:08 942，false
f:2，23:21:08 943，false
f:3，23:21:08 943，false
f耗时3025，23:21:08 945，false    //flatMap耗时3秒
c:A，23:21:08 949，false   //concatMap后，订阅者首先接收到的事件是【1】
c:B，23:21:08 950，false
c:C，23:21:08 950，false
c:D，23:21:10 953，false
c:E，23:21:10 953，false
c耗时5034，23:21:10 954，false    //concatMap耗时5秒

扩展用法：化解接口嵌套

可以利用 flatMap 操作符实现网络请求依次依赖，即：第一个接口的返回值包含第二个接口请求需要用到的数据。

首先是两个请求网络的操作：

private Observable<String> firstRequest(String parameter) {
   return Observable.create(emitter -> {
      SystemClock.sleep(2000);//模拟网络请求
      emitter.onNext(parameter + "，第一次修改：" + FORMAT.format(new Date(System.currentTimeMillis())));
      emitter.onComplete();
   });
}

 
private Observable<String> firstRequest(String parameter) {
   return Observable.create(emitter -> {
      SystemClock.sleep(2000);//模拟网络请求
      emitter.onNext(parameter + "，第一次修改：" + FORMAT.format(new Date(System.currentTimeMillis())));
      emitter.onComplete();
   });
}

private Observable<String> secondRequest(String parameter) {
   return Observable.create(emitter -> {
      SystemClock.sleep(3000);//模拟网络请求
      emitter.onNext(parameter + "，第二次修改：" + FORMAT.format(new Date(System.currentTimeMillis())));
      emitter.onComplete();
   });
}

 
private Observable<String> secondRequest(String parameter) {
   return Observable.create(emitter -> {
      SystemClock.sleep(3000);//模拟网络请求
      emitter.onNext(parameter + "，第二次修改：" + FORMAT.format(new Date(System.currentTimeMillis())));
      emitter.onComplete();
   });
}

然后可以通过 flatMap 将两者串联起来：

firstRequest("原始值：" + FORMAT.format(new Date(System.currentTimeMillis())))
      .subscribeOn(Schedulers.io()) // 在io线程进行网络请求
      .observeOn(AndroidSchedulers.mainThread()) // 在主线程处理请求结果
      .doOnNext(response -> log("【第一个网络请求结束，响应为】" + response))//true
      .observeOn(Schedulers.io()) // 回到 io 线程去处理下一个网络请求
      .flatMap(this::secondRequest)//实现多个网络请求依次依赖
      .observeOn(AndroidSchedulers.mainThread()) // 在主线程处理请求结果
      .subscribe(string -> log("【第二个网络请求结束，响应为】" + string));//true，5 秒

 
firstRequest("原始值：" + FORMAT.format(new Date(System.currentTimeMillis())))
      .subscribeOn(Schedulers.io()) // 在io线程进行网络请求
      .observeOn(AndroidSchedulers.mainThread()) // 在主线程处理请求结果
      .doOnNext(response -> log("【第一个网络请求结束，响应为】" + response))//true
      .observeOn(Schedulers.io()) // 回到 io 线程去处理下一个网络请求
      .flatMap(this::secondRequest)//实现多个网络请求依次依赖
      .observeOn(AndroidSchedulers.mainThread()) // 在主线程处理请求结果
      .subscribe(string -> log("【第二个网络请求结束，响应为】" + string));//true，5 秒

打印结果为：

【第一个网络请求结束，响应为】原始值：23:58:11 220，第一次修改：23:58:13 245，true
【第二个网络请求结束，响应为】原始值：23:58:11 220，第一次修改：23:58:13 245，第二次修改：23:58:16 256，true

 
【第一个网络请求结束，响应为】原始值：23:58:11 220，第一次修改：23:58:13 245，true
【第二个网络请求结束，响应为】原始值：23:58:11 220，第一次修改：23:58:13 245，第二次修改：23:58:16 256，true

简化形式的Demo代码为：

Observable.just("包青天").delay(1000, TimeUnit.MILLISECONDS) //第一个网络请求，返回姓名
      .flatMap(s -> Observable.just(s + "，男").delay(1000, TimeUnit.MILLISECONDS)) //第二个网络请求，返回性别
      .flatMap(s -> Observable.just(s + "，28岁").delay(1000, TimeUnit.MILLISECONDS)) //第三个网络请求，返回年龄
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(this::log); //包青天，男，28岁，耗时:3058毫秒，true

 
Observable.just("包青天").delay(1000, TimeUnit.MILLISECONDS) //第一个网络请求，返回姓名
      .flatMap(s -> Observable.just(s + "，男").delay(1000, TimeUnit.MILLISECONDS)) //第二个网络请求，返回性别
      .flatMap(s -> Observable.just(s + "，28岁").delay(1000, TimeUnit.MILLISECONDS)) //第三个网络请求，返回年龄
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(this::log); //包青天，男，28岁，耗时:3058毫秒，true

当然这种情况下使用 concatMap 的效果也是完全一样的，然而因为 concatMap 的核心是用来保证在合并时"有序"的，而这两种情况根本就没涉及到合并，所以这些情况下使用 concatMap是没有任何意义的。

flatMap 和 flatMapIterable

flatMapIterable 与 flatMap 在流程上大体都相同，唯一不同的是，flatMap 是将一个 Observable 转换成多个 Observables，每一个Observable 最后又得返回一个 Observable。而 flatMapIterable 在将一个 Observable 转换成多个 Observables 后，每一个 Observable 只能返回一个 Iterable 而不是另一个 Observable。

案例1：

Observable.just(Arrays.asList("篮球1", "足球1"))
      .flatMap(Observable::fromIterable) //返回一个 Observable
      .subscribe(string -> log("" + string));
Observable.just(Arrays.asList("篮球2", "足球2"))
      .flatMapIterable(list -> list) //返回一个 Iterable 而不是另一个 Observable
      .subscribe(string -> log("" + string));
Observable.fromIterable(Arrays.asList("篮球3", "足球3")) //和上面两种方式的结果一样
      .subscribe(string -> log("" + string));

 
Observable.just(Arrays.asList("篮球1", "足球1"))
      .flatMap(Observable::fromIterable) //返回一个 Observable
      .subscribe(string -> log("" + string));
Observable.just(Arrays.asList("篮球2", "足球2"))
      .flatMapIterable(list -> list) //返回一个 Iterable 而不是另一个 Observable
      .subscribe(string -> log("" + string));
Observable.fromIterable(Arrays.asList("篮球3", "足球3")) //和上面两种方式的结果一样
      .subscribe(string -> log("" + string));

篮球1，01:00:39 493，true
足球1，01:00:39 494，true
篮球2，01:00:39 496，true
足球2，01:00:39 496，true
篮球3，01:00:39 499，true
足球3，01:00:39 499，true

 
篮球1，01:00:39 493，true
足球1，01:00:39 494，true
篮球2，01:00:39 496，true
足球2，01:00:39 496，true
篮球3，01:00:39 499，true
足球3，01:00:39 499，true

案例2：

Observable.just(new Person(Arrays.asList("包青天", "哈哈")), new Person(Arrays.asList("白乾涛", "你好")))
      .map(person -> person.loves)
      .flatMap(Observable::fromIterable) //返回一个 Observable
      .flatMap(string -> Observable.fromArray(string.toCharArray())) //返回一个 Observable
      .subscribe(array -> log(Arrays.toString(array)));
Observable.just(new Person(Arrays.asList("广州", "上海")), new Person(Arrays.asList("武汉", "长沙")))
      .map(person -> person.loves)
      .flatMap(Observable::fromIterable) //返回一个 Observable
      .flatMapIterable(string -> Arrays.asList(string.toCharArray())) //返回一个 Iterable 而不是另一个 Observable
      .subscribe(array -> log(Arrays.toString(array)));

 
Observable.just(new Person(Arrays.asList("包青天", "哈哈")), new Person(Arrays.asList("白乾涛", "你好")))
      .map(person -> person.loves)
      .flatMap(Observable::fromIterable) //返回一个 Observable
      .flatMap(string -> Observable.fromArray(string.toCharArray())) //返回一个 Observable
      .subscribe(array -> log(Arrays.toString(array)));
Observable.just(new Person(Arrays.asList("广州", "上海")), new Person(Arrays.asList("武汉", "长沙")))
      .map(person -> person.loves)
      .flatMap(Observable::fromIterable) //返回一个 Observable
      .flatMapIterable(string -> Arrays.asList(string.toCharArray())) //返回一个 Iterable 而不是另一个 Observable
      .subscribe(array -> log(Arrays.toString(array)));

[包, 青, 天]，01:23:27 376，true
[哈, 哈]，01:23:27 376，true
[白, 乾, 涛]，01:23:27 377，true
[你, 好]，01:23:27 377，true
[广, 州]，01:23:27 380，true
[上, 海]，01:23:27 380，true
[武, 汉]，01:23:27 381，true
[长, 沙]，01:23:27 382，true

 
[包, 青, 天]，01:23:27 376，true
[哈, 哈]，01:23:27 376，true
[白, 乾, 涛]，01:23:27 377，true
[你, 好]，01:23:27 377，true
[广, 州]，01:23:27 380，true
[上, 海]，01:23:27 380，true
[武, 汉]，01:23:27 381，true
[长, 沙]，01:23:27 382，true

buffer

定期从被观察者发送的事件中获取一定数量的事件并放到缓存区中，然后把这些数据集合打包发射

请注意，如果源ObservableSource发出onError通知，则事件会立即传递而不是首先发到缓冲区[without first emitting the buffer it is in the process of assembling.]。

buffer(count)

每接收到 count 个数据包裹，将这 count 个包裹打包，发送给订阅者

一次订阅2个：

Observable.range(1, 5)
    .buffer(2) //缓存区大小，步长==缓存区大小，等价于buffer(count, count)
    .subscribe(list -> log(list.toString()), t -> log(""), () -> log("完成")); //[1, 2]，[3, 4]，[5]，完成

 
Observable.range(1, 5)
    .buffer(2) //缓存区大小，步长==缓存区大小，等价于buffer(count, count)
    .subscribe(list -> log(list.toString()), t -> log(""), () -> log("完成")); //[1, 2]，[3, 4]，[5]，完成

一次全部订阅(将所有元素组装到集合中的效果)：

Observable.range(1, 10).buffer(10).subscribe(list -> log(list.toString())); //[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

x
 
Observable.range(1, 10).buffer(10).subscribe(list -> log(list.toString())); //[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

buffer(count, skip)

生成的ObservableSource每隔 skip 项就会 emits buffers，每个 buffers 都包含 count 个 items。

队列效果(先进先出)：

Observable.range(1, 5)
      .buffer(3, 1) // 缓存区大小，步长(每次获取新事件的数量)
      .subscribe(list -> log(list.toString()));//[1, 2, 3]，[2, 3, 4]，[3, 4, 5]，[4, 5]，[5]

 
Observable.range(1, 5)
      .buffer(3, 1) // 缓存区大小，步长(每次获取新事件的数量)
      .subscribe(list -> log(list.toString()));//[1, 2, 3]，[2, 3, 4]，[3, 4, 5]，[4, 5]，[5]

每次剔除一个效果：

Observable.range(1, 5).buffer(5, 1)
    .subscribe(list -> log(list.toString()));//[1, 2, 3, 4, 5]，[2, 3, 4, 5]，[3, 4, 5]，[4, 5]，[5]

x
 
Observable.range(1, 5).buffer(5, 1)
    .subscribe(list -> log(list.toString()));//[1, 2, 3, 4, 5]，[2, 3, 4, 5]，[3, 4, 5]，[4, 5]，[5]

只取奇数个效果：

Observable.range(1, 5).buffer(1, 2).subscribe(list -> log(list.toString()));//[1]，[3]，[5]

x
 
Observable.range(1, 5).buffer(1, 2).subscribe(list -> log(list.toString()));//[1]，[3]，[5]

buffer(timespan, unit)

持续收集直到指定的每隔时间后，然后发射一次并清空缓存区。

周期性订阅多个结果：

Observable.create(emitter -> {
   for (int i = 0; i < 8; i++) {
      SystemClock.sleep(100);//模拟耗时操作
      emitter.onNext(i);
   }
}).buffer(250, TimeUnit.MICROSECONDS) //等价于 count = Integer.MAX_VALUE
      .subscribe(list -> log("缓存区中事件：" + list.toString())); //[0, 1]，[2, 3]，[4, 5, 6]，[7]

x
 
Observable.create(emitter -> {
   for (int i = 0; i < 8; i++) {
      SystemClock.sleep(100);//模拟耗时操作
      emitter.onNext(i);
   }
}).buffer(250, TimeUnit.MICROSECONDS) //等价于 count = Integer.MAX_VALUE
      .subscribe(list -> log("缓存区中事件：" + list.toString())); //[0, 1]，[2, 3]，[4, 5, 6]，[7]

buffer(timespan, unit, count)

当达到指定时间【或】缓冲区中达到指定数量时发射

Observable.create(emitter -> {
   for (int i = 0; i < 8; i++) {
      SystemClock.sleep(100);//每个对象均延迟后再单独发出去
      emitter.onNext(i);
   }
}).buffer(250, TimeUnit.MICROSECONDS, 2) //可以指定工作所在的线程
      .subscribe(list -> log("缓存区中事件：" + list.toString())); //[0, 1]，[]，[2, 3]，[]，[4, 5]，[6]，[7]

x
 
Observable.create(emitter -> {
   for (int i = 0; i < 8; i++) {
      SystemClock.sleep(100);//每个对象均延迟后再单独发出去
      emitter.onNext(i);
   }
}).buffer(250, TimeUnit.MICROSECONDS, 2) //可以指定工作所在的线程
      .subscribe(list -> log("缓存区中事件：" + list.toString())); //[0, 1]，[]，[2, 3]，[]，[4, 5]，[6]，[7]

完整测试案例

public class TransformOperatorActivity extends ListActivity {
	private static Format FORMAT = new SimpleDateFormat("HH:mm:ss SSS", Locale.getDefault());
	
	protected void onCreate(Bundle savedInstanceState) {
		super.onCreate(savedInstanceState);
		String[] array = {"0、map",
				"1、flatMap 基础用法",
				"2、flatMap和concatMap的区别",
				"3、flatMap 实现多个网络请求依次依赖",
				"4、flatMap 实现多个网络请求依次依赖简化代码",
				"5、flatMapIterable 案例1",
				"6、flatMapIterable 案例2",
				"7、buffer(int count)",
				"8、buffer(count, skip)",
				"9、buffer(timespan, unit, count)",
		};
		setListAdapter(new ArrayAdapter<>(this, android.R.layout.simple_list_item_1, Arrays.asList(array)));
	}
	
	private int i;
	
	@Override
	protected void onListItemClick(ListView l, View v, int position, long id) {
		i++;
		switch (position) {
			case 0:
				Observable.just(new Date()) // Date 类型
						.map(Date::getTime) // long 类型
						.map(time -> time + 1000 * 60 * 60)// 改变 long 类型时间的值
						.map(time -> new SimpleDateFormat("HH:mm:ss", Locale.getDefault()).format(new Date(time))) //String 类型
						.subscribe(this::log);
				break;
			case 1:
				Observable.just(new Person(Arrays.asList("篮球", "足球", "排球")), new Person(Arrays.asList("画画", "跳舞")))
						.map(person -> person.loves)
						.flatMap(Observable::fromIterable) //fromIterable：逐个发送集合中的元素
						.subscribe(this::log);
				break;
			case 2:
				long start = System.currentTimeMillis();
				Observable.just(Arrays.asList(1, 2, 3), Arrays.asList(4, 5))
						.flatMap(list -> Observable.fromIterable(list).delay(list.size(), TimeUnit.SECONDS))//flatMap是无序的
						.subscribe((s -> log("f:" + s)), e -> log("f"), () -> log("f耗时" + (System.currentTimeMillis() - start))); //3秒
				Observable.just(Arrays.asList("A", "B", "C"), Arrays.asList("D", "E"))
						.concatMap(list -> Observable.fromIterable(list).delay(list.size(), TimeUnit.SECONDS))//concatMap是有序的
						.subscribe(s -> log("c:" + s), e -> log("c"), () -> log("c耗时" + (System.currentTimeMillis() - start))); //5秒
				break;
			case 3:
				firstRequest("原始值：" + FORMAT.format(new Date(System.currentTimeMillis())))
						.subscribeOn(Schedulers.io()) // 在io线程进行网络请求
						.observeOn(AndroidSchedulers.mainThread()) // 在主线程处理请求结果
						.doOnNext(response -> log("【第一个网络请求结束，响应为】" + response))//true
						.observeOn(Schedulers.io()) // 回到 io 线程去处理下一个网络请求
						.flatMap(this::secondRequest)//实现多个网络请求依次依赖
						.observeOn(AndroidSchedulers.mainThread()) // 在主线程处理请求结果
						.subscribe(string -> log("【第二个网络请求结束，响应为】" + string));//true，5 秒
				break;
			case 4:
				Observable.just("包青天").delay(1000, TimeUnit.MILLISECONDS) //第一个网络请求，返回姓名
						.flatMap(s -> Observable.just(s + "，男").delay(1000, TimeUnit.MILLISECONDS)) //第二个网络请求，返回性别
						.flatMap(s -> Observable.just(s + "，28岁").delay(1000, TimeUnit.MILLISECONDS)) //第三个网络请求，返回年龄
						.subscribeOn(Schedulers.io())
						.observeOn(AndroidSchedulers.mainThread())
						.subscribe(this::log); //包青天，男，28岁，耗时:3058毫秒，true
				break;
			case 5:
				Observable.just(Arrays.asList("篮球1", "足球1"))
						.flatMap(Observable::fromIterable) //返回一个 Observable
						.subscribe(string -> log("" + string));
				Observable.just(Arrays.asList("篮球2", "足球2"))
						.flatMapIterable(list -> list) //返回一个 Iterable 而不是另一个 Observable
						.subscribe(string -> log("" + string));
				Observable.fromIterable(Arrays.asList("篮球3", "足球3")) //和上面两种方式的结果一样
						.subscribe(string -> log("" + string));
				break;
			case 6:
				Observable.just(new Person(Arrays.asList("包青天", "哈哈")), new Person(Arrays.asList("白乾涛", "你好")))
						.map(person -> person.loves)
						.flatMap(Observable::fromIterable) //返回一个 Observable
						.flatMap(string -> Observable.fromArray(string.toCharArray())) //返回一个 Observable
						.subscribe(array -> log(Arrays.toString(array)));
				Observable.just(new Person(Arrays.asList("广州", "上海")), new Person(Arrays.asList("武汉", "长沙")))
						.map(person -> person.loves)
						.flatMap(Observable::fromIterable) //返回一个 Observable
						.flatMapIterable(string -> Arrays.asList(string.toCharArray())) //返回一个 Iterable 而不是另一个 Observable
						.subscribe(array -> log(Arrays.toString(array)));
				break;
			case 7:
				if (i % 2 == 0) {
					Observable.range(1, 5).buffer(2)  //缓存区大小，步长==缓存区大小，等价于buffer(count, count)
							.subscribe(list -> log(list.toString()), t -> log(""), () -> log("完成")); //[1, 2]，[3, 4]，[5]，完成
				} else {
					Observable.range(1, 10).buffer(10)  //将所有元素组装到集合中的效果
							.subscribe(list -> log(list.toString())); //[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
				}
				break;
			case 8:
				if (i % 3 == 0) {
					Observable.range(1, 5).buffer(3, 1) // 缓存区大小，步长；队列效果(先进先出)
							.subscribe(list -> log(list.toString()));//[1, 2, 3]，[2, 3, 4]，[3, 4, 5]，[4, 5]，[5]
				} else if (i % 3 == 1) {
					Observable.range(1, 5).buffer(5, 1) //每次剔除一个效果
							.subscribe(list -> log(list.toString()));//[1, 2, 3, 4, 5]，[2, 3, 4, 5]，[3, 4, 5]，[4, 5]，[5]
				} else {
					Observable.range(1, 5).buffer(1, 2) //只取奇数个效果
							.subscribe(list -> log(list.toString()));//[1]，[3]，[5]
				}
				break;
			case 9:
				Observable<Integer> observable = Observable.create(emitter -> {
					for (int i = 0; i < 8; i++) {
						SystemClock.sleep(100);//模拟耗时操作
						emitter.onNext(i);
					}
					emitter.onComplete();
				});
				if (i % 3 == 0) { //周期性订阅多个结果：
					observable.buffer(250, TimeUnit.MILLISECONDS) //等价于 count = Integer.MAX_VALUE
							.subscribe(list -> log("缓存区中事件：" + list.toString())); //[0, 1]，[2, 3]，[4, 5, 6]，[7]
				} else { //当达到指定时间【或】缓冲区中达到指定数量时发射
					observable.buffer(250, TimeUnit.MILLISECONDS, 2) //可以指定工作所在的线程
							.subscribe(list -> log("缓存区中事件：" + list.toString())); //[0, 1]，[]，[2, 3]，[]，[4, 5]，[6]，[7]
				}
				break;
		}
	}
	
	private Observable<String> firstRequest(String parameter) {
		return Observable.create(emitter -> {
			SystemClock.sleep(2000);//模拟网络请求
			emitter.onNext(parameter + "，第一次修改：" + FORMAT.format(new Date(System.currentTimeMillis())));
			emitter.onComplete();
		});
	}
	
	private Observable<String> secondRequest(String parameter) {
		return Observable.create(emitter -> {
			SystemClock.sleep(3000);//模拟网络请求
			emitter.onNext(parameter + "，第二次修改：" + FORMAT.format(new Date(System.currentTimeMillis())));
			emitter.onComplete();
		});
	}
	
	private void log(String s) {
		String date = new SimpleDateFormat("HH:mm:ss SSS", Locale.getDefault()).format(new Date());
		Log.i("【bqt】", s + "，" + date + "，" + (Looper.myLooper() == Looper.getMainLooper()));
	}
}

 
public class TransformOperatorActivity extends ListActivity {
    private static Format FORMAT = new SimpleDateFormat("HH:mm:ss SSS", Locale.getDefault());
    
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        String[] array = {"0、map",
                "1、flatMap 基础用法",
                "2、flatMap和concatMap的区别",
                "3、flatMap 实现多个网络请求依次依赖",
                "4、flatMap 实现多个网络请求依次依赖简化代码",
                "5、flatMapIterable 案例1",
                "6、flatMapIterable 案例2",
                "7、buffer(int count)",
                "8、buffer(count, skip)",
                "9、buffer(timespan, unit, count)",
        };
        setListAdapter(new ArrayAdapter<>(this, android.R.layout.simple_list_item_1, Arrays.asList(array)));
    }
    
    private int i;
    
    @Override
    protected void onListItemClick(ListView l, View v, int position, long id) {
        i++;
        switch (position) {
            case 0:
                Observable.just(new Date()) // Date 类型
                        .map(Date::getTime) // long 类型
                        .map(time -> time + 1000 * 60 * 60)// 改变 long 类型时间的值
                        .map(time -> new SimpleDateFormat("HH:mm:ss", Locale.getDefault()).format(new Date(time))) //String 类型
                        .subscribe(this::log);
                break;
            case 1:
                Observable.just(new Person(Arrays.asList("篮球", "足球", "排球")), new Person(Arrays.asList("画画", "跳舞")))
                        .map(person -> person.loves)
                        .flatMap(Observable::fromIterable) //fromIterable：逐个发送集合中的元素
                        .subscribe(this::log);
                break;
            case 2:
                long start = System.currentTimeMillis();
                Observable.just(Arrays.asList(1, 2, 3), Arrays.asList(4, 5))
                        .flatMap(list -> Observable.fromIterable(list).delay(list.size(), TimeUnit.SECONDS))//flatMap是无序的
                        .subscribe((s -> log("f:" + s)), e -> log("f"), () -> log("f耗时" + (System.currentTimeMillis() - start))); //3秒
                Observable.just(Arrays.asList("A", "B", "C"), Arrays.asList("D", "E"))
                        .concatMap(list -> Observable.fromIterable(list).delay(list.size(), TimeUnit.SECONDS))//concatMap是有序的
                        .subscribe(s -> log("c:" + s), e -> log("c"), () -> log("c耗时" + (System.currentTimeMillis() - start))); //5秒
                break;
            case 3:
                firstRequest("原始值：" + FORMAT.format(new Date(System.currentTimeMillis())))
                        .subscribeOn(Schedulers.io()) // 在io线程进行网络请求
                        .observeOn(AndroidSchedulers.mainThread()) // 在主线程处理请求结果
                        .doOnNext(response -> log("【第一个网络请求结束，响应为】" + response))//true
                        .observeOn(Schedulers.io()) // 回到 io 线程去处理下一个网络请求
                        .flatMap(this::secondRequest)//实现多个网络请求依次依赖
                        .observeOn(AndroidSchedulers.mainThread()) // 在主线程处理请求结果
                        .subscribe(string -> log("【第二个网络请求结束，响应为】" + string));//true，5 秒
                break;
            case 4:
                Observable.just("包青天").delay(1000, TimeUnit.MILLISECONDS) //第一个网络请求，返回姓名
                        .flatMap(s -> Observable.just(s + "，男").delay(1000, TimeUnit.MILLISECONDS)) //第二个网络请求，返回性别
                        .flatMap(s -> Observable.just(s + "，28岁").delay(1000, TimeUnit.MILLISECONDS)) //第三个网络请求，返回年龄
                        .subscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread())
                        .subscribe(this::log); //包青天，男，28岁，耗时:3058毫秒，true
                break;
            case 5:
                Observable.just(Arrays.asList("篮球1", "足球1"))
                        .flatMap(Observable::fromIterable) //返回一个 Observable
                        .subscribe(string -> log("" + string));
                Observable.just(Arrays.asList("篮球2", "足球2"))
                        .flatMapIterable(list -> list) //返回一个 Iterable 而不是另一个 Observable
                        .subscribe(string -> log("" + string));
                Observable.fromIterable(Arrays.asList("篮球3", "足球3")) //和上面两种方式的结果一样
                        .subscribe(string -> log("" + string));
                break;
            case 6:
                Observable.just(new Person(Arrays.asList("包青天", "哈哈")), new Person(Arrays.asList("白乾涛", "你好")))
                        .map(person -> person.loves)
                        .flatMap(Observable::fromIterable) //返回一个 Observable
                        .flatMap(string -> Observable.fromArray(string.toCharArray())) //返回一个 Observable
                        .subscribe(array -> log(Arrays.toString(array)));
                Observable.just(new Person(Arrays.asList("广州", "上海")), new Person(Arrays.asList("武汉", "长沙")))
                        .map(person -> person.loves)
                        .flatMap(Observable::fromIterable) //返回一个 Observable
                        .flatMapIterable(string -> Arrays.asList(string.toCharArray())) //返回一个 Iterable 而不是另一个 Observable
                        .subscribe(array -> log(Arrays.toString(array)));
                break;
            case 7:
                if (i % 2 == 0) {
                    Observable.range(1, 5).buffer(2)  //缓存区大小，步长==缓存区大小，等价于buffer(count, count)
                            .subscribe(list -> log(list.toString()), t -> log(""), () -> log("完成")); //[1, 2]，[3, 4]，[5]，完成
                } else {
                    Observable.range(1, 10).buffer(10)  //将所有元素组装到集合中的效果
                            .subscribe(list -> log(list.toString())); //[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
                }
                break;
            case 8:
                if (i % 3 == 0) {
                    Observable.range(1, 5).buffer(3, 1) // 缓存区大小，步长；队列效果(先进先出)
                            .subscribe(list -> log(list.toString()));//[1, 2, 3]，[2, 3, 4]，[3, 4, 5]，[4, 5]，[5]
                } else if (i % 3 == 1) {
                    Observable.range(1, 5).buffer(5, 1) //每次剔除一个效果
                            .subscribe(list -> log(list.toString()));//[1, 2, 3, 4, 5]，[2, 3, 4, 5]，[3, 4, 5]，[4, 5]，[5]
                } else {
                    Observable.range(1, 5).buffer(1, 2) //只取奇数个效果
                            .subscribe(list -> log(list.toString()));//[1]，[3]，[5]
                }
                break;
            case 9:
                Observable<Integer> observable = Observable.create(emitter -> {
                    for (int i = 0; i < 8; i++) {
                        SystemClock.sleep(100);//模拟耗时操作
                        emitter.onNext(i);
                    }
                    emitter.onComplete();
                });
                if (i % 3 == 0) { //周期性订阅多个结果：
                    observable.buffer(250, TimeUnit.MILLISECONDS) //等价于 count = Integer.MAX_VALUE
                            .subscribe(list -> log("缓存区中事件：" + list.toString())); //[0, 1]，[2, 3]，[4, 5, 6]，[7]
                } else { //当达到指定时间【或】缓冲区中达到指定数量时发射
                    observable.buffer(250, TimeUnit.MILLISECONDS, 2) //可以指定工作所在的线程
                            .subscribe(list -> log("缓存区中事件：" + list.toString())); //[0, 1]，[]，[2, 3]，[]，[4, 5]，[6]，[7]
                }
                break;
        }
    }
    
    private Observable<String> firstRequest(String parameter) {
        return Observable.create(emitter -> {
            SystemClock.sleep(2000);//模拟网络请求
            emitter.onNext(parameter + "，第一次修改：" + FORMAT.format(new Date(System.currentTimeMillis())));
            emitter.onComplete();
        });
    }
    
    private Observable<String> secondRequest(String parameter) {
        return Observable.create(emitter -> {
            SystemClock.sleep(3000);//模拟网络请求
            emitter.onNext(parameter + "，第二次修改：" + FORMAT.format(new Date(System.currentTimeMillis())));
            emitter.onComplete();
        });
    }
    
    private void log(String s) {
        String date = new SimpleDateFormat("HH:mm:ss SSS", Locale.getDefault()).format(new Date());
        Log.i("【bqt】", s + "，" + date + "，" + (Looper.myLooper() == Looper.getMainLooper()));
    }
}

2018-9-18

RxJava 变换操作符 map flatMap concatMap buffer

标签：next 完全 lov complete 模拟 ids z-index 使用回调

原文地址：https://www.cnblogs.com/baiqiantao/p/9688484.html

踩

(0)

评论一句话评论（0）

分享档案

更多>

2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)

周排行