diff --git a/apm-collector/apm-collector-queue/collector-queue-disruptor-provider/src/main/java/org/skywalking/apm/collector/queue/disruptor/base/DisruptorQueueCreator.java b/apm-collector/apm-collector-queue/collector-queue-disruptor-provider/src/main/java/org/skywalking/apm/collector/queue/disruptor/base/DisruptorQueueCreator.java index c63cff78aa6e91dc03b8a410fbb79debbc777787..52082a2ce619b8d0c7aae06047dc216a6346e327 100644 --- a/apm-collector/apm-collector-queue/collector-queue-disruptor-provider/src/main/java/org/skywalking/apm/collector/queue/disruptor/base/DisruptorQueueCreator.java +++ b/apm-collector/apm-collector-queue/collector-queue-disruptor-provider/src/main/java/org/skywalking/apm/collector/queue/disruptor/base/DisruptorQueueCreator.java @@ -18,6 +18,7 @@ package org.skywalking.apm.collector.queue.disruptor.base; +import com.lmax.disruptor.ExceptionHandler; import com.lmax.disruptor.RingBuffer; import com.lmax.disruptor.dsl.Disruptor; import org.skywalking.apm.collector.queue.base.DaemonThreadFactory; @@ -25,12 +26,16 @@ import org.skywalking.apm.collector.queue.base.MessageHolder; import org.skywalking.apm.collector.queue.base.QueueCreator; import org.skywalking.apm.collector.queue.base.QueueEventHandler; import org.skywalking.apm.collector.queue.base.QueueExecutor; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * @author peng-yongsheng */ public class DisruptorQueueCreator implements QueueCreator { + private final Logger logger = LoggerFactory.getLogger(DisruptorQueueCreator.class); + @Override public QueueEventHandler create(int queueSize, QueueExecutor executor) { // Specify the size of the ring buffer, must be power of 2. if (!((((queueSize - 1) & queueSize) == 0) && queueSize != 0)) { @@ -39,6 +44,19 @@ public class DisruptorQueueCreator implements QueueCreator { // Construct the Disruptor Disruptor disruptor = new Disruptor<>(MessageHolderFactory.INSTANCE, queueSize, DaemonThreadFactory.INSTANCE); + disruptor.setDefaultExceptionHandler(new ExceptionHandler() { + @Override public void handleEventException(Throwable ex, long sequence, MessageHolder event) { + logger.error("Handle disruptor error event! message: {}.", event.getMessage(), ex); + } + + @Override public void handleOnStartException(Throwable ex) { + logger.error("create disruptor queue failed!", ex); + } + + @Override public void handleOnShutdownException(Throwable ex) { + logger.error("shutdown disruptor queue failed!", ex); + } + }); RingBuffer ringBuffer = disruptor.getRingBuffer(); DisruptorEventHandler eventHandler = new DisruptorEventHandler(ringBuffer, executor);