From 82fa56d66c9d894dffcc4ba93f1105d6b75c4056 Mon Sep 17 00:00:00 2001
From: redkale <redkale@qq.com>
Date: Thu, 20 Apr 2023 19:56:02 +0800
Subject: [PATCH] =?UTF-8?q?=E5=BA=9F=E5=BC=83ThreadHashExecutor?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../java/META-INF/application-template.xml    |  3 +-
 .../java/org/redkale/boot/Application.java    |  9 +---
 src/main/java/org/redkale/net/Context.java    | 26 +--------
 src/main/java/org/redkale/net/Response.java   |  7 ---
 src/main/java/org/redkale/net/Server.java     |  2 -
 src/main/java/org/redkale/net/WorkThread.java | 54 +------------------
 .../org/redkale/net/http/HttpResponse.java    |  5 --
 .../net/http/WebSocketReadHandler.java        |  8 +--
 .../org/redkale/net/sncp/SncpResponse.java    |  7 +--
 .../org/redkale/service/AbstractService.java  | 37 ++-----------
 .../org/redkale/source/DataJdbcSource.java    |  2 -
 .../org/redkale/util/ThreadHashExecutor.java  |  2 +
 12 files changed, 16 insertions(+), 146 deletions(-)

diff --git a/src/main/java/META-INF/application-template.xml b/src/main/java/META-INF/application-template.xml
index 246f48c9d..09c81072e 100644
--- a/src/main/java/META-INF/application-template.xml
+++ b/src/main/java/META-INF/application-template.xml
@@ -38,9 +38,8 @@
        【节点全局唯一】 @since 2.3.0
         全局Serivce执行的线程池， Application.workExecutor, 没配置该节点将自动创建一个。
         threads： 线程数，为0表示不启用workExecutor，只用IO线程。默认: CPU核数, 核数=1的情况下默认值为2
-        hash:     是否使用ThreadHashExecutor作为线程池，默认值为：false
     -->
-    <executor threads="4" hash="false"/>
+    <executor threads="4"/>
         
     <!--
        【节点全局唯一】
diff --git a/src/main/java/org/redkale/boot/Application.java b/src/main/java/org/redkale/boot/Application.java
index 2e861d51b..38114a88b 100644
--- a/src/main/java/org/redkale/boot/Application.java
+++ b/src/main/java/org/redkale/boot/Application.java
@@ -560,14 +560,9 @@ public final class Application {
             executorConf = DefaultAnyValue.create();
         }
         final int workThreads = executorConf.getIntValue("threads", Utility.cpus() * 4);
-        boolean workHash = executorConf.getBoolValue("hash", false);
         if (workThreads > 0) {
-            if (workHash) {
-                workExecutor0 = WorkThread.createHashExecutor(workThreads, "Redkale-HashWorkThread-%s");
-            } else {
-                //指定threads则不使用虚拟线程池
-                workExecutor0 = executorConf.getValue("threads") != null ? WorkThread.createExecutor(workThreads, "Redkale-WorkThread-%s") : WorkThread.createWorkExecutor(workThreads, "Redkale-WorkThread-%s");
-            }
+            //指定threads则不使用虚拟线程池
+            workExecutor0 = executorConf.getValue("threads") != null ? WorkThread.createExecutor(workThreads, "Redkale-WorkThread-%s") : WorkThread.createWorkExecutor(workThreads, "Redkale-WorkThread-%s");
         }
         this.workExecutor = workExecutor0;
         this.resourceFactory.register(RESNAME_APP_EXECUTOR, Executor.class, this.workExecutor);
diff --git a/src/main/java/org/redkale/net/Context.java b/src/main/java/org/redkale/net/Context.java
index 4278b4392..0ce8d59e4 100644
--- a/src/main/java/org/redkale/net/Context.java
+++ b/src/main/java/org/redkale/net/Context.java
@@ -30,8 +30,6 @@ public class Context {
     //Server的线程池
     protected final ExecutorService workExecutor;
 
-    protected final ThreadHashExecutor workHashExecutor;
-
     //SSL
     protected final SSLBuilder sslBuilder;
 
@@ -103,11 +101,6 @@ public class Context {
         this.writeTimeoutSeconds = writeTimeoutSeconds;
         this.jsonFactory = JsonFactory.root();
         this.bsonFactory = BsonFactory.root();
-        if (workExecutor instanceof ThreadHashExecutor) {
-            this.workHashExecutor = (ThreadHashExecutor) workExecutor;
-        } else {
-            this.workHashExecutor = null;
-        }
     }
 
     protected void executeDispatch(Request request, Response response) {
@@ -115,24 +108,7 @@ public class Context {
     }
 
     public void execute(Servlet servlet, Request request, Response response) {
-        if (workHashExecutor != null) {
-            response.updateNonBlocking(false);
-            workHashExecutor.execute(request.getHashid(), () -> {
-                try {
-                    long cha = System.currentTimeMillis() - request.getCreateTime();
-                    Traces.computeCurrTraceid(request.getTraceid());
-                    servlet.execute(request, response);
-                    if (cha > 1000 && response.context.logger.isLoggable(Level.WARNING)) {
-                        response.context.logger.log(Level.WARNING, "hash execute servlet delays=" + cha + "ms, request=" + request);
-                    } else if (cha > 100 && response.context.logger.isLoggable(Level.FINE)) {
-                        response.context.logger.log(Level.FINE, "hash execute servlet delay=" + cha + "ms, request=" + request);
-                    }
-                } catch (Throwable t) {
-                    response.context.logger.log(Level.WARNING, "Execute servlet occur exception. request = " + request, t);
-                    response.finishError(t);
-                }
-            });
-        } else if (workExecutor != null && response.inNonBlocking() && !servlet.isNonBlocking()) {
+        if (workExecutor != null && response.inNonBlocking() && !servlet.isNonBlocking()) {
             response.updateNonBlocking(false);
             workExecutor.execute(() -> {
                 try {
diff --git a/src/main/java/org/redkale/net/Response.java b/src/main/java/org/redkale/net/Response.java
index b7928d71b..f6731a80a 100644
--- a/src/main/java/org/redkale/net/Response.java
+++ b/src/main/java/org/redkale/net/Response.java
@@ -34,8 +34,6 @@ public abstract class Response<C extends Context, R extends Request<C>> {
 
     protected final ExecutorService workExecutor;
 
-    protected final ThreadHashExecutor workHashExecutor;
-
     protected final R request;
 
     protected final WorkThread thread;
@@ -124,7 +122,6 @@ public abstract class Response<C extends Context, R extends Request<C>> {
         this.thread = WorkThread.currWorkThread();
         this.writeBuffer = context != null ? ByteBuffer.allocateDirect(context.getBufferCapacity()) : null;
         this.workExecutor = context == null || context.workExecutor == null ? ForkJoinPool.commonPool() : context.workExecutor;
-        this.workHashExecutor = context == null ? null : context.workHashExecutor;
     }
 
     protected AsyncConnection removeChannel() {
@@ -165,10 +162,6 @@ public abstract class Response<C extends Context, R extends Request<C>> {
         return workExecutor;
     }
 
-    protected ThreadHashExecutor getWorkHashExecutor() {
-        return workHashExecutor;
-    }
-
     protected void updateNonBlocking(boolean nonBlocking) {
         this.inNonBlocking = nonBlocking;
     }
diff --git a/src/main/java/org/redkale/net/Server.java b/src/main/java/org/redkale/net/Server.java
index e4311ef12..63ee33de7 100644
--- a/src/main/java/org/redkale/net/Server.java
+++ b/src/main/java/org/redkale/net/Server.java
@@ -318,8 +318,6 @@ public abstract class Server<K extends Serializable, C extends Context, R extend
         int workThreads = 0;
         if (workExecutor instanceof ThreadPoolExecutor) {
             workThreads = ((ThreadPoolExecutor) workExecutor).getCorePoolSize();
-        } else if (workExecutor instanceof ThreadHashExecutor) {
-            workThreads = ((ThreadHashExecutor) workExecutor).getCorePoolSize();
         } else if (workExecutor != null) { //virtual thread pool
             workThreads = -1;
         }
diff --git a/src/main/java/org/redkale/net/WorkThread.java b/src/main/java/org/redkale/net/WorkThread.java
index fd166b058..b0b3426cb 100644
--- a/src/main/java/org/redkale/net/WorkThread.java
+++ b/src/main/java/org/redkale/net/WorkThread.java
@@ -9,7 +9,7 @@ import java.util.Collection;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.*;
 import java.util.function.Function;
-import org.redkale.util.*;
+import org.redkale.util.Utility;
 
 /**
  * 协议处理的自定义线程类
@@ -23,8 +23,6 @@ public class WorkThread extends Thread implements Executor {
 
     protected final ExecutorService workExecutor;
 
-    protected final ThreadHashExecutor hashExecutor;
-
     private final int index;   //WorkThread下标，从0开始
 
     private final int threads; //WorkThread个数
@@ -37,7 +35,6 @@ public class WorkThread extends Thread implements Executor {
         this.index = index;
         this.threads = threads;
         this.workExecutor = workExecutor;
-        this.hashExecutor = workExecutor instanceof ThreadHashExecutor ? (ThreadHashExecutor) workExecutor : null;
         this.setDaemon(true);
     }
 
@@ -46,19 +43,6 @@ public class WorkThread extends Thread implements Executor {
         return t instanceof WorkThread ? (WorkThread) t : null;
     }
 
-    public static ExecutorService createHashExecutor(final int threads, final String threadNameFormat) {
-        final AtomicReference<ExecutorService> ref = new AtomicReference<>();
-        final AtomicInteger counter = new AtomicInteger();
-        final ThreadGroup g = new ThreadGroup(String.format(threadNameFormat, "Group"));
-        return new ThreadHashExecutor(threads, (Runnable r) -> {
-            int i = counter.get();
-            int c = counter.incrementAndGet();
-            String threadName = String.format(threadNameFormat, formatIndex(threads, c));
-            Thread t = new WorkThread(g, threadName, i, threads, ref.get(), r);
-            return t;
-        });
-    }
-
     public static ExecutorService createWorkExecutor(final int threads, final String threadNameFormat) {
         final Function<String, ExecutorService> func = Utility.virtualExecutorFunction();
         return func == null ? createExecutor(threads, threadNameFormat) : func.apply(threadNameFormat);
@@ -137,30 +121,6 @@ public class WorkThread extends Thread implements Executor {
         }
     }
 
-    public void runWork(int hash, Runnable command) {
-        if (hashExecutor == null) {
-            if (workExecutor == null) {
-                command.run();
-            } else {
-                workExecutor.execute(command);
-            }
-        } else {
-            hashExecutor.execute(hash, command);
-        }
-    }
-
-    public void runWork(java.io.Serializable hash, Runnable command) {
-        if (hashExecutor == null) {
-            if (workExecutor == null) {
-                command.run();
-            } else {
-                workExecutor.execute(command);
-            }
-        } else {
-            hashExecutor.execute(hash, command);
-        }
-    }
-
     public void runAsync(Runnable command) {
         if (workExecutor == null) {
             ForkJoinPool.commonPool().execute(command);
@@ -169,18 +129,6 @@ public class WorkThread extends Thread implements Executor {
         }
     }
 
-    public void runAsync(int hash, Runnable command) {
-        if (hashExecutor == null) {
-            if (workExecutor == null) {
-                ForkJoinPool.commonPool().execute(command);
-            } else {
-                workExecutor.execute(command);
-            }
-        } else {
-            hashExecutor.execute(hash, command);
-        }
-    }
-
     public ExecutorService getWorkExecutor() {
         return workExecutor;
     }
diff --git a/src/main/java/org/redkale/net/http/HttpResponse.java b/src/main/java/org/redkale/net/http/HttpResponse.java
index 1ecd5ceb4..363d5c6a7 100644
--- a/src/main/java/org/redkale/net/http/HttpResponse.java
+++ b/src/main/java/org/redkale/net/http/HttpResponse.java
@@ -241,11 +241,6 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
         return super.getWorkExecutor();
     }
 
-    @Override
-    protected ThreadHashExecutor getWorkHashExecutor() {
-        return super.getWorkHashExecutor();
-    }
-
     @Override
     protected void updateNonBlocking(boolean nonBlocking) {
         super.updateNonBlocking(nonBlocking);
diff --git a/src/main/java/org/redkale/net/http/WebSocketReadHandler.java b/src/main/java/org/redkale/net/http/WebSocketReadHandler.java
index fb8fa05aa..c3d83ec38 100644
--- a/src/main/java/org/redkale/net/http/WebSocketReadHandler.java
+++ b/src/main/java/org/redkale/net/http/WebSocketReadHandler.java
@@ -316,7 +316,7 @@ public class WebSocketReadHandler implements CompletionHandler<Integer, ByteBuff
                 //消息处理
                 for (final WebSocketPacket packet : currPackets) {
                     if (packet.type == FrameType.TEXT) {
-                        ioReadThread.runWork(webSocket._userid, () -> {
+                        ioReadThread.runWork(() -> {
                             try {
                                 Convert convert = webSocket.getTextConvert();
                                 if (restMessageConsumer != null && convert != null) { //主要供RestWebSocket使用
@@ -329,7 +329,7 @@ public class WebSocketReadHandler implements CompletionHandler<Integer, ByteBuff
                             }
                         });
                     } else if (packet.type == FrameType.BINARY) {
-                        ioReadThread.runWork(webSocket._userid, () -> {
+                        ioReadThread.runWork(() -> {
                             try {
                                 Convert convert = webSocket.getBinaryConvert();
                                 if (restMessageConsumer != null && convert != null) { //主要供RestWebSocket使用
@@ -342,7 +342,7 @@ public class WebSocketReadHandler implements CompletionHandler<Integer, ByteBuff
                             }
                         });
                     } else if (packet.type == FrameType.PING) {
-                        ioReadThread.runWork(webSocket._userid, () -> {
+                        ioReadThread.runWork(() -> {
                             try {
                                 webSocket.onPing(packet.getPayload());
                             } catch (Exception e) {
@@ -350,7 +350,7 @@ public class WebSocketReadHandler implements CompletionHandler<Integer, ByteBuff
                             }
                         });
                     } else if (packet.type == FrameType.PONG) {
-                        ioReadThread.runWork(webSocket._userid, () -> {
+                        ioReadThread.runWork(() -> {
                             try {
                                 //if (debug) logger.log(Level.FINEST, "WebSocket onMessage by PONG FrameType : " + packet);
                                 webSocket.onPong(packet.getPayload());
diff --git a/src/main/java/org/redkale/net/sncp/SncpResponse.java b/src/main/java/org/redkale/net/sncp/SncpResponse.java
index 20d1885eb..dcf2349ae 100644
--- a/src/main/java/org/redkale/net/sncp/SncpResponse.java
+++ b/src/main/java/org/redkale/net/sncp/SncpResponse.java
@@ -11,7 +11,7 @@ import java.util.concurrent.*;
 import org.redkale.convert.bson.BsonWriter;
 import org.redkale.net.Response;
 import static org.redkale.net.sncp.SncpHeader.HEADER_SIZE;
-import org.redkale.util.*;
+import org.redkale.util.ByteArray;
 
 /**
  *
@@ -111,11 +111,6 @@ public class SncpResponse extends Response<SncpContext, SncpRequest> {
         return super.getWorkExecutor();
     }
 
-    @Override
-    protected ThreadHashExecutor getWorkHashExecutor() {
-        return super.getWorkHashExecutor();
-    }
-
     @Override
     protected void updateNonBlocking(boolean nonBlocking) {
         super.updateNonBlocking(nonBlocking);
diff --git a/src/main/java/org/redkale/service/AbstractService.java b/src/main/java/org/redkale/service/AbstractService.java
index 1bafb5163..f692cb47f 100644
--- a/src/main/java/org/redkale/service/AbstractService.java
+++ b/src/main/java/org/redkale/service/AbstractService.java
@@ -6,10 +6,10 @@
 package org.redkale.service;
 
 import java.util.concurrent.*;
-import org.redkale.annotation.*;
-import org.redkale.boot.*;
-import org.redkale.net.*;
-import org.redkale.util.*;
+import org.redkale.annotation.Resource;
+import org.redkale.boot.Application;
+import org.redkale.net.WorkThread;
+import org.redkale.util.ResourceFactory;
 
 /**
  *
@@ -55,35 +55,6 @@ public abstract class AbstractService implements Service {
         }
     }
 
-    /**
-     * 异步执行任务
-     *
-     * @param hash    hash值
-     * @param command 任务
-     */
-    protected void runAsync(int hash, Runnable command) {
-        ExecutorService executor = this.workExecutor;
-        if (executor != null) {
-            if (executor instanceof ThreadHashExecutor) {
-                ((ThreadHashExecutor) executor).execute(hash, command);
-            } else {
-                Thread thread = Thread.currentThread();
-                if (thread instanceof WorkThread) {
-                    ((WorkThread) thread).runAsync(hash, command);
-                } else {
-                    executor.execute(command);
-                }
-            }
-        } else {
-            Thread thread = Thread.currentThread();
-            if (thread instanceof WorkThread) {
-                ((WorkThread) thread).runAsync(hash, command);
-            } else {
-                ForkJoinPool.commonPool().execute(command);
-            }
-        }
-    }
-
     /**
      * 获取线程池
      *
diff --git a/src/main/java/org/redkale/source/DataJdbcSource.java b/src/main/java/org/redkale/source/DataJdbcSource.java
index 76cd8cac9..fe71a2166 100644
--- a/src/main/java/org/redkale/source/DataJdbcSource.java
+++ b/src/main/java/org/redkale/source/DataJdbcSource.java
@@ -2710,8 +2710,6 @@ public class DataJdbcSource extends AbstractDataSqlSource {
             int defMaxConns = Utility.cpus() * 4;
             if (workExecutor instanceof ThreadPoolExecutor) {
                 defMaxConns = ((ThreadPoolExecutor) workExecutor).getCorePoolSize();
-            } else if (workExecutor instanceof ThreadHashExecutor) {
-                defMaxConns = ((ThreadHashExecutor) workExecutor).getCorePoolSize();
             } else if (workExecutor != null) { //maybe virtual thread pool
                 defMaxConns = Math.min(1024, Utility.cpus() * 100);
             }
diff --git a/src/main/java/org/redkale/util/ThreadHashExecutor.java b/src/main/java/org/redkale/util/ThreadHashExecutor.java
index 9152de97b..67b1a4e69 100644
--- a/src/main/java/org/redkale/util/ThreadHashExecutor.java
+++ b/src/main/java/org/redkale/util/ThreadHashExecutor.java
@@ -16,9 +16,11 @@ import java.util.concurrent.atomic.AtomicInteger;
  * 详情见: https://redkale.org
  *
  * @author zhangjx
+ * @deprecated
  *
  * @since 2.1.0
  */
+@Deprecated(since = "2.8.0")
 public class ThreadHashExecutor extends AbstractExecutorService {
 
     private final LinkedBlockingQueue<Runnable>[] queues;