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();
    }