Dieses Tutorial basiert auf RxJava1 >Wenn der Beobachter Daten an den Beobachter sendet, müssen die Daten möglicherweise während der Datenübertragung weiter gefiltert werden. Das folgende Tutorial behandelt Funktionen für die meisten Filtervorgänge.
Eindeutig – doppelt gesendete Daten entfernen
Es gibt jetzt eine Klasse und die Informationen der Schüler wurden in das Schülersystem eingegeben. Es stellte sich heraus, dass dieselben Daten wiederholt wurden. Die Schülerdatenwarteschlange sieht wie folgt aus:
Die Ausgabe ist wie folgt:
private ArrayList<Student> initPersons() { ArrayList<Student> persons = new ArrayList<>(); persons.add(new Student("张三", 16)); persons.add(new Student("李四", 17)); persons.add(new Student("王二麻子", 18)); persons.add(new Student("张三", 22)); return persons; }
2.Filter
Observable .from(initPersons()) //过滤重复名字的项 .distinct(new Func1<Student, String>() { @Override public String call(Student student) { return student.name; } }) .subscribe(new Action1<Student>() { @Override public void call(Student student) { Log.i(TAG, "call: "+student); } });
IT520: call: Student{name='张三', age=16} IT520: call: Student{name='李四', age=17} IT520: call: Student{name='王二麻子', age=18}
3 .First
Observable.just(1, 2, 3, 4, 5) .filter(new Func1<Integer, Boolean>() { @Override public Boolean call(Integer item) { return( item < 4 ); } }) .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 Sequence complete.
Observable.just(1, 2, 3) .first() .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 Sequence complete.
Rufen Sie diese Funktion auf, um nur das letzte Element in der Warteschlange für gesendete Daten zu senden. Beispielcode
AusgabeÄhnliche Funktionen umfassen lastOrDefault(T)Observable.just(1, 2, 3) .last() .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: 3 Sequence complete.
Der ElementAt-Operator ruft das Datenelement an der angegebenen Indexposition der vom ursprünglichen Observable ausgegebenen Datensequenz ab und gibt es dann als eigene eindeutige Daten aus. Wenn Sie eine negative Zahl übergeben oder die Anzahl der Datenelemente im ursprünglichen Observable kleiner als index+1 ist, wird eine IndexOutOfBoundsException ausgelöst.
6.Take
Observable.just(1, 2, 3, 4, 1, 4) .elementAt(3) .subscribe(new Action1<Integer>() { @Override public void call(Integer value) { Log.i(TAG, "call: " + value); } });
com.m520it.rxjava I/IT520: call: 4
7.TakeLast
Observable.just(1, 2, 3, 4, 5, 6, 7, 8) .take(4) .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.
8.Sample
Observable.just(1, 2, 3, 4, 5, 6, 7, 8) .takeLast(4) .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: 5 Next: 6 Next: 7 Next: 8 Sequence complete.
Observable .interval(1000, TimeUnit.MILLISECONDS)//每秒发送1个数字 .sample(2000,TimeUnit.MILLISECONDS)//每2秒采样一次 .subscribe(new Action1<Long>() { @Override public void call(Long aLong) { Log.i(TAG, "call: "+aLong); } });
Observable.just(1, 2, 3, 4, 1) .ignoreElements()//不发送任何信息 直接发送onCompleted() .subscribe(new Subscriber<Integer>() { @Override public void onCompleted() { Log.i(TAG, "onCompleted: "); } @Override public void onError(Throwable e) { } @Override public void onNext(Integer integer) { Log.i(TAG, "onNext: "+integer); } });