修改了包名，并增加了Distinct相关操作符

Author: AndyJennifer

src/main/java/Common/CommonObserver.java src/main/java/common/CommonObserver.java

@@ -1,4 +1,4 @@
-package Common;
+package common;
 
 import io.reactivex.Observer;
 import io.reactivex.disposables.Disposable;

src/main/java/common/Shape.java

@@ -0,0 +1,20 @@
+package common;
+
+import com.sun.prism.paint.Color;
+
+/**
+ * Author:  andy.xwt
+ * Date:    2019/11/28 19:15
+ * Description:
+ */
+
+
+public class Shape {
+    public  Color color;
+    public String type;
+
+    public Shape(Color color, String type) {
+        this.color = color;
+        this.type = type;
+    }
+}

src/main/java/Backpressure/BackPressureOperator.java src/main/java/rxjava/backpressure/BackPressureOperator.java

@@ -1,4 +1,4 @@
-package Backpressure;
+package rxjava.backpressure;
 
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;

src/main/java/Backpressure/CommonSubscriber.java src/main/java/rxjava/backpressure/CommonSubscriber.java

@@ -1,4 +1,4 @@
-package Backpressure;
+package rxjava.backpressure;
 
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;

src/main/java/Combining/CombineLatest.java src/main/java/rxjava/combining/CombineLatest.java

@@ -1,4 +1,4 @@
-package Combining;
+package rxjava.combining;
 
 import java.util.concurrent.TimeUnit;
 

src/main/java/Combining/JoinOperator.java src/main/java/rxjava/combining/JoinOperator.java

@@ -1,4 +1,4 @@
-package Combining;
+package rxjava.combining;
 
 import java.util.concurrent.TimeUnit;
 

src/main/java/Combining/MergeOperator.java src/main/java/rxjava/combining/MergeOperator.java

@@ -1,4 +1,4 @@
-package Combining;
+package rxjava.combining;
 
 import java.util.concurrent.TimeUnit;
 

src/main/java/Combining/StartWithOperator.java src/main/java/rxjava/combining/StartWithOperator.java

@@ -1,6 +1,6 @@
-package Combining;
+package rxjava.combining;
 
-import Common.CommonObserver;
+import common.CommonObserver;
 import io.reactivex.Observable;
 
 /**

src/main/java/Combining/ZipOperator.java src/main/java/rxjava/combining/ZipOperator.java

@@ -1,4 +1,4 @@
-package Combining;
+package rxjava.combining;
 
 import java.util.concurrent.TimeUnit;
 

src/main/java/ConditionalAndBoolean/TakeUntilOperator.java src/main/java/rxjava/conditional/TakeUntilOperator.java

@@ -1,4 +1,4 @@
-package ConditionalAndBoolean;
+package rxjava.conditional;
 
 import java.util.concurrent.TimeUnit;
 

src/main/java/Creating/CreateOperator.java src/main/java/rxjava/creating/CreateOperator.java

@@ -1,9 +1,9 @@
-package Creating;
+package rxjava.creating;
 
 import io.reactivex.Observable;
 import io.reactivex.ObservableEmitter;
 import io.reactivex.ObservableOnSubscribe;
-import Common.CommonObserver;
+import common.CommonObserver;
 
 /**
  * Author:  andy.xwt

src/main/java/Creating/DeferOperator.java src/main/java/rxjava/creating/DeferOperator.java

@@ -1,10 +1,10 @@
-package Creating;
+package rxjava.creating;
 
 import java.util.concurrent.Callable;
 
 import io.reactivex.Observable;
 import io.reactivex.ObservableSource;
-import Common.CommonObserver;
+import common.CommonObserver;
 
 /**
  * Author:  andy.xwt

src/main/java/Creating/FromOperator.java src/main/java/rxjava/creating/FromOperator.java

@@ -1,10 +1,10 @@
-package Creating;
+package rxjava.creating;
 
 import java.util.Arrays;
 import java.util.List;
 
 import io.reactivex.Observable;
-import Common.CommonObserver;
+import common.CommonObserver;
 
 /**
  * Author:  andy.xwt

src/main/java/Creating/IntervalOperator.java src/main/java/rxjava/creating/IntervalOperator.java

@@ -1,4 +1,4 @@
-package Creating;
+package rxjava.creating;
 
 import java.util.concurrent.TimeUnit;
 

src/main/java/Creating/JustOperator.java src/main/java/rxjava/creating/JustOperator.java

@@ -1,4 +1,4 @@
-package Creating;
+package rxjava.creating;
 
 import io.reactivex.Observable;
 import io.reactivex.functions.Consumer;

src/main/java/Creating/NeverEmptyOperator.java src/main/java/rxjava/creating/NeverEmptyOperator.java

@@ -1,7 +1,7 @@
-package Creating;
+package rxjava.creating;
 
 import io.reactivex.Observable;
-import Common.CommonObserver;
+import common.CommonObserver;
 
 /**
  * Author:  andy.xwt

src/main/java/Creating/RangeOperator.java src/main/java/rxjava/creating/RangeOperator.java

@@ -1,7 +1,7 @@
-package Creating;
+package rxjava.creating;
 
 import io.reactivex.Observable;
-import Common.CommonObserver;
+import common.CommonObserver;
 
 /**
  * Author:  andy.xwt

src/main/java/Creating/RepeatOperator.java src/main/java/rxjava/creating/RepeatOperator.java

@@ -1,4 +1,4 @@
-package Creating;
+package rxjava.creating;
 
 import io.reactivex.Observable;
 import io.reactivex.functions.BooleanSupplier;

src/main/java/Creating/StartOperator.java src/main/java/rxjava/creating/StartOperator.java

@@ -1,9 +1,9 @@
-package Creating;
+package rxjava.creating;
 
 import java.util.concurrent.Callable;
 
 import io.reactivex.Observable;
-import Common.CommonObserver;
+import common.CommonObserver;
 
 /**
  * Author:  andy.xwt

src/main/java/Creating/TimerOperator.java src/main/java/rxjava/creating/TimerOperator.java

@@ -1,4 +1,4 @@
-package Creating;
+package rxjava.creating;
 
 import java.util.concurrent.TimeUnit;
 

src/main/java/ErrorHanding/ErrorOperator.java src/main/java/rxjava/error/ErrorOperator.java

@@ -1,6 +1,6 @@
-package ErrorHanding;
+package rxjava.error;
 
-import Common.CommonObserver;
+import common.CommonObserver;
 import io.reactivex.Observable;
 import io.reactivex.ObservableEmitter;
 import io.reactivex.ObservableOnSubscribe;

src/main/java/ErrorHanding/RetryOperator.java src/main/java/rxjava/error/RetryOperator.java

@@ -1,6 +1,6 @@
-package ErrorHanding;
+package rxjava.error;
 
-import Common.CommonObserver;
+import common.CommonObserver;
 import io.reactivex.Observable;
 import io.reactivex.ObservableEmitter;
 import io.reactivex.ObservableOnSubscribe;

src/main/java/rxjava/filtering/DistinctOperator.java

@@ -0,0 +1,126 @@
+package rxjava.filtering;
+
+
+import com.sun.prism.paint.Color;
+
+import java.util.Arrays;
+import java.util.List;
+
+import common.CommonObserver;
+import common.Shape;
+import io.reactivex.Observable;
+import io.reactivex.ObservableEmitter;
+import io.reactivex.ObservableOnSubscribe;
+import io.reactivex.functions.Consumer;
+import io.reactivex.functions.Function;
+
+/**
+ * Author:  andy.xwt
+ * Date:    2019/11/28 19:07
+ * Description:抑制（过滤掉）重复的数据项
+ * https://mcxiaoke.gitbooks.io/rxdocs/content/operators/Distinct.html
+ */
+
+
+class DistinctOperator {
+
+
+    public static List<common.Shape> mShapeLis = Arrays.asList(
+            new common.Shape(Color.RED, "circular"),//红色圆形
+            new common.Shape(Color.GREEN, "triangle"),//绿色三角形
+            new common.Shape(Color.RED, "rhombus"),//红色菱形
+            new common.Shape(Color.BLUE, "rhombus"),//蓝色菱形
+            new common.Shape(Color.BLUE, "triangle"),//蓝色三角形
+            new common.Shape(Color.WHITE, "triangle")//白色三角形
+    );
+
+
+    /**
+     * Distinct的过滤规则是：只允许还没有发射过的数据项通过。
+     * <p>
+     * 在某些实现中，有一些变体允许你调整判定两个数据不同(distinct)的标准。还有一些实现只比较一项数据和它的直接前驱，因此只会从序列中过滤掉连续重复的数据。
+     */
+    public static void testDistinct() {
+        //输出1，2，3
+        Observable.just(1, 2, 1, 1, 2, 3)
+                .distinct()
+                .subscribe(new CommonObserver());
+
+    }
+
+
+    /**
+     * 这个操作符有一个变体接受一个函数。这个函数根据原始Observable发射的数据项产生一个Key，然后，比较这些Key而不是数据本身，来判定两个数据是否是不同的。
+     * http://reactivex.io/RxJava/javadoc/rx/Observable.html#distinct(rx.functions.Func1
+     * 输出：
+     * new common.Shape(Color.RED, "circular"),//红色圆形
+     * new common.Shape(Color.GREEN, "triangle"),//绿色三角形
+     * new common.Shape(Color.RED, "rhombus"),//红色菱形
+     */
+    public static void testDistinctFun() {
+        Observable.create(new ObservableOnSubscribe<Shape>() {
+            @Override
+            public void subscribe(ObservableEmitter<Shape> emitter) throws Exception {
+                for (Shape shape : mShapeLis) {
+                    emitter.onNext(shape);
+                }
+            }
+        }).distinct(new Function<Shape, String>() {
+            @Override
+            public String apply(Shape shape) throws Exception {
+                return shape.type;//其实默认传入的是hashSet，这里返回的值就是hashset中的key
+            }
+        }).subscribe(new Consumer<Shape>() {
+            @Override
+            public void accept(Shape shape) throws Exception {
+                System.out.println(shape.type + "---->" + shape.color);
+            }
+        });
+    }
+
+    /**
+     * RxJava还是实现了一个distinctUntilChanged操作符。
+     * 它只判定一个数据和它的直接前驱是否是不同的!!!!!!!!!!!!!
+     * 它只判定一个数据和它的直接前驱是否是不同的!!!!!!!!!!!!!
+     * 它只判定一个数据和它的直接前驱是否是不同的!!!!!!!!!!!!!
+     * 和distinct(Func1)一样，根据一个函数产生的Key判定两个相邻的数据项是不是不同的
+     * distinct和distinctUntilChanged默认不在任何特定的调度器上执行。
+     * <p>
+     * 输出：
+     * new common.Shape(Color.RED, "circular"),//红色圆形
+     * new common.Shape(Color.GREEN, "triangle"),//绿色三角形
+     * new common.Shape(Color.RED, "rhombus"),//红色菱形
+     * new common.Shape(Color.BLUE, "rhombus"),//蓝色菱形
+     * new common.Shape(Color.WHITE, "triangle")//白色三角形
+     */
+    public static void testDistinctUntilChanged() {
+        Observable.create(new ObservableOnSubscribe<Shape>() {
+            @Override
+            public void subscribe(ObservableEmitter<Shape> emitter) throws Exception {
+                for (Shape shape : mShapeLis) {
+                    emitter.onNext(shape);
+                }
+            }
+            //判断的是上一个节点
+        }).distinctUntilChanged(new Function<Shape, Color>() {
+            @Override
+            public Color apply(Shape shape) throws Exception {
+                return shape.color;
+
+            }
+        }).subscribe(new Consumer<Shape>() {
+            @Override
+            public void accept(Shape shape) throws Exception {
+                System.out.println(shape.type + "---->" + shape.color);
+            }
+        });
+    }
+
+    public static void main(String[] args) {
+//        testDistinct();
+        testDistinctFun();
+//        testDistinctUntilChanged();
+    }
+
+
+}

src/main/java/Filtering/FilterOperator.java src/main/java/rxjava/filtering/FilterOperator.java

@@ -1,6 +1,6 @@
-package Filtering;
+package rxjava.filtering;
 
-import Common.CommonObserver;
+import common.CommonObserver;
 import io.reactivex.Observable;
 import io.reactivex.functions.Predicate;
 

src/main/java/Filtering/FirstOperator.java src/main/java/rxjava/filtering/FirstOperator.java

@@ -1,4 +1,4 @@
-package Filtering;
+package rxjava.filtering;
 
 import io.reactivex.Observable;
 import io.reactivex.functions.Consumer;

src/main/java/Filtering/OfTypeOperator.java src/main/java/rxjava/filtering/OfTypeOperator.java

@@ -1,4 +1,4 @@
-package Filtering;
+package rxjava.filtering;
 
 import io.reactivex.Observable;
 import io.reactivex.ObservableEmitter;

src/main/java/Filtering/SampleOperator.java src/main/java/rxjava/filtering/SampleOperator.java

@@ -1,4 +1,4 @@
-package Filtering;
+package rxjava.filtering;
 
 import java.util.concurrent.TimeUnit;
 

src/main/java/Filtering/TakeOperator.java src/main/java/rxjava/filtering/TakeOperator.java

@@ -1,8 +1,8 @@
-package Filtering;
+package rxjava.filtering;
 
 import java.util.concurrent.TimeUnit;
 
-import Common.CommonObserver;
+import common.CommonObserver;
 import io.reactivex.Observable;
 
 /**

src/main/java/Mathematical/ConcatOperator.java src/main/java/rxjava/mathematical/ConcatOperator.java

@@ -1,13 +1,13 @@
-package Mathematical;
+package rxjava.mathematical;
 
-import Common.CommonObserver;
+import common.CommonObserver;
 import io.reactivex.Observable;
 
 /**
  * Author:  andy.xwt
  * Date:    2019-02-06 00:33
  * Description:concat操作符 将多个被观察者发送的事件组合起来
- * 与{@link Combining.MergeOperator}的区别，是按照被观察者的顺序一个一个发送事件的。不会出现抢占的情况
+ * 与{@link rxjava.combining.MergeOperator}的区别，是按照被观察者的顺序一个一个发送事件的。不会出现抢占的情况
  */
 
 public class ConcatOperator {
@@ -26,7 +26,7 @@ static void testConcatArray() {
     }
     //delayError的作用是当整个事件流出现异常时，不会中断事件的传递，而是等到正常事件发送完毕后，才发送错误事件
     static void testConcatDelayError() {
-        Observable.concatArrayDelayError(Observable.range(1, 5), Observable.error(new RuntimeException("error")), Observable.range(5, 5))
+        Observable.concatArrayDelayError(Observable.range(1, 5), Observable.error(new RuntimeException("rxjava/error")), Observable.range(5, 5))
                 .subscribe(new CommonObserver());
 
     }

src/main/java/Observables/CompletableObservable.java src/main/java/rxjava/observables/CompletableObservable.java

@@ -1,6 +1,6 @@
-package Observables;
+package rxjava.observables;
 
-import Common.CommonObserver;
+import common.CommonObserver;
 import io.reactivex.Completable;
 import io.reactivex.CompletableEmitter;
 import io.reactivex.CompletableOnSubscribe;
@@ -24,7 +24,7 @@ private static void normalUse() {
             @Override
             public void subscribe(CompletableEmitter emitter) throws Exception {
 //                emitter.onComplete();
-                emitter.onError(new Throwable("error"));
+                emitter.onError(new Throwable("rxjava/error"));
             }
         }).subscribe(new Action() {
             @Override