序

本文主要研究一下reactive streams Publisher的doOn方法

doOn系列方法

这里以Flux为例 reactor-core-3.1.2.RELEASE-sources.jar!/reactor/core/publisher/Flux.java

doOnSubscribe

/**	 * Add behavior (side-effect) triggered when the {@link Flux} is subscribed.	 * 
	 * 	 * 
	 * @param onSubscribe the callback to call on {@link Subscriber#onSubscribe}	 *	 * @return an observed  {@link Flux}	 */	public final Flux
     
       doOnSubscribe(Consumer
       onSubscribe) {		Objects.requireNonNull(onSubscribe, "onSubscribe");		return doOnSignal(this, onSubscribe, null, null, null, null, null, null);	}
     

doOnRequest

/**	 * Add behavior (side-effect) triggering a {@link LongConsumer} when this {@link Flux}	 * receives any request.	 * 
	 *     Note that non fatal error raised in the callback will not be propagated and	 *     will simply trigger {@link Operators#onOperatorError(Throwable, Context)}.	 *	 * 
	 * 	 *	 * @param consumer the consumer to invoke on each request	 *	 * @return an observed  {@link Flux}	 */	public final Flux
     
       doOnRequest(LongConsumer consumer) {		Objects.requireNonNull(consumer, "consumer");		return doOnSignal(this, null, null, null, null, null, consumer, null);	}
     

doOnNext

/**	 * Add behavior (side-effect) triggered when the {@link Flux} emits an item.	 * 
	 * 	 * 
	 * @param onNext the callback to call on {@link Subscriber#onNext}	 *	 * @return an observed  {@link Flux}	 */	public final Flux
     
       doOnNext(Consumer
       onNext) {		Objects.requireNonNull(onNext, "onNext");		return doOnSignal(this, null, onNext, null, null, null, null, null);	}
     

doOnError

/**	 * Add behavior (side-effect) triggered when the {@link Flux} completes with an error.	 * 
	 * 	 * 
	 * @param onError the callback to call on {@link Subscriber#onError}	 *	 * @return an observed  {@link Flux}	 */	public final Flux
     
       doOnError(Consumer
       onError) {		Objects.requireNonNull(onError, "onError");		return doOnSignal(this, null, null, onError, null, null, null, null);	}
     

doOnEach

/**	 * Add behavior (side-effects) triggered when the {@link Flux} emits an item, fails with an error	 * or completes successfully. All these events are represented as a {@link Signal}	 * that is passed to the side-effect callback. Note that this is an advanced operator,	 * typically used for monitoring of a Flux.	 *	 * @param signalConsumer the mandatory callback to call on	 *   {@link Subscriber#onNext(Object)}, {@link Subscriber#onError(Throwable)} and	 *   {@link Subscriber#onComplete()}	 * @return an observed {@link Flux}	 * @see #doOnNext(Consumer)	 * @see #doOnError(Consumer)	 * @see #doOnComplete(Runnable)	 * @see #materialize()	 * @see Signal	 */	public final Flux
    
      doOnEach(Consumer
     > signalConsumer) {		return onAssembly(new FluxDoOnEach<>(this, signalConsumer));	}
    

doOnComplete

/**	 * Add behavior (side-effect) triggered when the {@link Flux} completes successfully.	 * 
	 * 	 * 
	 * @param onComplete the callback to call on {@link Subscriber#onComplete}	 *	 * @return an observed  {@link Flux}	 */	public final Flux
     
       doOnComplete(Runnable onComplete) {		Objects.requireNonNull(onComplete, "onComplete");		return doOnSignal(this, null, null, null, onComplete, null, null, null);	}
     

doOnCancel

/**	 * Add behavior (side-effect) triggered when the {@link Flux} is cancelled.	 * 
	 * 	 * 
	 * @param onCancel the callback to call on {@link Subscription#cancel}	 *	 * @return an observed  {@link Flux}	 */	public final Flux
     
       doOnCancel(Runnable onCancel) {		Objects.requireNonNull(onCancel, "onCancel");		return doOnSignal(this, null, null, null, null, null, null, onCancel);	}
     

doOnTerminate

/**	 * Add behavior (side-effect) triggered when the {@link Flux} terminates, either by	 * completing successfully or with an error.	 * 
	 * 	 * 
	 * @param onTerminate the callback to call on {@link Subscriber#onComplete} or {@link Subscriber#onError}	 *	 * @return an observed  {@link Flux}	 */	public final Flux
     
       doOnTerminate(Runnable onTerminate) {		Objects.requireNonNull(onTerminate, "onTerminate");		return doOnSignal(this,				null,				null,				e -> onTerminate.run(),				onTerminate,				null,				null,				null);	}
     

doOnSignal

@SuppressWarnings("unchecked")	static 
    
      Flux
     
       doOnSignal(Flux
      
        source,			@Nullable Consumer
        onSubscribe,			@Nullable Consumer
        onNext,			@Nullable Consumer
        onError,			@Nullable Runnable onComplete,			@Nullable Runnable onAfterTerminate,			@Nullable LongConsumer onRequest,			@Nullable Runnable onCancel) {		if (source instanceof Fuseable) {			return onAssembly(new FluxPeekFuseable<>(source,					onSubscribe,					onNext,					onError,					onComplete,					onAfterTerminate,					onRequest,					onCancel));		}		return onAssembly(new FluxPeek<>(source,				onSubscribe,				onNext,				onError,				onComplete,				onAfterTerminate,				onRequest,				onCancel));	}
      
     
    

实例

static class DemoSubsriber implements Subscriber
    
     {        static final Logger LOGGER = LoggerFactory.getLogger(DemoSubsriber.class);        @Override        public void onSubscribe(Subscription s) {            LOGGER.info("request max");            s.request(Integer.MAX_VALUE);        }        @Override        public void onNext(Integer integer) {            LOGGER.info("get data:{}",integer);        }        @Override        public void onError(Throwable t) {        }        @Override        public void onComplete() {            LOGGER.info("onComplete");        }    }    @Test    public void testDoOnMethods(){        Flux.range(1, 3)//                .log()                .doOnSubscribe(e -> LOGGER.info("doOnSubscribe:{}",e))                .doOnEach(e -> LOGGER.info("doOnEach:{}",e))                .doOnError(e -> LOGGER.info("doOnError:{}",e))                .doOnNext(e -> LOGGER.info("doOnNext:{}",e))                .doOnRequest(e -> LOGGER.info("doOnRequest:{}",e))                .doOnTerminate(new Runnable() {                    @Override                    public void run() {                        LOGGER.info("doOnTerminate");                    }                })                .doOnCancel(new Runnable() {                    @Override                    public void run() {                        LOGGER.info("doOnCancel");                    }                })                .doOnComplete(new Runnable() {                    @Override                    public void run() {                        LOGGER.info("doOnComplete");                    }                })                .subscribe(new DemoSubsriber());    }
    

输出

23:14:16.837 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework23:14:16.850 [main] INFO com.example.demo.FluxTest - doOnSubscribe:reactor.core.publisher.FluxRange$RangeSubscription@20398b7c23:14:16.852 [main] INFO com.example.demo.FluxTest$DemoSubsriber - request max23:14:16.852 [main] INFO com.example.demo.FluxTest - doOnRequest:214748364723:14:16.852 [main] INFO com.example.demo.FluxTest - doOnEach:reactor.core.publisher.FluxDoOnEach$DoOnEachSubscriber@228677823:14:16.852 [main] INFO com.example.demo.FluxTest - doOnNext:123:14:16.852 [main] INFO com.example.demo.FluxTest$DemoSubsriber - get data:123:14:16.852 [main] INFO com.example.demo.FluxTest - doOnEach:reactor.core.publisher.FluxDoOnEach$DoOnEachSubscriber@228677823:14:16.852 [main] INFO com.example.demo.FluxTest - doOnNext:223:14:16.852 [main] INFO com.example.demo.FluxTest$DemoSubsriber - get data:223:14:16.852 [main] INFO com.example.demo.FluxTest - doOnEach:reactor.core.publisher.FluxDoOnEach$DoOnEachSubscriber@228677823:14:16.852 [main] INFO com.example.demo.FluxTest - doOnNext:323:14:16.852 [main] INFO com.example.demo.FluxTest$DemoSubsriber - get data:323:14:16.853 [main] INFO com.example.demo.FluxTest - doOnEach:onComplete()23:14:16.853 [main] INFO com.example.demo.FluxTest - doOnTerminate23:14:16.853 [main] INFO com.example.demo.FluxTest - doOnComplete23:14:16.856 [main] INFO com.example.demo.FluxTest$DemoSubsriber - onComplete

小结

doOn系列方法是publisher的同步钩子方法，在subscriber触发一系列事件的时候触发。

doc