在create过程中

• emit.onComplete()

• emit.onError(e)

在create之后

• observable.filter(item -> needContinue)，相当于调用了onComplete()

• throw e, 会进入onError回调