Java 类com.lmax.disruptor.ExceptionHandler 实例源码

@Test
public void shouldSupportSpecifyingAExceptionHandlerForEventProcessors()
    throws Exception
{
    AtomicReference<Throwable> eventHandled = new AtomicReference<Throwable>();
    ExceptionHandler exceptionHandler = new StubExceptionHandler(eventHandled);
    RuntimeException testException = new RuntimeException();
    ExceptionThrowingEventHandler handler = new ExceptionThrowingEventHandler(testException);

    disruptor.handleExceptionsWith(exceptionHandler);
    disruptor.handleEventsWith(handler);

    publishEvent();

    final Throwable actualException = waitFor(eventHandled);
    assertSame(testException, actualException);
}

@Test
public void shouldOnlyApplyExceptionsHandlersSpecifiedViaHandleExceptionsWithOnNewEventProcessors()
    throws Exception
{
    AtomicReference<Throwable> eventHandled = new AtomicReference<Throwable>();
    ExceptionHandler exceptionHandler = new StubExceptionHandler(eventHandled);
    RuntimeException testException = new RuntimeException();
    ExceptionThrowingEventHandler handler = new ExceptionThrowingEventHandler(testException);

    disruptor.handleExceptionsWith(exceptionHandler);
    disruptor.handleEventsWith(handler);
    disruptor.handleExceptionsWith(new FatalExceptionHandler());

    publishEvent();

    final Throwable actualException = waitFor(eventHandled);
    assertSame(testException, actualException);
}

@Test
public void shouldSupportSpecifyingADefaultExceptionHandlerForEventProcessors()
    throws Exception
{
    AtomicReference<Throwable> eventHandled = new AtomicReference<Throwable>();
    ExceptionHandler exceptionHandler = new StubExceptionHandler(eventHandled);
    RuntimeException testException = new RuntimeException();
    ExceptionThrowingEventHandler handler = new ExceptionThrowingEventHandler(testException);

    disruptor.setDefaultExceptionHandler(exceptionHandler);
    disruptor.handleEventsWith(handler);

    publishEvent();

    final Throwable actualException = waitFor(eventHandled);
    assertSame(testException, actualException);
}

@Test
public void shouldApplyDefaultExceptionHandlerToExistingEventProcessors()
    throws Exception
{
    AtomicReference<Throwable> eventHandled = new AtomicReference<Throwable>();
    ExceptionHandler exceptionHandler = new StubExceptionHandler(eventHandled);
    RuntimeException testException = new RuntimeException();
    ExceptionThrowingEventHandler handler = new ExceptionThrowingEventHandler(testException);

    disruptor.handleEventsWith(handler);
    disruptor.setDefaultExceptionHandler(exceptionHandler);

    publishEvent();

    final Throwable actualException = waitFor(eventHandled);
    assertSame(testException, actualException);
}

public SiddhiContext() {
    SiddhiExtensionLoader.loadSiddhiExtensions(siddhiExtensions);
    siddhiDataSources = new ConcurrentHashMap<String, DataSource>();
    statisticsConfiguration = new StatisticsConfiguration(new SiddhiMetricsFactory());
    extensionHolderMap = new ConcurrentHashMap<Class, AbstractExtensionHolder>();
    configManager = new InMemoryConfigManager();
    defaultDisrupterExceptionHandler = new ExceptionHandler<Object>() {
        @Override
        public void handleEventException(Throwable throwable, long l, Object event) {
            log.error("Disruptor encountered an error processing" + " [sequence: " + l + ", event: " + event
                    .toString() + "]", throwable);
        }

        @Override
        public void handleOnStartException(Throwable throwable) {
            log.error("Disruptor encountered an error on start", throwable);
        }

        @Override
        public void handleOnShutdownException(Throwable throwable) {
            log.error("Disruptor encountered an error on shutdown", throwable);
        }
    };
}

private static ExceptionHandler getExceptionHandler() {
    final String cls = System.getProperty("AsyncLogger.ExceptionHandler");
    if (cls == null) {
        LOGGER.debug("No AsyncLogger.ExceptionHandler specified");
        return null;
    }
    try {
        @SuppressWarnings("unchecked")
        final
        Class<? extends ExceptionHandler> klass = (Class<? extends ExceptionHandler>) Class
                .forName(cls);
        final ExceptionHandler result = klass.newInstance();
        LOGGER.debug("AsyncLogger.ExceptionHandler=" + result);
        return result;
    } catch (final Exception ignored) {
        LOGGER.debug(
                "AsyncLogger.ExceptionHandler not set: error creating "
                        + cls + ": ", ignored);
        return null;
    }
}

private static synchronized void initDisruptor() {
    if (disruptor != null) {
        LOGGER.trace("AsyncLoggerConfigHelper not starting new disruptor, using existing object. Ref count is {}.", count);
        return;
    }
    LOGGER.trace("AsyncLoggerConfigHelper creating new disruptor. Ref count is {}.", count);
    final int ringBufferSize = calculateRingBufferSize();
    final WaitStrategy waitStrategy = createWaitStrategy();
    executor = Executors.newSingleThreadExecutor(threadFactory);
    disruptor = new Disruptor<Log4jEventWrapper>(FACTORY, ringBufferSize,
            executor, ProducerType.MULTI, waitStrategy);
    final EventHandler<Log4jEventWrapper>[] handlers = new Log4jEventWrapperHandler[] {//
    new Log4jEventWrapperHandler() };
    final ExceptionHandler errorHandler = getExceptionHandler();
    disruptor.handleExceptionsWith(errorHandler);
    disruptor.handleEventsWith(handlers);

    LOGGER.debug(
            "Starting AsyncLoggerConfig disruptor with ringbuffer size={}, waitStrategy={}, exceptionHandler={}...",
            disruptor.getRingBuffer().getBufferSize(), waitStrategy.getClass().getSimpleName(), errorHandler);
    disruptor.start();
}

private static ExceptionHandler getExceptionHandler() {
    final String cls = System
            .getProperty("AsyncLoggerConfig.ExceptionHandler");
    if (cls == null) {
        return null;
    }
    try {
        @SuppressWarnings("unchecked")
        final Class<? extends ExceptionHandler> klass = (Class<? extends ExceptionHandler>) Class
                .forName(cls);
        final ExceptionHandler result = klass.newInstance();
        return result;
    } catch (final Exception ignored) {
        LOGGER.debug(
                "AsyncLoggerConfig.ExceptionHandler not set: error creating "
                        + cls + ": ", ignored);
        return null;
    }
}

/**
 * <p>Specify an exception handler to be used for event handlers and worker pools created by this Disruptor.</p>
 * <p>
 * <p>The exception handler will be used by existing and future event handlers and worker pools created by this Disruptor instance.</p>
 *
 * @param exceptionHandler the exception handler to use.
 */
@SuppressWarnings("unchecked")
public void setDefaultExceptionHandler(final ExceptionHandler<? super T> exceptionHandler) {
    checkNotStarted();
    if (!(this.exceptionHandler instanceof ExceptionHandlerWrapper)) {
        throw new IllegalStateException("setDefaultExceptionHandler can not be used after handleExceptionsWith");
    }
    ((ExceptionHandlerWrapper<T>) this.exceptionHandler).switchTo(exceptionHandler);
}

/**
 * Specify the {@link ExceptionHandler} to use with the event handler.
 *
 * @param exceptionHandler the exception handler to use.
 */
public void with(ExceptionHandler<? super T> exceptionHandler)
{
    ((BatchEventProcessor<T>) consumerRepository.getEventProcessorFor(eventHandler))
        .setExceptionHandler(exceptionHandler);
    consumerRepository.getBarrierFor(eventHandler).alert();
}

/**
 * @param ring             source byte ring containing inbound messages
 * @param sequenceBarrier  on which it is waiting.
 * @param delegate   is the delegate to which message are dispatched.
 * @param exceptionHandler to be called back when an error occurs
 *                         as {@link com.lmax.disruptor.Sequencer#INITIAL_CURSOR_VALUE}
 */
public AbstractByteRingConsumerEx(ByteRing ring,
                                  SequenceBarrier sequenceBarrier,
                                  RingBufferBlockProcessor delegate,
                                  ExceptionHandler exceptionHandler) {
    super(ring, sequenceBarrier);
    this.delegate = delegate;
    this.exceptionHandler = exceptionHandler;
}

public static RingBufferBlockProcessor createLogger(File file, int bufferSize, ExceptionHandler exceptionHandler) {
    try {
        OutputStream os = new BufferedOutputStream(new FileOutputStream(file, true), bufferSize);
        BufferLogger result = new BufferLogger(os, exceptionHandler);
        new Thread(result, "FIX Log Flusher").start();
        return result;
    } catch (FileNotFoundException e) {
        throw new RuntimeException("Cannot write into " + file, e);
    }
}

public ExceptionHandler<Object> getDisruptorExceptionHandler() {
    if (disruptorExceptionHandler != null) {
        return disruptorExceptionHandler;
    } else {
        return siddhiContext.getDefaultDisrupterExceptionHandler();
    }
}

/**
 * Set a new {@link ExceptionHandler} for handling exceptions propagated out of the {@link com.lmax.disruptor.BatchEventProcessor}
 *
 * @param exceptionHandler to replace the existing exceptionHandler.
 */
public void setExceptionHandler(final ExceptionHandler exceptionHandler) {
    if (null == exceptionHandler) {
        throw new NullPointerException("Exception handler cannot be null.");
    }

    this.exceptionHandler = exceptionHandler;
}

/**
 * Increases the reference count and creates and starts a new Disruptor and associated thread if none currently
 * exists.
 *
 * @see #stop()
 */
@Override
public synchronized void start() {
    if (disruptor != null) {
        LOGGER.trace("AsyncLoggerConfigDisruptor not starting new disruptor for this configuration, "
                + "using existing object.");
        return;
    }
    LOGGER.trace("AsyncLoggerConfigDisruptor creating new disruptor for this configuration.");
    ringBufferSize = DisruptorUtil.calculateRingBufferSize("AsyncLoggerConfig.RingBufferSize");
    final WaitStrategy waitStrategy = DisruptorUtil.createWaitStrategy("AsyncLoggerConfig.WaitStrategy");

    final ThreadFactory threadFactory = new Log4jThreadFactory("AsyncLoggerConfig-", true, Thread.NORM_PRIORITY) {
        @Override
        public Thread newThread(final Runnable r) {
            final Thread result = super.newThread(r);
            backgroundThreadId = result.getId();
            return result;
        }
    };
    asyncQueueFullPolicy = AsyncQueueFullPolicyFactory.create();

    translator = mutable ? MUTABLE_TRANSLATOR : TRANSLATOR;
    factory = mutable ? MUTABLE_FACTORY : FACTORY;
    disruptor = new Disruptor<>(factory, ringBufferSize, threadFactory, ProducerType.MULTI, waitStrategy);

    final ExceptionHandler<Log4jEventWrapper> errorHandler = DisruptorUtil.getAsyncLoggerConfigExceptionHandler();
    disruptor.setDefaultExceptionHandler(errorHandler);

    final Log4jEventWrapperHandler[] handlers = {new Log4jEventWrapperHandler()};
    disruptor.handleEventsWith(handlers);

    LOGGER.debug("Starting AsyncLoggerConfig disruptor for this configuration with ringbufferSize={}, "
            + "waitStrategy={}, exceptionHandler={}...", disruptor.getRingBuffer().getBufferSize(), waitStrategy
            .getClass().getSimpleName(), errorHandler);
    disruptor.start();
    super.start();
}

public void switchTo(final ExceptionHandler<? super T> exceptionHandler)
{
    this.delegate = exceptionHandler;
}

public RingBuffer2StreamProcessor(OutputStream os, ExceptionHandler exceptionHandler) {
    this.os = os;
    this.exceptionHandler = exceptionHandler;
}

protected BufferLogger(OutputStream os, ExceptionHandler exceptionHandler) {
    super(os, exceptionHandler);
}

public void handleExceptionWith(ExceptionHandler<Object> exceptionHandler) {
    siddhiAppContext.setDisruptorExceptionHandler(exceptionHandler);
}

public ExceptionHandler<Object> getDefaultDisrupterExceptionHandler() {
    return defaultDisrupterExceptionHandler;
}

public void setDisruptorExceptionHandler(ExceptionHandler<Object> disruptorExceptionHandler) {
    this.disruptorExceptionHandler = disruptorExceptionHandler;
}

public final void setExceptionHandler(ExceptionHandler<LogEvent<E>> exceptionHandler) {
    this.exceptionHandler = exceptionHandler;
}

/**
 * Specify the {@link ExceptionHandler} to use with the event handler.
 *
 * @param exceptionHandler the exception handler to use.
 */
public void with(ExceptionHandler exceptionHandler)
{
    ((BatchEventProcessor<?>) consumerRepository.getEventProcessorFor(eventHandler)).setExceptionHandler(exceptionHandler);
    consumerRepository.getBarrierFor(eventHandler).alert();
}

/**
 * Creates and starts a new Disruptor and associated thread if none currently exists.
 *
 * @see #stop()
 */
@Override
public synchronized void start() {
    if (disruptor != null) {
        LOGGER.trace(
                "[{}] AsyncLoggerDisruptor not starting new disruptor for this context, using existing object.",
                contextName);
        return;
    }
    LOGGER.trace("[{}] AsyncLoggerDisruptor creating new disruptor for this context.", contextName);
    ringBufferSize = DisruptorUtil.calculateRingBufferSize("AsyncLogger.RingBufferSize");
    final WaitStrategy waitStrategy = DisruptorUtil.createWaitStrategy("AsyncLogger.WaitStrategy");

    final ThreadFactory threadFactory = new Log4jThreadFactory("AsyncLogger[" + contextName + "]", true, Thread.NORM_PRIORITY) {
        @Override
        public Thread newThread(final Runnable r) {
            final Thread result = super.newThread(r);
            backgroundThreadId = result.getId();
            return result;
        }
    };
    asyncQueueFullPolicy = AsyncQueueFullPolicyFactory.create();

    disruptor = new Disruptor<>(RingBufferLogEvent.FACTORY, ringBufferSize, threadFactory, ProducerType.MULTI,
            waitStrategy);

    final ExceptionHandler<RingBufferLogEvent> errorHandler = DisruptorUtil.getAsyncLoggerExceptionHandler();
    disruptor.setDefaultExceptionHandler(errorHandler);

    final RingBufferLogEventHandler[] handlers = {new RingBufferLogEventHandler()};
    disruptor.handleEventsWith(handlers);

    LOGGER.debug("[{}] Starting AsyncLogger disruptor for this context with ringbufferSize={}, waitStrategy={}, "
            + "exceptionHandler={}...", contextName, disruptor.getRingBuffer().getBufferSize(), waitStrategy
            .getClass().getSimpleName(), errorHandler);
    disruptor.start();

    LOGGER.trace("[{}] AsyncLoggers use a {} translator", contextName, useThreadLocalTranslator ? "threadlocal"
            : "vararg");
    super.start();
}

/**
 * <p>Specify an exception handler to be used for any future event handlers.</p>
 * <p>
 * <p>Note that only event handlers set up after calling this method will use the exception handler.</p>
 *
 * @param exceptionHandler the exception handler to use for any future {@link EventProcessor}.
 * @deprecated This method only applies to future event handlers. Use setDefaultExceptionHandler instead which applies to existing and new event handlers.
 */
public void handleExceptionsWith(final ExceptionHandler<? super T> exceptionHandler) {
    this.exceptionHandler = exceptionHandler;
}

/**
 * @param ring             source byte ring containing inbound messages
 * @param sequenceBarrier  on which it is waiting.
 * @param delegate         is the delegate to which message are dispatched.
 * @param exceptionHandler to be called back when an error occurs
 *                         as {@link com.lmax.disruptor.Sequencer#INITIAL_CURSOR_VALUE}
 */
public VariableBlockSizeRingConsumer(ByteRing ring, SequenceBarrier sequenceBarrier, RingBufferBlockProcessor delegate, ExceptionHandler exceptionHandler) {
    super(ring, sequenceBarrier, delegate, exceptionHandler);
}

/**
     * @param ring             source byte ring containing inbound messages
     * @param sequenceBarrier  on which it is waiting.
     * @param delegate         is the delegate to which message are dispatched.
     * @param exceptionHandler to be called back when an error occurs
*                         as {@link com.lmax.disruptor.Sequencer#INITIAL_CURSOR_VALUE}
     * @param blockSize
     */
    public FixedBlockSizeRingConsumer(ByteRing ring, SequenceBarrier sequenceBarrier, RingBufferBlockProcessor delegate, ExceptionHandler exceptionHandler, int blockSize) {
        super(ring, sequenceBarrier, delegate, exceptionHandler);
        this.blockSize = blockSize;
    }

/**
 * Specify an exception handler to be used for any future event handlers.<p/>
 *
 * Note that only event handlers set up after calling this method will use the exception handler.
 *
 * @param exceptionHandler the exception handler to use for any future {@link EventProcessor}.
 */
public void handleExceptionsWith(final ExceptionHandler exceptionHandler) {
    this.exceptionHandler = exceptionHandler;
}