Wiedereinführung in Betreiber mit RxJava Teil 1

Dieser Artikel ist der 10. Artikel von RxJava Adventskalender 2016.

Ich stürzte mich mit Schwung in den RxJava-Adventskalender, fragte mich aber, was ich schreiben sollte. Übrigens dachte ich, ich wäre etwas verwirrt und beschloss, die Operatoren erneut zu studieren.

Es gibt viele von ihnen, also habe ich beschlossen, sie zu teilen, also habe ich vor, sie insgesamt fünfmal zu teilen.

Zum ersten Mal werden wir uns zwei Dinge ansehen, das Erstellen von Observablen und das Transformieren von Observablen, beschrieben in ReactiveX-Einführung.

Creating Betreiber, die hauptsächlich neue "Observable" erstellen

Create Ein Bediener, der durch Scratchen "Observable" erstellt. Unter Android ist dies diejenige, die Sie häufig in der asynchronen Kommunikation sehen.

Die Verwendung ist wie folgt.

public class Create {
    public static void main(String[] args) {
        Observable.create(e -> {
            Person person = new Person();
            person.age = 100;
            person.name = "nshiba";
            e.onNext(person);
            e.onComplete();
        }).subscribe(System.out::println);
    }

    private static class Person {
        int age;
        String name;

        @Override
        public String toString() {
            return name + ":" + String.valueOf(age);
        }
    }
}

Ausgabe

nshiba:100

Es ist, als würde man einen Wert mit "onNext" übergeben und am Ende "onComplete" aufrufen.

Im Fehlerfall wird "onError" aufgerufen.

Defer defer ist der Operator, der erstellt wird, wenn Sie das auszuführende Observable abonnieren`.

Normales "Erstellen" erstellt ein "Observable", das im laufenden Betrieb ausgeführt werden kann, aber "Zurückstellen" verzögert die Erstellung des "Observable" selbst.

Observable observable = Observable.defer(() -> observer -> {
    observer.onNext("test");
    observer.onComplete();
});

//In diesem Moment können Sie ein neues Observable in defer erstellen
observable.subscribe(System.out::println);

Empty/Never/Throw Ich denke, diese Operatoren werden hauptsächlich zum Testen in begrenzten Anwendungen verwendet.

Empty Erstellen Sie ein "Observable", das keinen Wert hat, aber normal endet. Das heißt, nur "onComplete" wird aufgerufen.

Never Erstellen Sie ein "Observable", das keinen Wert hat und niemals beendet wird.

Throw Erstellt ein Observable, das keinen Wert hat, aber mit dem angegebenen Fehler beendet wird.

From Konvertiert verschiedene Objekte in Observable. Ich glaube, ich konvertiere oft Listen, also habe ich ein Beispiel mit fromArray gemacht.

int[] nums = new int[] {1, 2, 3, 4, 5};
Observable
        .fromArray(nums)
        .subscribe(ints -> {
            System.out.println("onNext");
            System.out.println(Arrays.toString(ints));
        },
        throwable -> {
            System.out.println("onError");
        },
        () -> {
            System.out.println("onComplete");
        });

Ausgabe

onNext
[1, 2, 3, 4, 5]
onComplete

Interval Erzeugt ein Observable, das in bestimmten regelmäßigen Intervallen einen ganzzahligen Wert ausgibt. Sie können auch angeben, wie viel zuerst verzögert werden soll.

Observable
        .interval(1, TimeUnit.SECONDS)
        .subscribe(System.out::print);

Ausgabe

01234567789...

Just Erzeugt "Observable" mit dem Objekt, das direkt an das Argument übergeben wird. Wenn Sie mehr als einen übergeben, wird "onNext" entsprechend aufgerufen. Wenn Sie mehr als einen übergeben, tritt kein Fehler auf, selbst wenn die Typen nicht einheitlich sind.

Observable.just(3, 1, 5, 4, "test")
        .subscribe(num -> {
            System.out.println("onNext: " + num);
        }, throwable -> {
            System.out.println("onError");
        }, () -> {
            System.out.println("onComplete");
        });

Ausgabe

onNext: 3
onNext: 1
onNext: 5
onNext: 4
onNext: test
onComplete

Range Erzeugt eine "Observable", die eine Ganzzahl im angegebenen Bereich ausgibt.

Observable.range(0, 10)
        .subscribe(i -> {
            System.out.println("onNext: " + i);
        }, throwable -> {
            System.out.println("onError");
        }, () -> {
            System.out.println("onComplete");
        });

Ausgabe

onNext: 0
onNext: 1
onNext: 2
onNext: 3
onNext: 4
onNext: 5
onNext: 6
onNext: 7
onNext: 8
onNext: 9
onComplete

Repeat Erzeugt ein "Observable", das eine bestimmte Anzahl von Malen wiederholt.

Observable.just(1, 2, 3, 4, 5)
        .repeat(3)
        .subscribe(i -> {
            System.out.println("onNext: " + i);
        }, throwable -> {
            System.out.println("onError");
        }, () -> {
            System.out.println("onComplete");
        });

Ausgabe

onNext: 1
onNext: 2
onNext: 3
onNext: 1
onNext: 2
onNext: 3
onNext: 1
onNext: 2
onNext: 3
onComplete

Start Erstellen Sie ein Observable, das den Rückgabewert einer Methode ausgibt, die einen berechneten Wert zurückgeben kann. Es gibt etwas Ähnliches wie "Erstellen", aber dieses hat einen beliebigen Rückgabewert und ruft nicht "onNext", "onComplete" auf.

Observable.fromCallable(() -> {
    String str = "java";
    str += ":" + "RxJava";
    return str;
}).subscribe(System.out::println);

Ausgabe

java:RxJava

Timer Erstellt ein "Observable", dessen Wert nach einer Verzögerung der angegebenen Zeit ausgegeben wird.

System.out.println(System.currentTimeMillis());
Observable.timer(3, TimeUnit.SECONDS)
        .subscribe(aLong -> {
            System.out.println(System.currentTimeMillis());
        });

Ausgabe

1480975677330
1480975680651

Transforming Buffer Ein Operator, der eine Liste erstellt, indem er einen Stream in bestimmten Intervallen aufteilt.

Observable.range(1, 5)
        .buffer(3)
        .subscribe(System.out::println);

Ausgabe

[1, 2, 3]
[4, 5]

FlatMap Ein Operator, der verarbeitet, was in den Stream kommt, und es dann zu einem neuen "Observable" zusammenfasst.

Observable.just(1, 2, 3)
        .flatMap(i -> Observable.range(i, i * 2))
        .subscribe(System.out::print);

Ausgabe

122345345678

GroupBy Ein Operator, der Streams gemäß den Bedingungen in Gruppen unterteilt. Wenn Sie denselben Wert für das zurückgeben, was Sie in derselben Gruppe haben möchten, befindet er sich in derselben Gruppe.

Observable.range(1, 10)
        .groupBy(integer -> integer % 3)
        .subscribe(integerIntegerGroupedObservable -> {
            integerIntegerGroupedObservable.toList().subscribe(System.out::println);
        });

Ausgabe

[3, 6, 9]
[1, 4, 7, 10]
[2, 5, 8]

Map Ein Operator, der den Wert ändern kann, der in den Stream fließt. Der Unterschied zu "FlatMap" oben besteht darin, dass "FlatMap" "Observable" und "Map" den Wert selbst zurückgibt.

Observable.just(1,2,3)
        .map(i -> i * 10)
        .subscribe(System.out::println);

Ausgabe

10
20
30

Scan Operator, der nacheinander auf die Liste zugreift. Greifen Sie zwei mal zwei zu. Zuerst werden der erste und der zweite Wert an die Argumente übergeben, und danach wird der Wert, der im vorherigen Prozess an den Rückgabewert übergeben wurde, an das erste Argument übergeben, und der nächste Wert wird an das zweite Argument übergeben. Der zum Zeitpunkt des Abonnierens übergebene Wert ist der Wert, der an das erste Element + Rückgabewert übergeben wird.

Observable.range(1, 5)
        .scan((sum, item) -> sum + item)
        .subscribe(System.out::println);

Ausgabe

1
3
6
10
15

Window Ein Operator, der einen Stream in bestimmten Intervallen aufteilt und mit dem geteilten Stream ein neues "Observable" erstellt. Ähnlich wie oben "Puffer", aber "Puffer" gibt "List " aus, während "Window" "Observable " ausgibt.

Observable.range(1,5)
        .window(3)
        .subscribe(integerObservable -> {
            integerObservable.toList().subscribe(System.out::println);
        });

Ausgabe

[1, 2, 3]
[4, 5]

Schließlich

Ich hoffe, ich kann andere Dinge wie diese tun. Der Quellcode ist öffentlich zugänglich-> nshiba / rx-samples

Wenn Sie Fehler haben, wäre ich Ihnen dankbar, wenn Sie diese in die Kommentare oder Probleme auf github schreiben könnten.