欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  Java

深入浅出RxJava_03[被观察者创建操作]的详细介绍

程序员文章站 2022-04-23 08:10:16
...

本教程基于RxJava1.x版本进行全面讲解,后续课程将陆续更新,敬请关注…

以下的函数都是用来创建被观察者Observable的函数。我们可以根据需要创建对应的函数。

  1. Create - 最原始的Observable创建函数

  2. Defer - 创建订阅后才创建的Observable

  3. Empty/Never/Throw - 创建不发送数据/以发送一个异常的数据的bservable

  4. Just - 创建发送1-9个值的Observable

  5. From - 创建发送一个队列的Observable

  6. Interval&Timer -创建一个类似定时器的Observable

  7. Range - 创建一个发送一个特定整数型的Observable

  8. Repeat - 创建一个设定重复发送次数的Observable

1.Create

你可以使用Create操作符从头开始创建一个Observable,给这个操作符传递一个接受观察者作为参数的函数,编写这个函数让它的行为表现为一个Observable–恰当的调用观察者的onNext,onError和onCompleted方法。

一个形式正确的有限Observable必须尝试调用观察者的onCompleted正好一次或者它的onError正好一次,而且此后不能再调用观察者的任何其它方法。

建议你在传递给create方法的函数中检查观察者的isUnsubscribed状态,以便在没有观察者的时候,让你的Observable停止发射数据或者做昂贵的运算。

示例代码:

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.

2.Defer

直到有观察者订阅时才创建Observable,并且为每个观察者创建一个新的Observable。

Defer操作符会一直等待直到有观察者订阅它,然后它使用Observable工厂方法生成一个Observable。它对每个观察者都这样做,因此尽管每个订阅者都以为自己订阅的是同一个Observable,事实上每个订阅者获取的是它们自己的单独的数据序列。

代码如下:

Observable<String> defer = Observable.defer(new Func0<Observable<String>>() {
    //当observable被创建的时候顺便调用observable内部的call()方法并在方法中发送消息
    //每subscribe()就会在call()中返回一个新的实例对象
    @Override
    public Observable<String> call() {
        Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("Hello Android !");
            }
        });
        return observable;
    }
});
defer.subscribe(new Action1<String>() {
    @Override
    public void call(String s) {
    }
});

defer.subscribe(new Action1<String>() {
    @Override
    public void call(String s) {
    }
});

3.Empty/Never/Throw

Empty

创建一个不发射任何数据但是正常终止的Observable

Never

创建一个不发射数据也不终止的Observable

Throw

创建一个不发射数据以一个错误终止的Observable

这三个操作符生成的Observable行为非常特殊和受限。测试的时候很有用,有时候也用于结合其它的Observables,或者作为其它需要Observable的操作符的参数。

RxJava将这些操作符实现为 empty,never和error。error操作符需要一个Throwable参数,你的Observable会以此终止。这些操作符默认不在任何特定的调度器上执行,但是empty和error有一个可选参数是Scheduler,如果你传递了Scheduler参数,它们会在这个调度器上发送通知。

4.Just

如果被观察者想发射一个数据给观察者,那么代码是这样:

Observable
    .create(new Observable.OnSubscribe<String>() {
        @Override
        public void call(Subscriber<? super String> subscriber) {
            subscriber.onNext("hello Android");
        }
    })
    .subscribe(new Action1<String>() {
        @Override
        public void call(String s) {
            Log.i(TAG, "call: "+s);
        }
    });

上面的代码如想发送数据 首先要实现Observable.OnSubscribe接口,我们可以使用just函数代替。示例代码如下:

Observable
    .just("hello Android")
    .subscribe(new Action1<String>() {
        @Override
        public void call(String s) {
            Log.i(TAG, "call: "+s);
        }
    });

输出:

onNext: hello Android
Sequence complete.

Just将单个数据转换为发射那个数据的Observable。

类似的函数有:

  • just(T t1);

  • just(T t1,T t2);

  • just(T t1,T t2,T t3);

  • just(T t1,T t2,T t3,T t4);

  • just(T t1,T t2,T t3,T t4,T t5);

RxJava将这个操作符实现为just函数,它接受一至九个参数,返回一个按参数列表顺序发射这些数据的Observable。

5.From

相比于发送一个到9个数据的just函数,From函数发送一个队列的数据,输出的长度无限。

  1. 学校开了一个兴趣班,开学班里来了几个同学。下面就是同学类的定义:

    public static class Student {
    
        public String name;
        public int age;
    
        public Student(String name, int age) {
            this.name = name;
            this.age = age;
        }
    
        @Override
        public String toString() {
            return "Student{" +
                    "name='" + name + '\'' +
                    ", age=" + age +
                    '}';
        }
    }
  2. 这里创建一个集合 用来存储每个同学的信息。

    private ArrayList initPersons() {
        ArrayList<Student> persons = new ArrayList<>();
        persons.add(new Student("张三", 16));
        persons.add(new Student("李四", 17));
        persons.add(new Student("王二麻子", 18));
        return persons;
    }
  3. 接下来老师点评 每个都要喊一声到,这里通过打印来说明每个同学的签到情况。

    ArrayList persons = initPersons();
    for (int i = 0; i < persons.size(); i++) {
        //打印每个同学
        Log.i(TAG,persons.get(i).toString());
    }

RxJava这样转换:

//1.定义被观察者,用来发送一个队列的事件
Observable<Student> observable = Observable.from(persons);

//2.当开始订阅的时候 接收被观察者发送过来的一系列事件
observable.subscribe(new Action1<Student>() {
    @Override
    public void call(Student student) {
        Log.i(TAG, "call: "+student.toString());
    }
});

6.Interval&Timer

创建一个按固定时间间隔发射整数序列的Observable。

RxJava将这个操作符实现为interval方法。它接受一个表示时间间隔的参数和一个表示时间单位的参数。

//3000毫米发送一个请求 该请求包含了一个自增长的整数型变量
Observable<Long> observable = Observable.interval(3000,         TimeUnit.MILLISECONDS, Schedulers.io());

observable.subscribe(new Action1<Long>() {
    @Override
    public void call(Long i) {
        // such as printf RxIoScheduler-2call: 685
        Log.i(TAG, Thread.currentThread().getName()+"call: "+i);
    }
});

输出:

com.m520it.rxjava I/IT520: RxIoScheduler-2call: 0
com.m520it.rxjava I/IT520: RxIoScheduler-2call: 1
com.m520it.rxjava I/IT520: RxIoScheduler-2call: 2
com.m520it.rxjava I/IT520: RxIoScheduler-2call: 3
...

上面的代码在子线程中根据某种时间间隔不断打印数据。该操作类似于Timer任务,你也可以认为它是用来替换Timer定时器的。

类似的函数还有timer()函数。

Observable<Long> timer = Observable.timer(3000, 2000, TimeUnit.MILLISECONDS);
    timer.subscribe(new Action1<Long>() {
        @Override
        public void call(Long aLong) {
            Log.i(TAG, "call: "+aLong);
        }
    });

7.Range

创建一个发射特定整数序列的Observable。

Range操作符发射一个范围内的有序整数序列,你可以指定范围的起始和长度。

比如下面这个班级的所有学生都在一个集合里,现在要打印出来,可以是这样的:

private ArrayList initPersons() {
    ArrayList<Student> persons = new ArrayList<>();
    persons.add(new Student("张三", 16));
    persons.add(new Student("李四", 17));
    persons.add(new Student("王二麻子", 18));
    return persons;
}


final ArrayList<Student> students = initPersons();
//这里发射一个起始值0,长度students.size()的索引 用来遍历队列
Observable
        .range(0,students.size())
        .subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer index) {
                Log.i(TAG, "call: "+students.get(index));
            }
        });

8.Repeat

创建一个发射特定数据重复多次的Observable

Repeat重复地发射数据。某些实现允许你重复的发射某个数据序列,还有一些允许你限制重复的次数。

//设置重复发送3次
Observable<String> observable = Observable.just("Hello Android").repeat(3);

Action1<String> action1 = new Action1<String>() {

    @Override
    public void call(String s) {
        Log.i(TAG, "call: " + s);
    }
};
observable.subscribe(action1);

以上就是深入浅出RxJava_03[被观察者创建操作]的详细介绍的内容,更多相关内容请关注PHP中文网(www.php.cn)!