RxJava 创建操作符 create just from defer timer interval

其他 2018-09-18 22:49:23 阅读次数: 0

常见的创建操作符

基础创建
  • create：通过调用观察者的方法从头创建一个observable，创建被观察者对象最基本的操作符
快速创建
  • just：【快捷使用方式】快速创建被观察者对象并直接发送传入的事件，最多发送10个事件(因为其是基于方法重载而非可变参数)
  • fromArray：和 just 类似，区别是，fromArray是基于可变参数的，所以其可接受的参数数量是没有限制的
  • fromIterable：快速创建被观察者对象并逐个发送集合中的元素
  • empty、error、never：直接发送Complete事件，或onError事件，或不发送任何事件
延迟创建
  • defer：在观察者订阅之前不创建这个observable，订阅后才会调用defer创建被观察者对象
  • timer：延迟指定时间后，发送1个值为0的Long类型对象
  • interval、intervalRange：每隔指定时间就发送一个Long类型的对象(从0开始每次递增1)
  • range、rangeLong：连续发送一个int(long)类型的序列，作用类似于 intervalRange，默认在主线程上

create

Observable.create(emitter -> {
   emitter.onNext(1); //打印 onNext:1
   emitter.onNext(2); //打印 onNext:2
   emitter.onNext(3); //打印 onNext:3
   emitter.onComplete(); //打印 onComplete
}).subscribe(s -> log("onNext:" + s), t -> log("onError"), () -> log("onComplete"));
6
6
 
1 
Observable.create(emitter -> {
2 
   emitter.onNext(1); //打印 onNext:1
3 
   emitter.onNext(2); //打印 onNext:2
4 
   emitter.onNext(3); //打印 onNext:3
5 
   emitter.onComplete(); //打印 onComplete
6 
}).subscribe(s -> log("onNext:" + s), t -> log("onError"), () -> log("onComplete"));

just

public static <T> Observable<T> just(T item1, T item2, T item3, T item4, T item5, T item6, T item7, T item8, T item9, T item10)
1
1
 
1 
public static <T> Observable<T> just(T item1, T item2, T item3, T item4, T item5, T item6, T item7, T item8, T item9, T item10)
Observable.just(1, 2, 3).subscribe(s -> log("onNext:" + s), t -> log("onError"), () -> log("onComplete"));//和上面一样
Observable.just(new int[]{7, 8, 9}).subscribe(arr -> log("onNext:" + Arrays.toString(arr)));//发送一个对象，onNext:[7, 8, 9]
Observable.just(Arrays.asList(4, 5, 6)).subscribe(list -> log("onNext:" + list.toString()));//发送一个对象，onNext:[4, 5, 6]
3
3
 
1 
Observable.just(1, 2, 3).subscribe(s -> log("onNext:" + s), t -> log("onError"), () -> log("onComplete"));//和上面一样
2 
Observable.just(new int[]{7, 8, 9}).subscribe(arr -> log("onNext:" + Arrays.toString(arr)));//发送一个对象，onNext:[7, 8, 9]
3 
Observable.just(Arrays.asList(4, 5, 6)).subscribe(list -> log("onNext:" + list.toString()));//发送一个对象，onNext:[4, 5, 6]

fromArray

public static <T> Observable<T> fromArray(T... items)
1
1
 
1 
public static <T> Observable<T> fromArray(T... items)
Observable.fromArray(1, 2, 3).subscribe(i -> log("*" + i));//整个发送传入的对象，*1 *2 *3
Observable.fromArray(new int[]{4, 5, 6}).subscribe(arr -> log(Arrays.toString(arr)));//整个发送传入的对象，[4, 5, 6]
Observable.fromArray(new int[]{7, 8}, new int[]{9, 10}).subscribe(arr -> log(Arrays.toString(arr)));//[7, 8]  [9, 10]
3
3
 
1 
Observable.fromArray(1, 2, 3).subscribe(i -> log("*" + i));//整个发送传入的对象，*1 *2 *3
2 
Observable.fromArray(new int[]{4, 5, 6}).subscribe(arr -> log(Arrays.toString(arr)));//整个发送传入的对象，[4, 5, 6]
3 
Observable.fromArray(new int[]{7, 8}, new int[]{9, 10}).subscribe(arr -> log(Arrays.toString(arr)));//[7, 8]  [9, 10]

fromIterable

public static <T> Observable<T> fromIterable(Iterable<? extends T> source)
1
1
 
1 
public static <T> Observable<T> fromIterable(Iterable<? extends T> source)
Observable.fromIterable(Arrays.asList(1, 2, 3)).subscribe(i -> log("*" + i));//逐个发送对象，*1 *2 *3
1
1
 
1 
Observable.fromIterable(Arrays.asList(1, 2, 3)).subscribe(i -> log("*" + i));//逐个发送对象，*1 *2 *3

empty、error、never

Observable.empty().subscribe(o -> log("onNext"), 
      t -> log("onError"), () -> log("onComplete"), d -> log("onSubscribe"));//onSubscribe onComplete
Observable.error(new Throwable("")).subscribe(o -> log("onNext"),
      t -> log("onError"), () -> log("onComplete"), d -> log("onSubscribe"));//onSubscribe onError
Observable.never().subscribe(o -> log("onNext"),
      t -> log("onError"), () -> log("onComplete"), d -> log("onSubscribe"));//onSubscribe
6
6
 
1 
Observable.empty().subscribe(o -> log("onNext"),
2 
      t -> log("onError"), () -> log("onComplete"), d -> log("onSubscribe"));//onSubscribe onComplete
3 
Observable.error(new Throwable("")).subscribe(o -> log("onNext"),
4 
      t -> log("onError"), () -> log("onComplete"), d -> log("onSubscribe"));//onSubscribe onError
5 
Observable.never().subscribe(o -> log("onNext"),
6 
      t -> log("onError"), () -> log("onComplete"), d -> log("onSubscribe"));//onSubscribe

defer

Format format = new SimpleDateFormat("HH:mm:ss SSS", Locale.getDefault());//订阅前不创建
Observable<String> observable = Observable.defer(() -> Observable.just(format.format(new Date())));
log("当前时间：" + format.format(new Date()));   //当前时间：21:19:12 523
SystemClock.sleep(1000);
observable.subscribe(l -> log("发送对象：" + l));//订阅后才会调用defer创建被观察者对象：21:19:13 028
5
5
 
1 
Format format = new SimpleDateFormat("HH:mm:ss SSS", Locale.getDefault());//订阅前不创建
2 
Observable<String> observable = Observable.defer(() -> Observable.just(format.format(new Date())));
3 
log("当前时间：" + format.format(new Date()));   //当前时间：21:19:12 523
4 
SystemClock.sleep(1000);
5 
observable.subscribe(l -> log("发送对象：" + l));//订阅后才会调用defer创建被观察者对象：21:19:13 028

timer

Observable.timer(1000, TimeUnit.MILLISECONDS) //延迟指定时间后，发送1个值为0的Long类型对象，默认在子线程上
      .doOnSubscribe(s -> log2("doOnSubscribe1")) //doOnSubscribe1，21:33:02 109，true
      .subscribe(l -> log2("onNext1：" + l)); //onNext1：0，21:33:03 112，false
Observable.timer(1000, TimeUnit.MILLISECONDS, AndroidSchedulers.mainThread()) //指定订阅者接收事件所在线程
      .doOnSubscribe(s -> log2("doOnSubscribe2")) //doOnSubscribe2，21:33:02 111，true
      .subscribe(l -> log2("onNext2：" + l)); //onNext2：0，21:33:03 113，true
6
6
 
1 
Observable.timer(1000, TimeUnit.MILLISECONDS) //延迟指定时间后，发送1个值为0的Long类型对象，默认在子线程上
2 
      .doOnSubscribe(s -> log2("doOnSubscribe1")) //doOnSubscribe1，21:33:02 109，true
3 
      .subscribe(l -> log2("onNext1：" + l)); //onNext1：0，21:33:03 112，false
4 
Observable.timer(1000, TimeUnit.MILLISECONDS, AndroidSchedulers.mainThread()) //指定订阅者接收事件所在线程
5 
      .doOnSubscribe(s -> log2("doOnSubscribe2")) //doOnSubscribe2，21:33:02 111，true
6 
      .subscribe(l -> log2("onNext2：" + l)); //onNext2：0，21:33:03 113，true

interval、intervalRange

log2("开始时间"); //每次递增1，默认在子线程上，可指定线程调度器
Observable.interval(5000, 1000, TimeUnit.MILLISECONDS)//首次延迟时间、间隔时间，
      .subscribe(l -> log2("接收的对象2：" + l)); //不会自动结束的
Observable.intervalRange(100, 3, 0, 1, TimeUnit.SECONDS) //起始值，发送总数量
      .subscribe(l -> log2("接收的对象1：" + l), t -> log2("onError"), () -> log2("接收的对象1：onComplete"));
5
5
 
1 
log2("开始时间"); //每次递增1，默认在子线程上，可指定线程调度器
2 
Observable.interval(5000, 1000, TimeUnit.MILLISECONDS)//首次延迟时间、间隔时间，
3 
      .subscribe(l -> log2("接收的对象2：" + l)); //不会自动结束的
4 
Observable.intervalRange(100, 3, 0, 1, TimeUnit.SECONDS) //起始值，发送总数量
5 
      .subscribe(l -> log2("接收的对象1：" + l), t -> log2("onError"), () -> log2("接收的对象1：onComplete"));
开始时间，21:53:29 350，true
接收的对象1：100，21:53:29 365，false
接收的对象1：101，21:53:30 366，false
接收的对象1：102，21:53:31 366，false
接收的对象1：onComplete，21:53:31 366，false

接收的对象2：0，21:53:34 363，false
接收的对象2：1，21:53:35 364，false
8
 
1 
开始时间，21:53:29 350，true
2 
接收的对象1：100，21:53:29 365，false
3 
接收的对象1：101，21:53:30 366，false
4 
接收的对象1：102，21:53:31 366，false
5 
接收的对象1：onComplete，21:53:31 366，false
6
7 
接收的对象2：0，21:53:34 363，false
8 
接收的对象2：1，21:53:35 364，false

range、rangeLong

Observable.range(10, 3).subscribe(i -> log2("" + i), t -> log2(""), () -> log2("onComplete1"));
Observable.rangeLong(20, 2).subscribe(i -> log2("" + i), t -> log2(""), () -> log2("onComplete2"));
2
2
 
1 
Observable.range(10, 3).subscribe(i -> log2("" + i), t -> log2(""), () -> log2("onComplete1"));
2 
Observable.rangeLong(20, 2).subscribe(i -> log2("" + i), t -> log2(""), () -> log2("onComplete2"));

完整测试案例

public class CreateOperatorActivity extends ListActivity {
	private Format format = new SimpleDateFormat("HH:mm:ss SSS", Locale.getDefault());
	
	protected void onCreate(Bundle savedInstanceState) {
		super.onCreate(savedInstanceState);
		String[] array = {"create",
				"just",
				"fromIterable",
				"fromArray",
				"empty、error、never",
				"defer",
				"timer",
				"interval、intervalRange",
				"range、rangeLong",
				"",};
		setListAdapter(new ArrayAdapter<>(this, android.R.layout.simple_list_item_1, Arrays.asList(array)));
	}
	
	@Override
	protected void onListItemClick(ListView listView, View v, int position, long id) {
		switch (position) {
			case 0:
				Observable.create(emitter -> {
					emitter.onNext(1); //onNext:1
					emitter.onNext(2); //onNext:2
					emitter.onNext(3); //onNext:3
					emitter.onComplete(); //onComplete
				}).subscribe(s -> log("onNext:" + s), t -> log("onError"), () -> log("onComplete"));
				break;
			case 1:
				Observable.just(1, 2, 3).subscribe(s -> log("onNext:" + s), t -> log("onError"), () -> log("onComplete"));//和上面一样
				Observable.just(Arrays.asList(4, 5, 6)).subscribe(list -> log("onNext:" + list.toString()));//发送一个对象，onNext:[4, 5, 6]
				Observable.just(new int[]{7, 8, 9}).subscribe(arr -> log("onNext:" + Arrays.toString(arr)));//发送一个对象，onNext:[7, 8, 9]
				break;
			case 2:
				Observable.fromIterable(Arrays.asList(1, 2, 3)).subscribe(i -> log("*" + i));//逐个发送对象，*1 *2 *3
				break;
			case 3:
				Observable.fromArray(1, 2, 3).subscribe(i -> log("*" + i));//整个发送传入的对象，*1 *2 *3
				Observable.fromArray(new int[]{4, 5, 6}).subscribe(arr -> log(Arrays.toString(arr)));//整个发送传入的对象，[4, 5, 6]
				Observable.fromArray(new int[]{7, 8}, new int[]{9, 10}).subscribe(arr -> log(Arrays.toString(arr)));//[7, 8]  [9, 10]
				break;
			case 4:
				Observable.empty().subscribe(o -> log("onNext"),
						t -> log("onError"), () -> log("onComplete"), d -> log("onSubscribe"));//onSubscribe onComplete
				Observable.error(new Throwable("")).subscribe(o -> log("onNext"),
						t -> log("onError"), () -> log("onComplete"), d -> log("onSubscribe"));//onSubscribe onError
				Observable.never().subscribe(o -> log("onNext"),
						t -> log("onError"), () -> log("onComplete"), d -> log("onSubscribe"));//onSubscribe
				break;
			case 5:
				Observable<String> observable = Observable.defer(() -> Observable.just(format.format(new Date())));//订阅前不创建
				log("当前时间：" + format.format(new Date()));   //当前时间：21:19:12 523
				SystemClock.sleep(1000);
				observable.subscribe(l -> log("发送对象：" + l));//订阅后才会调用defer创建被观察者对象：21:19:13 028
				break;
			case 6:
				Observable.timer(1000, TimeUnit.MILLISECONDS) //延迟指定时间后，发送1个值为0的Long类型对象，默认在子线程上
						.doOnSubscribe(s -> log2("doOnSubscribe1")) //doOnSubscribe1，21:33:02 109，true
						.subscribe(l -> log2("onNext1：" + l)); //onNext1：0，21:33:03 112，false
				Observable.timer(1000, TimeUnit.MILLISECONDS, AndroidSchedulers.mainThread()) //指定订阅者接收事件所在线程
						.doOnSubscribe(s -> log2("doOnSubscribe2")) //doOnSubscribe2，21:33:02 111，true
						.subscribe(l -> log2("onNext2：" + l)); //onNext2：0，21:33:03 113，true
				break;
			case 7:
				log2("开始时间"); //每次递增1，默认在子线程上，可指定线程调度器
				Observable.interval(5000, 1000, TimeUnit.MILLISECONDS)//首次延迟时间、间隔时间，
						.subscribe(l -> log2("接收的对象2：" + l)); //不会自动结束的
				Observable.intervalRange(100, 3, 0, 1, TimeUnit.SECONDS) //起始值，发送总数量
						.subscribe(l -> log2("接收的对象1：" + l), t -> log2("onError"), () -> log2("接收的对象1：onComplete"));
				break;
			case 8:
				Observable.range(10, 3).subscribe(i -> log2("" + i), t -> log2(""), () -> log2("onComplete1"));
				Observable.rangeLong(20, 2).subscribe(i -> log2("" + i), t -> log2(""), () -> log2("onComplete2"));
				break;
			case 9:
				
				break;
		}
	}
	
	private void log(String s) {
		Log.i("【bqt】", s);
	}
	
	private void log2(String s) {
		Log.i("【bqt】", s + "，" + format.format(new Date()) + "，" + (Looper.myLooper() == Looper.getMainLooper()));
	}
}
89
89
 
1 
public class CreateOperatorActivity extends ListActivity {
2 
    private Format format = new SimpleDateFormat("HH:mm:ss SSS", Locale.getDefault());
3
4 
    protected void onCreate(Bundle savedInstanceState) {
5 
        super.onCreate(savedInstanceState);
6 
        String[] array = {"create",
7 
                "just",
8 
                "fromIterable",
9 
                "fromArray",
10 
                "empty、error、never",
11 
                "defer",
12 
                "timer",
13 
                "interval、intervalRange",
14 
                "range、rangeLong",
15 
                "",};
16 
        setListAdapter(new ArrayAdapter<>(this, android.R.layout.simple_list_item_1, Arrays.asList(array)));
17 
    }
18
19 
    @Override
20 
    protected void onListItemClick(ListView listView, View v, int position, long id) {
21 
        switch (position) {
22 
            case 0:
23 
                Observable.create(emitter -> {
24 
                    emitter.onNext(1); //onNext:1
25 
                    emitter.onNext(2); //onNext:2
26 
                    emitter.onNext(3); //onNext:3
27 
                    emitter.onComplete(); //onComplete
28 
                }).subscribe(s -> log("onNext:" + s), t -> log("onError"), () -> log("onComplete"));
29 
                break;
30 
            case 1:
31 
                Observable.just(1, 2, 3).subscribe(s -> log("onNext:" + s), t -> log("onError"), () -> log("onComplete"));//和上面一样
32 
                Observable.just(Arrays.asList(4, 5, 6)).subscribe(list -> log("onNext:" + list.toString()));//发送一个对象，onNext:[4, 5, 6]
33 
                Observable.just(new int[]{7, 8, 9}).subscribe(arr -> log("onNext:" + Arrays.toString(arr)));//发送一个对象，onNext:[7, 8, 9]
34 
                break;
35 
            case 2:
36 
                Observable.fromIterable(Arrays.asList(1, 2, 3)).subscribe(i -> log("*" + i));//逐个发送对象，*1 *2 *3
37 
                break;
38 
            case 3:
39 
                Observable.fromArray(1, 2, 3).subscribe(i -> log("*" + i));//整个发送传入的对象，*1 *2 *3
40 
                Observable.fromArray(new int[]{4, 5, 6}).subscribe(arr -> log(Arrays.toString(arr)));//整个发送传入的对象，[4, 5, 6]
41 
                Observable.fromArray(new int[]{7, 8}, new int[]{9, 10}).subscribe(arr -> log(Arrays.toString(arr)));//[7, 8]  [9, 10]
42 
                break;
43 
            case 4:
44 
                Observable.empty().subscribe(o -> log("onNext"),
45 
                        t -> log("onError"), () -> log("onComplete"), d -> log("onSubscribe"));//onSubscribe onComplete
46 
                Observable.error(new Throwable("")).subscribe(o -> log("onNext"),
47 
                        t -> log("onError"), () -> log("onComplete"), d -> log("onSubscribe"));//onSubscribe onError
48 
                Observable.never().subscribe(o -> log("onNext"),
49 
                        t -> log("onError"), () -> log("onComplete"), d -> log("onSubscribe"));//onSubscribe
50 
                break;
51 
            case 5:
52 
                Observable<String> observable = Observable.defer(() -> Observable.just(format.format(new Date())));//订阅前不创建
53 
                log("当前时间：" + format.format(new Date()));   //当前时间：21:19:12 523
54 
                SystemClock.sleep(1000);
55 
                observable.subscribe(l -> log("发送对象：" + l));//订阅后才会调用defer创建被观察者对象：21:19:13 028
56 
                break;
57 
            case 6:
58 
                Observable.timer(1000, TimeUnit.MILLISECONDS) //延迟指定时间后，发送1个值为0的Long类型对象，默认在子线程上
59 
                        .doOnSubscribe(s -> log2("doOnSubscribe1")) //doOnSubscribe1，21:33:02 109，true
60 
                        .subscribe(l -> log2("onNext1：" + l)); //onNext1：0，21:33:03 112，false
61 
                Observable.timer(1000, TimeUnit.MILLISECONDS, AndroidSchedulers.mainThread()) //指定订阅者接收事件所在线程
62 
                        .doOnSubscribe(s -> log2("doOnSubscribe2")) //doOnSubscribe2，21:33:02 111，true
63 
                        .subscribe(l -> log2("onNext2：" + l)); //onNext2：0，21:33:03 113，true
64 
                break;
65 
            case 7:
66 
                log2("开始时间"); //每次递增1，默认在子线程上，可指定线程调度器
67 
                Observable.interval(5000, 1000, TimeUnit.MILLISECONDS)//首次延迟时间、间隔时间，
68 
                        .subscribe(l -> log2("接收的对象2：" + l)); //不会自动结束的
69 
                Observable.intervalRange(100, 3, 0, 1, TimeUnit.SECONDS) //起始值，发送总数量
70 
                        .subscribe(l -> log2("接收的对象1：" + l), t -> log2("onError"), () -> log2("接收的对象1：onComplete"));
71 
                break;
72 
            case 8:
73 
                Observable.range(10, 3).subscribe(i -> log2("" + i), t -> log2(""), () -> log2("onComplete1"));
74 
                Observable.rangeLong(20, 2).subscribe(i -> log2("" + i), t -> log2(""), () -> log2("onComplete2"));
75 
                break;
76 
            case 9:
77
78 
                break;
79 
        }
80 
    }
81
82 
    private void log(String s) {
83 
        Log.i("【bqt】", s);
84 
    }
85
86 
    private void log2(String s) {
87 
        Log.i("【bqt】", s + "，" + format.format(new Date()) + "，" + (Looper.myLooper() == Looper.getMainLooper()));
88 
    }
89 
}
2018-9-18

猜你喜欢

转载自www.cnblogs.com/baiqiantao/p/9671820.html
今日推荐
周排行
Leetcode简单题61~80
解决zookeeper磁盘IO高的问题
多线程相关方法详解
Maven-setting.xml文件详解
Maven 项目的 classpath 理解
渊亭科技大数据笔试题
配置JVM内存分配
计算机网络个人学习笔记 （三）网络层 ：第三部分 连载
js中两个等号(==)和三个等号(===)的区别
用C程序自动打开电脑上的程序
每日归档
更多
2024-09-18(0)
2024-09-17(0)
2024-09-16(0)
2024-09-15(0)
2024-09-14(0)
2024-09-13(0)
2024-09-12(0)
2024-09-11(0)
2024-09-10(0)
2024-09-09(0)

代码天地 © 2018 codetd.com

境外服务器   最新电视剧2022