聊聊reactor-logback的AsyncAppender

public class AsyncAppender extends ContextAwareBase
		implements Appender<ILoggingEvent>, AppenderAttachable<ILoggingEvent>,
		           CoreSubscriber<ILoggingEvent> {

	private final AppenderAttachableImpl<ILoggingEvent>    aai      =
			new AppenderAttachableImpl<ILoggingEvent>();
	private final FilterAttachableImpl<ILoggingEvent>      fai      =
			new FilterAttachableImpl<ILoggingEvent>();
	private final AtomicReference<Appender<ILoggingEvent>> delegate =
			new AtomicReference<Appender<ILoggingEvent>>();

	private String                            name;
	private WorkQueueProcessor<ILoggingEvent> processor;

	private int     backlog           = 1024 * 1024;
	private boolean includeCallerData = false;
	private boolean started           = false;

	//......
}	

public interface CoreSubscriber<T> extends Subscriber<T> {

	/**
	 * Request a {@link Context} from dependent components which can include downstream
	 * operators during subscribing or a terminal {@link org.reactivestreams.Subscriber}.
	 *
	 * @return a resolved context or {@link Context#empty()}
	 */
	default Context currentContext(){
		return Context.empty();
	}

	/**
	 * Implementors should initialize any state used by {@link #onNext(Object)} before
	 * calling {@link Subscription#request(long)}. Should further {@code onNext} related
	 * state modification occur, thread-safety will be required.
	 * <p>
	 *    Note that an invalid request {@code <= 0} will not produce an onError and
	 *    will simply be ignored or reported through a debug-enabled
	 *    {@link reactor.util.Logger}.
	 *
	 * {@inheritDoc}
	 */
	@Override
	void onSubscribe(Subscription s);
}

	public void onSubscribe(Subscription s) {
		try {
			doStart();
		}
		catch (Throwable t) {
			addError(t.getMessage(), t);
		}
		finally {
			started = true;
			s.request(Long.MAX_VALUE);
		}
	}

	protected void doStart() {
	}	

	public void onNext(ILoggingEvent iLoggingEvent) {
		aai.appendLoopOnAppenders(iLoggingEvent);
	}

	public void onError(Throwable t) {
		addError(t.getMessage(), t);
	}

	public void onComplete() {
		try {
			Appender<ILoggingEvent> appender = delegate.getAndSet(null);
			if (appender != null){
				doStop();
				appender.stop();
				aai.detachAndStopAllAppenders();
			}
		}
		catch (Throwable t) {
			addError(t.getMessage(), t);
		}
		finally {
			started = false;
		}
	}

	protected void doStop() {
	}	

	public void doAppend(ILoggingEvent evt) throws LogbackException {
		if (getFilterChainDecision(evt) == FilterReply.DENY) {
			return;
		}
		evt.prepareForDeferredProcessing();
		if (includeCallerData) {
			evt.getCallerData();
		}
		try {
			queueLoggingEvent(evt);
		}
		catch (Throwable t) {
			addError(t.getMessage(), t);
		}
	}

	protected void queueLoggingEvent(ILoggingEvent evt) {
		if (null != delegate.get()) {
			processor.onNext(evt);
		}
	}	

	public void start() {
		startDelegateAppender();

		processor = WorkQueueProcessor.<ILoggingEvent>builder().name("logger")
		                                                       .bufferSize(backlog)
		                                                       .autoCancel(false)
		                                                       .build();
		processor.subscribe(this);
	}

	private void startDelegateAppender() {
		Appender<ILoggingEvent> delegateAppender = delegate.get();
		if (null != delegateAppender && !delegateAppender.isStarted()) {
			delegateAppender.start();
		}
	}

	public void addAppender(Appender<ILoggingEvent> newAppender) {
		if (delegate.compareAndSet(null, newAppender)) {
			aai.addAppender(newAppender);
		}
		else {
			throw new IllegalArgumentException(delegate.get() + " already attached.");
		}
	}		

	public void stop() {
		processor.onComplete();
	}