add Disruptor

4dkankan-utils-disruptormq/pom.xml

@@ -0,0 +1,43 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>4dkankan-utils</artifactId>
+        <groupId>com.fdkankan</groupId>
+        <version>3.0.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>4dkankan-utils-disruptormq</artifactId>
+
+    <properties>
+        <disruptor.version>3.4.4</disruptor.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+            <version>1.2.11</version>
+        </dependency>
+        <dependency>
+            <groupId>com.lmax</groupId>
+            <artifactId>disruptor</artifactId>
+            <version>${disruptor.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <optional>true</optional>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-autoconfigure</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter</artifactId>
+        </dependency>
+    </dependencies>
+</project>

4dkankan-utils-disruptormq/src/main/java/com/fdkankan/disruptormq/BaseDisruptorQueue.java

@@ -0,0 +1,84 @@
+package com.fdkankan.disruptormq;
+
+/**
+ * @author Xiewj
+ * @date 2024/1/17
+ */
+
+import com.fdkankan.disruptormq.DisruptorExcepitonHandler;
+import com.lmax.disruptor.dsl.Disruptor;
+import com.lmax.disruptor.dsl.ProducerType;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
+import org.springframework.util.CollectionUtils;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * 可以在sprinboot中使用，不同业务可以声明不同的DisruptorQueue，使用不同的Handler处理
+ */
+@SuppressWarnings("all")
+public abstract class BaseDisruptorQueue<T> implements ApplicationContextAware {
+    @Autowired
+    private  DisruptorProperties disruptorProperties;
+    private Disruptor< DisruptorEvent<T>> disruptor;
+    private Map<String, DisruptorHandler> disruptorHandlerMap;
+
+    public void offer(T t) {
+        if (Objects.nonNull(t)) {
+            disruptor.publishEvent((event, sequence) -> event.setPayload(t));
+        }
+    }
+
+    public void offer(List<T> ts) {
+        if (!CollectionUtils.isEmpty(ts)) {
+            ts.stream().filter(Objects::nonNull).forEach(this::offer);
+        }
+    }
+
+    protected void initDisruptor(String handlerName) {
+        this.disruptor = getDisruptor(handlerName);
+    }
+
+    private Disruptor<DisruptorEvent<T>> getDisruptor(String handlerName) {
+        DisruptorProperties.HandlerProperties handlerProperties = getDisruptorHandlerConfig(handlerName);
+        Disruptor<DisruptorEvent<T>> disruptor = new Disruptor<>(DisruptorEvent::new, handlerProperties.getRingBufferSize(), //
+                new CustomizableThreadFactory(handlerName + "-disruptor-"), //
+                handlerProperties.isMultiProducer() ? ProducerType.MULTI : ProducerType.SINGLE, //
+                disruptorProperties.getWaitType().getWaitStrategy());
+        // 对于一条消息，handleEventsWith方法返回的EventHandlerGroup，Group中的每个消费者都会对消息进行消费，各个消费者之间不存在竞争（类似发布订阅）
+        // disruptor.handleEventsWith(getDisruptorHandler(handlerName));
+        // 对于一条消息，handleEventsWithWorkerPool方法返回的EventHandlerGroup，Group的消费者对于同一条消息不重复消费（类似点对点）
+        disruptor.handleEventsWithWorkerPool(getDisruptorHandler(handlerName));
+        disruptor.setDefaultExceptionHandler(new DisruptorExcepitonHandler<>());
+        disruptor.start();
+        Runtime.getRuntime().addShutdownHook(new Thread(disruptor::shutdown, handlerName + "-disruptor-shutdown"));
+        return disruptor;
+    }
+
+    private DisruptorProperties.HandlerProperties getDisruptorHandlerConfig(String handlerName) {
+        Map<String, DisruptorProperties.HandlerProperties> handlers = disruptorProperties.getHandlers();
+        DisruptorProperties.HandlerProperties handlerProperties = handlers.get(handlerName);
+        return Optional.ofNullable(handlerProperties).orElseGet(() -> {
+            DisruptorProperties.HandlerProperties defaultHandlerProperties = new DisruptorProperties.HandlerProperties();
+            defaultHandlerProperties.setRingBufferSize(1024 * 1024);
+            defaultHandlerProperties.setMultiProducer(false);
+            return defaultHandlerProperties;
+        });
+    }
+
+    private DisruptorHandler<T> getDisruptorHandler(String handlerName) {
+        return disruptorHandlerMap.get(handlerName);
+    }
+
+    @Override
+    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+        disruptorHandlerMap = applicationContext.getBeansOfType(DisruptorHandler.class);
+    }
+}

4dkankan-utils-disruptormq/src/main/java/com/fdkankan/disruptormq/DisruptorEvent.java

@@ -0,0 +1,12 @@
+package com.fdkankan.disruptormq;
+
+import lombok.Data;
+
+/**
+ * @author Xiewj
+ * @date 2024/1/17
+ */
+@Data
+public class DisruptorEvent<T> {
+    private T payload;
+}

4dkankan-utils-disruptormq/src/main/java/com/fdkankan/disruptormq/DisruptorExcepitonHandler.java

@@ -0,0 +1,26 @@
+package com.fdkankan.disruptormq;
+
+import com.lmax.disruptor.ExceptionHandler;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * @author Xiewj
+ * @date 2024/1/17
+ */
+@Slf4j
+public class DisruptorExcepitonHandler<T> implements ExceptionHandler<T> {
+    @Override
+    public void handleEventException(Throwable ex, long sequence, T event) {
+        log.error("DisruptorEvent Handle处理错误，sequence={}，event={}，e={}", sequence, event.toString(), ex.toString(), ex);
+    }
+
+    @Override
+    public void handleOnStartException(Throwable ex) {
+        log.error("DisruptorEvent handleOnStart处理错误，e={}", ex.toString(), ex);
+    }
+
+    @Override
+    public void handleOnShutdownException(Throwable ex) {
+        log.error("DisruptorEvent handleOnShutdown处理错误，e={}", ex.toString(), ex);
+    }
+}

4dkankan-utils-disruptormq/src/main/java/com/fdkankan/disruptormq/DisruptorHandler.java

@@ -0,0 +1,29 @@
+package com.fdkankan.disruptormq;
+
+/**
+ * @author Xiewj
+ * @date 2024/1/17
+ */
+
+import com.fdkankan.disruptormq.DisruptorEvent;
+import com.lmax.disruptor.EventHandler;
+import com.lmax.disruptor.WorkHandler;
+
+/**
+ * 消费者接口
+ * EventHandler用于EventHandlerGroup，WorkHandler用于WorkPool
+ * 同时实现两接口，该类对象可同时用于EventHandlerGroup和WorkPool
+ */
+public interface DisruptorHandler<T> extends EventHandler<DisruptorEvent<T>>, WorkHandler<DisruptorEvent<T>> {
+    @Override
+    default void onEvent(DisruptorEvent<T> event, long sequence, boolean endOfBatch) throws Exception {
+        onEvent(event);
+    }
+
+    @Override
+    default void onEvent(DisruptorEvent<T> event) throws Exception {
+        consumer(event.getPayload());
+    }
+
+    void consumer(T t) throws Exception;
+}

4dkankan-utils-disruptormq/src/main/java/com/fdkankan/disruptormq/DisruptorProperties.java

@@ -0,0 +1,73 @@
+package com.fdkankan.disruptormq;
+
+import com.lmax.disruptor.*;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.Getter;
+import lombok.ToString;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * @author Xiewj
+ * @date 2024/1/17
+ */
+@Data
+@Component
+@ConditionalOnProperty(prefix  = "disruptor",name = "wait-type")
+public class DisruptorProperties {
+    /**
+     * 等待类型
+     */
+    private WaitType waitType = WaitType.BLOCKING_WAIT;
+    /**
+     * handler配置参数
+     */
+    private Map<String, HandlerProperties> handlers = new LinkedHashMap<>();
+
+    @Data
+    public static class HandlerProperties {
+        /**
+         * 缓冲区大小，默认1024*1024，必须是2的N次幂
+         */
+        private int ringBufferSize = 1024 * 1024;
+        /**
+         * 单生产者/多生产者
+         */
+        private boolean multiProducer = false;
+    }
+
+    /**
+     * 决定一个消费者将如何等待生产者将Event置入Disruptor的策略，用来权衡当生产者无法将新的事件放进RingBuffer时的处理策略
+     * （例如：当生产者太快，消费者太慢，会导致生成者获取不到新的事件槽来插入新事件，则会根据该策略进行处理，默认会堵塞）
+     */
+    @Getter
+    @ToString
+    @AllArgsConstructor
+    public enum WaitType {
+        /**
+         * BlockingWaitStrategy是最低效的策略，但其对CPU的消耗最小并且在各种不同部署环境中能提供更加一致的性能表现
+         */
+        BLOCKING_WAIT(new BlockingWaitStrategy()),
+        /**
+         * SleepingWaitStrategy的性能表现跟BlockingWaitStrategy差不多，对CPU的消耗也类似，但其对生产者线程的影响最小，适合用于异步日志类似的场景
+         */
+        SLEEPING_WAIT(new SleepingWaitStrategy()),
+        /**
+         * YieldingWaitStrategy是可以被用在低延迟系统中的两个策略之一，这种策略在减低系统延迟的同时也会增加CPU运算量
+         * YieldingWaitStrategy策略会循环等待sequence增加到合适的值，循环中调用Thread.yield()允许其他准备好的线程执行
+         * 如果需要高性能而且事件消费者线程比逻辑内核少的时候，推荐使用YieldingWaitStrategy策略，例如在开启超线程时
+         */
+        YIELDING_WAIT(new YieldingWaitStrategy()),
+        /**
+         * BusySpinWaitStrategy是性能最高的等待策略，同时也是对部署环境要求最高的策略
+         * 这个性能最好用在事件处理线程比物理内核数目还要小的时候，例如在禁用超线程技术时
+         */
+        BUSY_SPIN_WAIT(new BusySpinWaitStrategy());
+        private final WaitStrategy waitStrategy;
+    }
+}

4dkankan-utils-disruptormq/src/main/resources/META-INF/spring.factories

@@ -0,0 +1 @@
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=com.fdkankan.disruptormq.DisruptorProperties