Disruptor 试水

pom.xml

@@ -27,6 +27,7 @@
         <springboot.version>3.2.5</springboot.version>
         <rxjava.version>3.1.8</rxjava.version>
         <springdoc.version>2.5.0</springdoc.version>
+        <disruptor.version>4.0.0</disruptor.version>
 
         <gb28181.docking.version>0.1.0</gb28181.docking.version>
     </properties>
@@ -40,6 +41,12 @@
                 <type>pom</type>
             </dependency>
 
+            <dependency>
+                <groupId>com.lmax</groupId>
+                <artifactId>disruptor</artifactId>
+                <version>${disruptor.version}</version>
+            </dependency>
+
             <dependency>
                 <groupId>io.reactivex.rxjava3</groupId>
                 <artifactId>rxjava</artifactId>

starter/pom.xml

@@ -30,6 +30,11 @@
             <artifactId>springdoc-openapi-starter-webmvc-api</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>com.lmax</groupId>
+            <artifactId>disruptor</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>io.reactivex.rxjava3</groupId>
             <artifactId>rxjava</artifactId>

starter/src/test/java/cn/skcks/study/disruptor/DisruptorTest.java

@@ -0,0 +1,131 @@
+package cn.skcks.study.disruptor;
+
+
+import com.lmax.disruptor.BusySpinWaitStrategy;
+import com.lmax.disruptor.EventFactory;
+import com.lmax.disruptor.EventHandler;
+import com.lmax.disruptor.RingBuffer;
+import com.lmax.disruptor.dsl.Disruptor;
+import com.lmax.disruptor.dsl.ProducerType;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+import java.util.concurrent.ThreadPoolExecutor;
+
+@Slf4j
+public class DisruptorTest {
+    /**
+     * cpu 核心数
+     */
+    public static final int CPU_NUM = Runtime.getRuntime().availableProcessors();
+    /**
+     * 最大线程数
+     */
+    public static final int MAX_POOL_SIZE = CPU_NUM * 2;
+    /**
+     * 允许线程空闲时间（单位：默认为秒）
+     */
+    private static final int KEEP_ALIVE_TIME = 20;
+    /**
+     * 队列长度
+     */
+    public static final int TASK_NUM = 5;
+    /**
+     * 线程名称(前缀)
+     */
+    public static final String THREAD_NAME_PREFIX = "custom-disruptor-executor";
+
+    public static class LongEvent
+    {
+        private long value;
+
+        public void set(long value)
+        {
+            this.value = value;
+        }
+
+        @Override
+        public String toString()
+        {
+            return "LongEvent{" + "value=" + value + '}';
+        }
+    }
+
+    public static class LongEventFactory implements EventFactory<LongEvent>
+    {
+        @Override
+        public LongEvent newInstance()
+        {
+            return new LongEvent();
+        }
+    }
+
+    public static class LongEventHandler implements EventHandler<LongEvent>
+    {
+        @Override
+        public void onEvent(LongEvent event, long sequence, boolean endOfBatch)
+        {
+            log.info("Event: {}, sequence: {}, endOfBatch: {}", event, sequence, endOfBatch);
+        }
+    }
+
+    public static void main(String[] args) throws Exception
+    {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        executor.setCorePoolSize(CPU_NUM);
+        executor.setMaxPoolSize(MAX_POOL_SIZE);
+        executor.setQueueCapacity(TASK_NUM);
+        executor.setKeepAliveSeconds(KEEP_ALIVE_TIME);
+        executor.setThreadNamePrefix(THREAD_NAME_PREFIX);
+        // 线程池对拒绝任务的处理策略
+        // CallerRunsPolicy：由调用线程（提交任务的线程）处理该任务
+        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
+        // 初始化
+        executor.initialize();
+
+        // ExecutorService executor = Executors.newCachedThreadPool();
+        int bufferSize = 64;
+
+        // LongEventFactory factory = new LongEventFactory();
+        // Disruptor<LongEvent> disruptor =
+        //         new Disruptor<>(factory, bufferSize, DaemonThreadFactory.INSTANCE, ProducerType.SINGLE, new BusySpinWaitStrategy());
+
+        // Disruptor<LongEvent> disruptor =
+        //         new Disruptor<>(LongEvent::new, bufferSize, DaemonThreadFactory.INSTANCE, ProducerType.SINGLE, new BusySpinWaitStrategy());
+
+        Disruptor<LongEvent> disruptor =
+                new Disruptor<>(LongEvent::new, bufferSize, executor, ProducerType.SINGLE, new BusySpinWaitStrategy());
+
+        // disruptor.handleEventsWith((event, sequence, endOfBatch) ->
+        //         System.out.println("Event: " + event));
+
+        disruptor.handleEventsWith(new LongEventHandler(),new LongEventHandler(),new LongEventHandler());
+        disruptor.start();
+
+        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
+        new Thread(()->{
+            for (long l = 0; l<256; l++)
+            {
+                // bb.putLong(0, l);
+                // ringBuffer.publishEvent((event, sequence, buffer) -> event.set(buffer.getLong(0)), bb);
+
+                long finalL = l;
+                ringBuffer.publishEvent((event, sequence, buffer) -> event.set(finalL));
+            }
+        }).start();
+
+        // ByteBuffer bb = ByteBuffer.allocate(8);
+        for (long l = 0; l<128; l++)
+        {
+            // bb.putLong(0, l);
+            // ringBuffer.publishEvent((event, sequence, buffer) -> event.set(buffer.getLong(0)), bb);
+
+            long finalL = l;
+            ringBuffer.publishEvent((event, sequence, buffer) -> event.set(finalL));
+            Thread.sleep(1);
+        }
+
+        disruptor.shutdown(); //关闭 disruptor  阻塞直至所有事件都得到处理
+        executor.shutdown();
+    }
+}

starter/src/test/java/cn/skcks/study/rxjava/RxJavaStudyStarterTest.java

@@ -9,6 +9,7 @@ import io.reactivex.rxjava3.core.Observable;
 import io.reactivex.rxjava3.disposables.Disposable;
 import io.reactivex.rxjava3.functions.Consumer;
 import io.reactivex.rxjava3.schedulers.Schedulers;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.junit.jupiter.api.Test;
 import org.springframework.boot.test.context.SpringBootTest;
@@ -16,6 +17,7 @@ import org.springframework.boot.test.context.SpringBootTest;
 import javax.sip.ListeningPoint;
 import javax.sip.message.Request;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 
 @Slf4j
 @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE)
@@ -70,16 +72,9 @@ public class RxJavaStudyStarterTest {
         take.subscribe(System.out::println);
     }
 
+    @SneakyThrows
     @Test
     void gb28181SipTest(){
-        Consumer<Request> serverConsumer = request -> {
-            SIPRequest sipRequest = (SIPRequest) request;
-            String method = request.getMethod();
-            log.info("收到SIP请求: {}", method);
-            Thread.sleep(RandomUtil.randomLong(10,500));
-            log.info("处理SIP请求: {}", sipRequest.getCallId().getCallId());
-        };
-
         String localIp = "127.0.0.1";
         int localPort = 5060;
         String localId = "44050100000000000001";
@@ -97,24 +92,43 @@ public class RxJavaStudyStarterTest {
                 .targetId(serverId)
                 .targetIp(serverIp)
                 .targetPort(serverPort)
+
                 .transport(ListeningPoint.UDP)
                 .build();
 
-        Flowable<Request> requestObservable = Flowable.intervalRange(1,100, 0,10 ,TimeUnit.MILLISECONDS)
+        AtomicLong produce = new AtomicLong(0);
-                .onBackpressureBuffer()
+        AtomicLong consume = new AtomicLong(0);
+
+        Flowable<Request> requestObservable = Flowable.intervalRange(0,10,0,100,TimeUnit.MILLISECONDS)
+                .onBackpressureDrop()
                 .map(i -> {
                     String callId = IdUtil.fastSimpleUUID();
+                    produce.getAndAdd(1);
                     return requestBuilder.createNoAuthorizationRequest(callId, 3600);
                 })
+                .cache()
+                .share()
+                .distinct(request -> ((SIPRequest) request).getCallId().getCallId())
                 .observeOn(Schedulers.computation())
                 .subscribeOn(Schedulers.io());
 
-        requestObservable.subscribe(serverConsumer);
+        Flowable<Request> requestFlowable = requestObservable.publish().refCount();
-        requestObservable.subscribe(serverConsumer);
-        requestObservable.subscribe(serverConsumer);
-        requestObservable.subscribe(serverConsumer);
-        requestObservable.subscribe(serverConsumer);
 
-        requestObservable.blockingSubscribe();
+        Consumer<Request> serverConsumer = request -> {
+            SIPRequest sipRequest = (SIPRequest) request;
+            String method = request.getMethod();
+            log.info("收到SIP请求: {}", method);
+            Thread.sleep(RandomUtil.randomLong(10,100));
+            log.info("处理SIP请求: {}", sipRequest.getCallId().getCallId());
+            consume.getAndAdd(1);
+        };
+
+        requestFlowable.subscribe(serverConsumer);
+        requestFlowable.subscribe(serverConsumer);
+        // requestObservable.subscribe(serverConsumer);
+        // requestObservable.blockingSubscribe();
+
+        Thread.sleep(TimeUnit.SECONDS.toMillis(30));
+        log.info("生产数量: {}, 消费数量: {}", produce.get(), consume.get());
     }
 }