你可以實現(xiàn)你自己的Observable操作符，本文展示怎么做。

如果你的操作符是被用于創(chuàng)造一個Observable，而不是變換或者響應一個Observable，使用 create(?) 方法，不要試圖手動實現(xiàn) Observable。另外，你可以按照下面的用法說明創(chuàng)建一個自定義的操作符。

如果你的操作符是用于Observable發(fā)射的單獨的數(shù)據(jù)項，按照下面的說明做：Sequence Operators 。如果你的操作符是用于變換Observable發(fā)射的整個數(shù)據(jù)序列，按照這個說明做：Transformational Operators 。

提示： 在一個類似于Groovy的語言Xtend中，你可以以 extension methods 的方式實現(xiàn)你自己的操作符 ,不使用本文的方法，它們也可以鏈式調(diào)用。詳情參見 RxJava and Xtend

序列操作符

下面的例子向你展示了怎樣使用lift(?)操作符將你的自定義操作符（在這個例子中是 myOperator）與標準的RxJava操作符（如ofType和map）一起使用：

fooObservable = barObservable.ofType(Integer).map({it*2}).lift(new MyOperator<T>()).map({"transformed by myOperator: " + it});

下面這部分向你展示了你的操作符的腳手架形式，以便它能正確的與lift()搭配使用。

實現(xiàn)你的操作符

將你的自定義操作符定義為實現(xiàn)了 Operator 接口的一個公開類, 就像這樣：

public class MyOperator<T> implements Operator<T> {
  public MyOperator( /* any necessary params here */ ) {
    /* 這里添加必要的初始化代碼 */
  }

  @Override
  public Subscriber<? super T> call(final Subscriber<? super T> s) {
    return new Subscriber<t>(s) {
      @Override
      public void onCompleted() {
        /* 這里添加你自己的onCompleted行為，或者僅僅傳遞完成通知： */
        if(!s.isUnsubscribed()) {
          s.onCompleted();
        }
      }

      @Override
      public void onError(Throwable t) {
        /* 這里添加你自己的onError行為, 或者僅僅傳遞錯誤通知：*/
        if(!s.isUnsubscribed()) {
          s.onError(t);
        }
      }

      @Override
      public void onNext(T item) {
        /* 這個例子對結(jié)果的每一項執(zhí)行排序操作，然后返回這個結(jié)果 */
        if(!s.isUnsubscribed()) {
          transformedItem = myOperatorTransformOperation(item);
          s.onNext(transformedItem);
        }
      }
    };
  }
}

變換操作符

下面的例子向你展示了怎樣使用 compose(?) 操作符將你得自定義操作符（在這個例子中，是一個名叫myTransformer的操作符，它將一個發(fā)射整數(shù)的Observable轉(zhuǎn)換為發(fā)射字符串的）與標準的RxJava操作符（如ofType和map）一起使用：

fooObservable = barObservable.ofType(Integer).map({it*2}).compose(new MyTransformer<Integer,String>()).map({"transformed by myOperator: " + it});

下面這部分向你展示了你的操作符的腳手架形式，以便它能正確的與compose()搭配使用。

實現(xiàn)你的變換器

將你的自定義操作符定義為實現(xiàn)了 Transformer 接口的一個公開類，就像這樣：

public class MyTransformer<Integer,String> implements Transformer<Integer,String> {
  public MyTransformer( /* any necessary params here */ ) {
    /* 這里添加必要的初始化代碼 */
  }

  @Override
  public Observable<String> call(Observable<Integer> source) {
    /* 
     * 這個簡單的例子Transformer應用一個map操作，
     * 這個map操作將發(fā)射整數(shù)變換為發(fā)射整數(shù)的字符串表示。
     */
    return source.map( new Func1<Integer,String>() {
      @Override
      public String call(Integer t1) {
        return String.valueOf(t1);
      }
    } );
  }
}

參見

• “Don’t break the chain: use RxJava’s compose() operator” by Dan Lew

其它需要考慮的

• 在發(fā)射任何數(shù)據(jù)（或者通知）給訂閱者之前，你的序列操作符可能需要檢查它的 Subscriber.isUnsubscribed(?) 狀態(tài)，如果沒有訂閱者了，沒必要浪費時間生成數(shù)據(jù)項。
• 請注意：你的序列操作符必須復合Observable協(xié)議的核心原則：
  • 它可能調(diào)用訂閱者的 onNext(?) 方法任意次，但是這些調(diào)用必須是不重疊的。
  • 它只能調(diào)用訂閱者的 onCompleted(?) 或 onError(?) 正好一次，但不能都調(diào)用，而且不能在這之后調(diào)用訂閱者的 onNext(?) 方法。
  • 如果你不能保證你得操作符遵從這兩個原則，你可以給它添加 serialize(?) 操作符，它會強制保持正確的行為。
• 請關注這里 Issue #1962 &mdash；需要有一個計劃創(chuàng)建一個測試腳手架，你可以用它來寫測試驗證你的新操作符遵從了Observable協(xié)議。
• 不要讓你的操作符阻塞別的操作。
• When possible, you should compose new operators by combining existing operators, rather than implementing them with new code. RxJava itself does this with some of its standard operators, for example:
• 如果可能，你應該組合現(xiàn)有的操作符創(chuàng)建你的新操作符，而不是從零開始實現(xiàn)它。RxJava自身的標準操作符也是這樣做的，例如：
  • first(?) 被定義為 take(1).single(?)
  • ignoreElements(?) 被定義為 filter(alwaysFalse(?))
  • reduce(a) 被定義為 scan(a).last(?)
• 如果你的操作符使用了函數(shù)或者lambda表達式作為參數(shù)，請注意它們可能是異常的來源，而且要準備好捕獲這些異常，并且使用 onError() 通知訂閱者。
  • 某些異常被認為是致命的，對它們來說，調(diào)用 onError()毫無意義，那樣或者是無用的，或者只是對問題的妥協(xié)。你可以使用 Exceptions.throwIfFatal(throwable) 方法過濾掉這些致命的異常，并重新拋出它們，而不是試圖發(fā)射關于它們的通知。
• 一般說來，一旦發(fā)生錯誤應立即通知訂閱者，而不是首先嘗試發(fā)射更多的數(shù)據(jù)。
• 請注意 null 可能是Observable發(fā)射的一個合法數(shù)據(jù)。頻繁發(fā)生錯誤的一個來源是：測試一些變量并且將持有一個非 null 值作為是否發(fā)射了數(shù)據(jù)的替代。一個值為 null 的數(shù)據(jù)仍然是一個發(fā)射數(shù)據(jù)項，它與沒有發(fā)射任何東西是不能等同的。
• 想讓你的操作符在反壓(backpressure)場景中變得得好可能會非常棘手?？梢詤⒖糄ávid Karnok的博客 Advanced RxJava，這里有一個涉及到的各種因素和怎樣處理它們的很值得看的討論。