dubbo async template

README.md

@@ -52,4 +52,5 @@ router <--> das
 das shardingjdbc
 table struct, sharding roles
 logic biz
- 
+ 
+请求响应消息队列异步处理，减少线程 cpu 占用, dubbo async

im-client/pom.xml

@@ -25,6 +25,11 @@
             <groupId>io.netty</groupId>
             <artifactId>netty-all</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.beust</groupId>
+            <artifactId>jcommander</artifactId>
+            <version>1.82</version>
+        </dependency>
     </dependencies>
 
 </project>

im-client/src/main/java/net/sopod/soim/client/ClientMain.java

@@ -4,6 +4,9 @@ import net.sopod.soim.client.net.ImNetClient;
 
 /**
  * Main
+ * jcommander:
+ *   https://www.codenong.com/b-jcommander-parsing-command-line-parameters/
+ *   https://www.mianshigee.com/project/jcommander
  *
  * @author tmy
  * @date 2022-03-27 22:32

im-core/src/main/java/net/sopod/soim/core/registry/ProtoMessageHandlerRegistry.java

@@ -47,6 +47,7 @@ public class ProtoMessageHandlerRegistry {
             try {
                 Class<?> type = genericTypes.size() == 0 ? Object.class : Class.forName(genericTypes.get(0));
                 MessageHandler<?> existHandler = TYPE_HANDLER_MAP.putIfAbsent(type, handler);
+                logger.debug("registe msg type {} for handler {}", type, handler);
                 if (existHandler != null) {
                     // 消息类型有重复的 handler!
                     throw new IllegalStateException("msg type " + type + " handler duplicate; " +

im-entry/pom.xml

@@ -21,6 +21,16 @@
             <artifactId>im-core</artifactId>
             <version>${soim.version}</version>
         </dependency>
+        <dependency>
+            <groupId>net.sopod</groupId>
+            <artifactId>im-segment-id-api</artifactId>
+            <version>${soim.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>net.sopod</groupId>
+            <artifactId>im-logic-user-api</artifactId>
+            <version>${soim.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.projectlombok</groupId>
             <artifactId>lombok</artifactId>
@@ -62,6 +72,10 @@
             <groupId>org.apache.dubbo</groupId>
             <artifactId>dubbo-registry-nacos</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.lmax</groupId>
+            <artifactId>disruptor</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

im-entry/src/main/java/net/sopod/soim/entry/config/EntryServerConfig.java

@@ -0,0 +1,31 @@
+package net.sopod.soim.entry.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.stereotype.Component;
+
+/**
+ * EntryServerConfig
+ *
+ * @author tmy
+ * @date 2022-04-11 16:08
+ */
+@Component
+@ConfigurationProperties(prefix = "entry")
+@EnableConfigurationProperties
+@Data
+public class EntryServerConfig {
+
+    /** 消息消费者线程数 */
+    private Integer workerSize;
+
+    public Integer getWorkerSize() {
+        if (workerSize == null || workerSize < 1) {
+            workerSize = Runtime.getRuntime().availableProcessors();
+            workerSize = Math.max(workerSize, 4);
+        }
+        return workerSize;
+    }
+
+}

im-entry/src/main/java/net/sopod/soim/entry/handler/HelloHandler.java

@@ -4,8 +4,18 @@ import com.google.protobuf.MessageLite;
 import net.sopod.soim.core.handler.NetUserMessageHandler;
 import net.sopod.soim.core.session.NetUser;
 import net.sopod.soim.data.msg.hello.HelloPB;
+import net.sopod.soim.logic.api.segmentid.core.SegmentIdGenerator;
+import net.sopod.soim.logic.user.service.UserService;
+import org.apache.dubbo.config.annotation.DubboReference;
+import org.apache.dubbo.config.annotation.Method;
+import org.apache.dubbo.rpc.RpcContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
+import java.util.concurrent.CompletableFuture;
+
 /**
  * HelloHandler
  *
@@ -15,11 +25,27 @@ import org.springframework.stereotype.Service;
 @Service
 public class HelloHandler extends NetUserMessageHandler<HelloPB.Hello> {
 
+    private static final Logger logger = LoggerFactory.getLogger(HelloHandler.class);
+
+    @DubboReference(methods = {@Method(name = "sayHello", async = true)})
+    private UserService userService;
+
+    @Autowired
+    private SegmentIdGenerator segmentIdGenerator;
+
     @Override
     public MessageLite handle(NetUser netUser, HelloPB.Hello msg) {
-        System.out.println("get hello message");
+        logger.info("get hello message: {}", segmentIdGenerator.nextId("im-entry-hello"));
-        System.out.println(msg);
+        logger.info("msg={}, {}", msg.getId(), msg.getStr());
-        System.out.println(msg.getStr());
+        CompletableFuture<String> hi = userService.sayHi("黄绿");
+        hi.whenComplete((res, err) -> {
+           logger.info("async hi, {}", res);
+        });
+        String hello = userService.sayHello("lastJet");
+        logger.info("hello:{}", hello);
+        RpcContext.getServiceContext().getCompletableFuture().whenComplete((res, err) -> {
+            logger.info("async sayHello, {}", res);
+        });
         return null;
     }
 

im-entry/src/main/java/net/sopod/soim/entry/worker/DisruptorTest.java

@@ -0,0 +1,78 @@
+package net.sopod.soim.entry.worker;
+
+import com.lmax.disruptor.*;
+import com.lmax.disruptor.dsl.Disruptor;
+import com.lmax.disruptor.dsl.ProducerType;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import net.sopod.soim.common.util.ImClock;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Worker
+ *
+ * @author tmy
+ * @date 2022-04-11 17:05
+ */
+public class DisruptorTest {
+    public static class LongEvent {
+        int id;
+        private Long value;
+        public Long getValue() {
+            return value;
+        }
+        public void setValue(Long value) {
+        this.value = value;
+    }
+    }
+    public static class LongEventFactory implements EventFactory<LongEvent> {
+        private static final AtomicInteger counter = new AtomicInteger(1);
+        @Override
+        public LongEvent newInstance() {
+            int count = counter.getAndIncrement();
+            System.out.println("instance:"+ count);
+            LongEvent event = new LongEvent();
+            event.id = count;
+            return event;
+        }
+    }
+    public static class LongEventHandler implements EventHandler<LongEvent> {
+        @Override
+        public void onEvent(LongEvent event, long sequence, boolean endOfBatch) throws Exception {
+            System.out.println("消费者:"+event.id + "," + sequence + "," + event.getValue() + "," + Thread.currentThread().getName());
+            Thread.sleep(20);
+            // event值置空，回收内存
+            event.setValue(null);
+        }
+    }
+    public static void main(String[] args) throws InterruptedException, InsufficientCapacityException {
+        LongEventFactory eventFactory = new LongEventFactory();
+        int ringBufferSize = 128;//64 * 1024;
+        Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(
+                eventFactory,
+                ringBufferSize,
+                new DefaultThreadFactory("disruptor"),
+                ProducerType.SINGLE,
+                new BlockingWaitStrategy()
+        );
+        disruptor.handleEventsWith(new LongEventHandler());
+        disruptor.start();
+
+        // 7.创建RingBuffer容器
+        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
+        for (long i = 0; i < 1000; i++) {
+            long start = ImClock.millis();
+            // 没有空闲位置时会阻塞
+            long sequence = ringBuffer.tryNext();
+            //long sequence = ringBuffer.next();
+            LongEvent longEvent = ringBuffer.get(sequence);
+            longEvent.setValue(i);
+            Thread.sleep(10);
+            ringBuffer.publish(sequence);
+            System.out.println("生产者:" + i + "," + (ImClock.millis() - start) + "ms");
+        }
+
+        Thread.sleep(5000);
+        disruptor.shutdown();
+    }
+}

im-entry/src/main/resources/application.yml

@@ -5,4 +5,4 @@ dubbo:
   application:
     name: ${spring.application.name}
   registry:
-    address: nacos://124.222.131.236:3848
+    address: nacos://124.222.131.236:3848

im-logic-api/im-logic-user-api/src/main/java/net/sopod/soim/logic/user/service/UserService.java

@@ -1,5 +1,8 @@
 package net.sopod.soim.logic.user.service;
 
+
+import java.util.concurrent.CompletableFuture;
+
 /**
  * UserService
  *
@@ -9,8 +12,10 @@ package net.sopod.soim.logic.user.service;
 public interface UserService {
 
     /**
-     * 
+     * 异步接口测试
      */
-    void a();
+    CompletableFuture<String> sayHi(String name);
+
+    String sayHello(String name);
 
 }

im-logic/im-logic-user/src/main/java/net/sopod/soim/logic/user/LogicUserApplication.java

@@ -1,15 +1,21 @@
 package net.sopod.soim.logic.user;
 
+import org.apache.dubbo.config.spring.context.annotation.EnableDubbo;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
 /**
  * UserMain
  *
  * @author tmy
  * @date 2022-03-26 01:41
  */
+@EnableDubbo(scanBasePackages = {"net.sopod.soim.logic.user.service"})
+@SpringBootApplication
 public class LogicUserApplication {
 
     public static void main(String[] args) {
-
+        SpringApplication.run(LogicUserApplication.class, args);
     }
 
 }

im-logic/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/UserServiceImpl.java

@@ -1,10 +1,26 @@
 package net.sopod.soim.logic.user.service;
 
+import org.apache.dubbo.config.annotation.DubboService;
+
+import java.util.concurrent.CompletableFuture;
+
 /**
  * UserServiceImpl
  *
  * @author tmy
  * @date 2022-04-04 10:03
  */
-public class UserServiceImpl  {
+@DubboService
+public class UserServiceImpl implements UserService {
+
+    @Override
+    public CompletableFuture<String> sayHi(String name) {
+        return CompletableFuture.completedFuture("hi," + name + "!");
+    }
+
+    @Override
+    public String sayHello(String name) {
+        return "hello," + name + "!";
+    }
+
 }

im-logic/im-logic-user/src/main/resources/application.yml

@@ -0,0 +1,11 @@
+spring:
+  application:
+    name: im-logic-user
+
+dubbo:
+  application:
+    name: ${spring.application.name}
+  registry:
+    address: nacos://124.222.131.236:3848
+  protocol:
+    port: 3002

im-logic/im-segment-id/src/main/resources/application.yml

@@ -6,6 +6,15 @@ spring:
     url: jdbc:mysql://cd-cdb-mrz9fw80.sql.tencentcdb.com:61843/soim_db?serverTimezone=GMT%2B8
     username: root
     password: sopod@2347#
+    hikari:
+      minimum-idle: 1
+      maximum-pool-size: 8
+      connection-timeout: 2000
+      idle-timeout: 600000 # 10分钟空闲关闭
+      max-lifetime: 1200000 # 20分钟最大存活时间
+      validation-timeout: 2000
+      connection-init-sql: select 1
+      keepalive-time: 60000 # 连接存活时间，小于maxLifetime, 最小30秒, 空闲30秒后移除连接测试通过再添加回池
   redis:
     host: 124.222.131.236
     port: 3379