添加了4个基础的Subject

Author: AndyJennifer

src/main/java/Common/CommonObserver.java

@@ -14,22 +14,22 @@ public class CommonObserver implements Observer<Integer> {
 
 
     public void onSubscribe(Disposable d) {
-        System.out.println("onSubscribe: --->");
+        System.out.println("---onSubscribe---");
     }
 
     @Override
     public void onNext(Integer integer) {
-        System.out.println("Next: " + integer);
+        System.out.println("onNext: " + integer);
     }
 
     @Override
     public void onError(Throwable error) {
-        System.err.println("Error: " + error.getMessage());
+        System.err.println("onError: " + error.getMessage());
     }
 
     @Override
     public void onComplete() {
-        System.out.println("Sequence complete.");
+        System.out.println("---onComplete---");
     }
 
 

src/main/java/Observables/Subject/AsyncSubjectOpe.java

@@ -0,0 +1,40 @@
+package Observables.Subject;
+
+import io.reactivex.Observable;
+import io.reactivex.functions.Consumer;
+import io.reactivex.subjects.AsyncSubject;
+
+/**
+ * Author:  andy.xwt
+ * Date:    2019/11/20 00:28
+ * Description:
+ * 一个AsyncSubject只在原始Observable完成后，发射来自原始Observable的最后一个值。（如果原始Observable没有发射任何值，
+ * AsyncObject也不发射任何值）它会把这最后一个值发射给任何后续的观察者。
+ * 如果原始的Observable因为发生了错误而终止，AsyncSubject将不会发射任何数据，只是简单的向前传递这个错误通知。
+ * https://mcxiaoke.gitbooks.io/rxdocs/content/Subject.html
+ */
+
+
+class AsyncSubjectOpe {
+
+    public static void main(String[] args) {
+        AsyncSubject<Integer> subject = AsyncSubject.create();
+        Observable.range(1, 4).subscribe(subject);
+
+        //第一个观察者可以收到4事件
+        subject.subscribe(new Consumer<Integer>() {
+            @Override
+            public void accept(Integer integer) throws Exception {
+                System.out.println("first observer-->"+integer);
+            }
+        });
+
+        //第二个观察者只能收到4事件
+        subject.subscribe(new Consumer<Integer>() {
+            @Override
+            public void accept(Integer integer) throws Exception {
+                System.out.println("second observer-->"+integer);
+            }
+        });
+    }
+}

src/main/java/Observables/Subject/BehaviorSubjectOpe.java

@@ -0,0 +1,45 @@
+package Observables.Subject;
+
+import io.reactivex.functions.Consumer;
+import io.reactivex.subjects.BehaviorSubject;
+
+/**
+ * Author:  andy.xwt
+ * Date:    2019/11/20 00:18
+ * Description:
+ * 当Observer订阅了一个BehaviorSubject，
+ * 它一开始就会释放Observable最近释放的一个数据对象，当还没有任何数据释放时，它则是一个默认值。接下来就会释放Observable释放的所有数据。
+ * 如果Observable因异常终止，BehaviorSubject将不会向后续的Observer释放数据，但是会向Observer传递一个异常通知。
+ * https://mcxiaoke.gitbooks.io/rxdocs/content/Subject.html
+ */
+
+
+class BehaviorSubjectOpe {
+
+    public static void main(String[] args) {
+        BehaviorSubject<Integer> subject = BehaviorSubject.createDefault(0);
+
+        //第一个观察者可以收到0,1，2,3,4事件
+        subject.subscribe(new Consumer<Integer>() {
+            @Override
+            public void accept(Integer integer) throws Exception {
+                System.out.println("first observer-->" + integer);
+            }
+        });
+
+        subject.onNext(1);
+        subject.onNext(2);
+
+        //第一个观察者可以收到2,3,4事件
+        subject.subscribe(new Consumer<Integer>() {
+            @Override
+            public void accept(Integer integer) throws Exception {
+                System.out.println("second observer-->" + integer);
+            }
+        });
+        subject.onNext(3);
+        subject.onNext(4);
+
+
+    }
+}

src/main/java/Observables/Subject/PublishSubjectOpe.java

@@ -0,0 +1,43 @@
+package Observables.Subject;
+
+import io.reactivex.functions.Consumer;
+import io.reactivex.subjects.PublishSubject;
+
+/**
+ * Author:  andy.xwt
+ * Date:    2019/11/19 23:50
+ * Description:
+ * PublishSubject只会把在订阅发生的时间点之后来自原始Observable的数据发射给观察者。需要注意的是，
+ * PublishSubject可能会一创建完成就立刻开始发射数据（除非你可以阻止它发生），
+ * 因此这里有一个风险：在Subject被创建后到有观察者订阅它之前这个时间段内，一个或多个数据可能会丢失。
+ * 如果要确保来自原始Observable的所有数据都被分发，你需要这样做：
+ * 或者使用Create创建那个Observable以便手动给它引入"冷"Observable的行为（当所有观察者都已经订阅时才开始发射数据），
+ * 或者改用ReplaySubject。
+ * https://mcxiaoke.gitbooks.io/rxdocs/content/Subject.html
+ */
+
+
+class PublishSubjectOpe {
+
+    public static void main(String[] args) {
+        PublishSubject<Integer> subject = PublishSubject.create();
+        //第一个观察者可以收到1，2,3事件
+        subject.subscribe(new Consumer<Integer>() {
+            @Override
+            public void accept(Integer integer) throws Exception {
+                System.out.println("first observer-->"+integer);
+            }
+        });
+        subject.onNext(1);
+        subject.onNext(2);
+
+        //第二个观察者只能收到3事件
+        subject.subscribe(new Consumer<Integer>() {
+            @Override
+            public void accept(Integer integer) throws Exception {
+                System.out.println("second observer-->"+integer);
+            }
+        });
+        subject.onNext(3);
+    }
+}

src/main/java/Observables/Subject/ReplaySubjectOpe.java

@@ -0,0 +1,41 @@
+package Observables.Subject;
+
+import io.reactivex.Observable;
+import io.reactivex.functions.Consumer;
+import io.reactivex.subjects.ReplaySubject;
+
+/**
+ * Author:  andy.xwt
+ * Date:    2019/11/19 23:59
+ * Description:
+ * 该Subject会接收数据，当被订阅时，将所有接收到的数据全部发送给订阅者。
+ * ReplaySubject会发射所有来自原始Observable的数据给观察者，无论它们是何时订阅的。
+ * 也有其它版本的ReplaySubject，在重放缓存增长到一定大小的时候或过了一段时间后会丢弃旧的数据（原始Observable发射的）。
+ * https://mcxiaoke.gitbooks.io/rxdocs/content/Subject.html
+ */
+
+
+class ReplaySubjectOpe {
+
+    public static void main(String[] args) {
+        ReplaySubject<Integer> subject = ReplaySubject.create();
+        Observable.range(1, 4).subscribe(subject);
+
+        //第一个观察者可以收到1，2,3,4事件
+        subject.subscribe(new Consumer<Integer>() {
+            @Override
+            public void accept(Integer integer) throws Exception {
+                System.out.println("first observer-->"+integer);
+            }
+        });
+
+        //第二个观察者只能收到1,2,3,4事件
+        subject.subscribe(new Consumer<Integer>() {
+            @Override
+            public void accept(Integer integer) throws Exception {
+                System.out.println("second observer-->"+integer);
+            }
+        });
+
+    }
+}