--使用线程池模式提交request--

AwakenCN · Aug 26, 2023 · 4167ef4 · 4167ef4
1 parent 6b61227
commit 4167ef4
Show file tree

Hide file tree

Showing 5 changed files with 79 additions and 28 deletions.
diff --git a/src/main/java/com/github/unclecatmyself/bootstrap/handler/DefaultAbstractHandler.java b/src/main/java/com/github/unclecatmyself/bootstrap/handler/DefaultAbstractHandler.java
@@ -2,16 +2,15 @@
 
 import com.alibaba.fastjson.JSON;
 import com.github.unclecatmyself.core.bean.InChatMessage;
-import com.github.unclecatmyself.core.utils.DateUtil;
+import com.github.unclecatmyself.core.thread.MessageTask;
 import com.github.unclecatmyself.support.HandlerService;
 import com.github.unclecatmyself.core.bean.vo.SendServerVO;
-import com.github.unclecatmyself.core.constant.Constants;
 import com.github.unclecatmyself.core.constant.HttpConstant;
 import com.github.unclecatmyself.core.constant.LogConstant;
 import com.github.unclecatmyself.core.constant.UndefinedInChatConstant;
 import com.github.unclecatmyself.core.exception.HandlerNotFoundException;
 import com.github.unclecatmyself.core.utils.HttpUtil;
-import com.github.unclecatmyself.support.MessageFactory;
+import com.github.unclecatmyself.support.InChatInitializer;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
@@ -23,7 +22,6 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.UnsupportedEncodingException;
-import java.lang.reflect.Method;
 
 /**
  * Create by UncleCatMySelf in 2018/12/06
@@ -123,22 +121,8 @@ protected void readTextMessage(ChannelHandlerContext ctx, TextWebSocketFrame msg
         }
         logger.debug("readTextMessage success, body: {}", msg.text());
         InChatMessage message = JSON.parseObject(msg.text(), InChatMessage.class);
-        int now = DateUtil.getSecond();
-        message.setTime(now);
-        Method method = MessageFactory.getInstance().getMethod(message.getType());
-        if (null == method) {
-            logger.warn("message#{} unregistered!", message.getType());
-            return;
-        }
-        if(!Constants.LOGIN.equals(message.getType())) {
-            handlerService.verify(channel, message);
-        }
-        //TODO 后续改用线程池，异步调用
-        try {
-            method.invoke(handlerService, channel, message);
-        } catch (Exception e) {
-            logger.error("readTextMessage invoke method error, {}", e.getMessage(), e);
-        }
+        MessageTask task = new MessageTask(handlerService, message, channel);
+        InChatInitializer.optionExecutor.submit(task);
     }
 
     @Override

diff --git a/src/main/java/com/github/unclecatmyself/bootstrap/server/NettyBootstrapServer.java b/src/main/java/com/github/unclecatmyself/bootstrap/server/NettyBootstrapServer.java
@@ -3,6 +3,7 @@
 import com.github.unclecatmyself.core.bean.InitNetty;
 import com.github.unclecatmyself.core.config.AutoConfig;
 import com.github.unclecatmyself.core.config.RedisConfig;
+import com.github.unclecatmyself.core.thread.DefaultThreadFactory;
 import com.github.unclecatmyself.core.utils.PlatformUtil;
 import com.github.unclecatmyself.core.utils.RemotingUtil;
 import com.github.unclecatmyself.core.utils.UniqueIpUtils;
@@ -161,13 +162,7 @@ public void calculateStrategy(int ioModel) {
     }
 
     public static ThreadFactory buildThreadFactory(String threadName) {
-        return new ThreadFactory() {
-            private final AtomicInteger index = new AtomicInteger(0);
-
-            public Thread newThread(Runnable r) {
-                return new Thread(r, threadName + index.incrementAndGet());
-            }
-        };
+        return new DefaultThreadFactory(threadName);
     }
 
 }
diff --git a/src/main/java/com/github/unclecatmyself/core/constant/SessionKey.java b/src/main/java/com/github/unclecatmyself/core/constant/SessionKey.java
@@ -38,7 +38,7 @@ public class SessionKey {
     /**
      * 通道创建时间
      */
-    public static final AttributeKey<Long> CHANNEL_CREATE_TIME = AttributeKey.valueOf("channel_create_time");
+    public static final AttributeKey<Integer> CHANNEL_CREATE_TIME = AttributeKey.valueOf("channel_create_time");
     /**
      * 是否http协议
      **/

diff --git a/src/main/java/com/github/unclecatmyself/core/thread/MessageTask.java b/src/main/java/com/github/unclecatmyself/core/thread/MessageTask.java
@@ -0,0 +1,67 @@
+package com.github.unclecatmyself.core.thread;
+
+import com.github.unclecatmyself.core.bean.InChatMessage;
+import com.github.unclecatmyself.core.constant.Constants;
+import com.github.unclecatmyself.core.constant.SessionKey;
+import com.github.unclecatmyself.core.utils.DateUtil;
+import com.github.unclecatmyself.support.HandlerService;
+import com.github.unclecatmyself.support.MessageFactory;
+import io.netty.channel.Channel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.Method;
+
+/**
+ * @author noseparte
+ * @implSpec 消息线程
+ * @since 2023/8/26 - 11:52
+ * @version 1.0
+ */
+public class MessageTask implements Runnable {
+
+    private final static Logger logger = LoggerFactory.getLogger(MessageTask.class);
+
+    private HandlerService handlerService;
+    private final InChatMessage message;
+    private final Channel channel;
+
+    public MessageTask(InChatMessage message, Channel channel) {
+        this.message = message;
+        this.channel = channel;
+    }
+
+    public MessageTask(HandlerService handlerService, InChatMessage message, Channel channel) {
+        this.handlerService = handlerService;
+        this.message = message;
+        this.channel = channel;
+    }
+
+    @Override
+    public void run() {
+        doTask();
+    }
+
+
+    /**
+     * 改用线程池，异步调用
+     */
+    private void doTask() {
+        int now = DateUtil.getSecond();
+        message.setTime(now);
+        channel.attr(SessionKey.CHANNEL_CREATE_TIME).set(now);
+        Method method = MessageFactory.getInstance().getMethod(message.getType());
+        if (null == method) {
+            logger.warn("message#{} unregistered!", message.getType());
+            return;
+        }
+        if(!Constants.LOGIN.equals(message.getType())) {
+            handlerService.verify(channel, message);
+        }
+        try {
+            method.invoke(handlerService, channel, message);
+        } catch (Exception e) {
+            logger.error("readTextMessage invoke method error, {}", e.getMessage(), e);
+        }
+    }
+}
diff --git a/src/main/java/com/github/unclecatmyself/support/InChatInitializer.java b/src/main/java/com/github/unclecatmyself/support/InChatInitializer.java
@@ -1,21 +1,26 @@
 package com.github.unclecatmyself.support;
 
 import com.github.unclecatmyself.core.bean.InitNetty;
+import com.github.unclecatmyself.core.thread.StandardThreadExecutor;
 
 /**
  * Created by MySelf on 2019/8/26.
  */
 public class InChatInitializer extends InitNetty {
 
+    public static final StandardThreadExecutor optionExecutor = new StandardThreadExecutor();//操作池
+
     @Override
     public int getWebPort() {
         return 8090;
     }
+
     //分布式
     @Override
     public Boolean getDistributed() {
         return false;
     }
+
     //加密
     @Override
     public boolean isSsl() {