[ISSUE#161]optimize the thread pool ThreadFactory implementation

Author: mxsm

src/main/java/io/openmessaging/storage/dledger/DLedgerRpcNettyService.java

@@ -41,8 +41,6 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.rocketmq.remoting.ChannelEventListener;
 import org.apache.rocketmq.remoting.netty.NettyClientConfig;
 import org.apache.rocketmq.remoting.netty.NettyRemotingClient;
@@ -69,32 +67,11 @@ public class DLedgerRpcNettyService extends DLedgerRpcService {
 
     private DLedgerServer dLedgerServer;
 
-    private ExecutorService futureExecutor = Executors.newFixedThreadPool(4, new ThreadFactory() {
-        private AtomicInteger threadIndex = new AtomicInteger(0);
+    private ExecutorService futureExecutor = Executors.newFixedThreadPool(4, new NamedThreadFactory("FutureExecutor"));
 
-        @Override
-        public Thread newThread(Runnable r) {
-            return new Thread(r, "FutureExecutor_" + this.threadIndex.incrementAndGet());
-        }
-    });
-
-    private ExecutorService voteInvokeExecutor = Executors.newCachedThreadPool(new ThreadFactory() {
-        private AtomicInteger threadIndex = new AtomicInteger(0);
-
-        @Override
-        public Thread newThread(Runnable r) {
-            return new Thread(r, "voteInvokeExecutor_" + this.threadIndex.incrementAndGet());
-        }
-    });
-
-    private ExecutorService heartBeatInvokeExecutor = Executors.newCachedThreadPool(new ThreadFactory() {
-        private AtomicInteger threadIndex = new AtomicInteger(0);
+    private ExecutorService voteInvokeExecutor = Executors.newCachedThreadPool(new NamedThreadFactory("voteInvokeExecutor"));
 
-        @Override
-        public Thread newThread(Runnable r) {
-            return new Thread(r, "heartBeatInvokeExecutor_" + this.threadIndex.incrementAndGet());
-        }
-    });
+    private ExecutorService heartBeatInvokeExecutor = Executors.newCachedThreadPool(new NamedThreadFactory("heartBeatInvokeExecutor"));
 
     public DLedgerRpcNettyService(DLedgerServer dLedgerServer) {
         this(dLedgerServer, null, null, null);

src/main/java/io/openmessaging/storage/dledger/DLedgerServer.java

@@ -101,12 +101,7 @@ public DLedgerServer(DLedgerConfig dLedgerConfig, NettyServerConfig nettyServerC
         dLedgerRpcService = new DLedgerRpcNettyService(this, nettyServerConfig, nettyClientConfig, channelEventListener);
         dLedgerEntryPusher = new DLedgerEntryPusher(dLedgerConfig, memberState, dLedgerStore, dLedgerRpcService);
         dLedgerLeaderElector = new DLedgerLeaderElector(dLedgerConfig, memberState, dLedgerRpcService);
-        executorService = Executors.newSingleThreadScheduledExecutor(r -> {
-            Thread t = new Thread(r);
-            t.setDaemon(true);
-            t.setName("DLedgerServer-ScheduledExecutor");
-            return t;
-        });
+        executorService = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory(null, "DLedgerServer-ScheduledExecutor", true));
         this.fsmCaller = Optional.empty();
     }
 

src/main/java/io/openmessaging/storage/dledger/NamedThreadFactory.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2017-2022 The DLedger Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.openmessaging.storage.dledger;
+
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class NamedThreadFactory implements ThreadFactory {
+
+    private AtomicInteger threadIndex;
+
+    private final String threadNamePrefix;
+
+    private boolean isDemoThread;
+
+    public NamedThreadFactory(final String threadNamePrefix, boolean isDemoThread) {
+        this(new AtomicInteger(0), threadNamePrefix, isDemoThread);
+    }
+
+    public NamedThreadFactory(AtomicInteger threadIndex, final String threadNamePrefix, boolean isDemoThread) {
+        this.threadIndex = threadIndex;
+        this.threadNamePrefix = threadNamePrefix;
+        this.isDemoThread = isDemoThread;
+    }
+
+    public NamedThreadFactory(final String threadNamePrefix) {
+        this(threadNamePrefix, false);
+    }
+
+    /**
+     * Constructs a new {@code Thread}.  Implementations may also initialize priority, name, daemon status, {@code
+     * ThreadGroup}, etc.
+     *
+     * @param r a runnable to be executed by new thread instance
+     * @return constructed thread, or {@code null} if the request to create a thread is rejected
+     */
+    @Override
+    public Thread newThread(Runnable r) {
+
+        StringBuilder threadName = new StringBuilder(threadNamePrefix);
+        if (null != threadIndex) {
+            threadName.append("-").append(threadIndex.incrementAndGet());
+        }
+        Thread thread = new Thread(r, threadName.toString());
+        if (isDemoThread) {
+            thread.setDaemon(true);
+        }
+        return thread;
+    }
+}

src/test/java/io/openmessaging/storage/dledger/NamedThreadFactoryTest.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2017-2022 The DLedger Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.openmessaging.storage.dledger;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class NamedThreadFactoryTest {
+
+    @Test
+    public void testNamedThreadFactory() {
+
+        NamedThreadFactory threadFactory = new NamedThreadFactory("DledgerThread");
+        Runnable runnable = () -> {
+        };
+        Thread thread = threadFactory.newThread(runnable);
+        Assertions.assertEquals("DledgerThread-1", thread.getName());
+        Assertions.assertFalse(thread.isDaemon());
+
+        threadFactory = new NamedThreadFactory("DledgerThread", true);
+        Thread thread1 = threadFactory.newThread(runnable);
+        Assertions.assertEquals("DledgerThread-1", thread1.getName());
+        Assertions.assertTrue(thread1.isDaemon());
+
+        threadFactory = new NamedThreadFactory(null, "DledgerThread", true);
+        Thread thread2 = threadFactory.newThread(runnable);
+        Assertions.assertEquals("DledgerThread", thread2.getName());
+        Assertions.assertTrue(thread2.isDaemon());
+
+        threadFactory = new NamedThreadFactory(new AtomicInteger(0), "DledgerThread", true);
+        Thread thread3 = threadFactory.newThread(runnable);
+        Assertions.assertEquals("DledgerThread-1", thread3.getName());
+        Assertions.assertTrue(thread2.isDaemon());
+
+    }
+
+}