diff --git a/README.md b/README.md index 91c50fc8..2a4e8ba4 100644 --- a/README.md +++ b/README.md @@ -176,8 +176,9 @@ feat[module]: 新增某一项功能 perf[module]: 优化了模块代码或者优化了什么功能 fix[module]: 修改了什么bug test[module]: 测试了什么东西 -doc[module]: 增加了什么文档 del[module]: 删除了某些功能或者无用代码 +ref[module]: 重命名或者重构了模块 +doc[module]: 增加了什么文档 ``` Ⅶ. License diff --git a/event/src/main/java/com/zfoo/event/manager/EventBus.java b/event/src/main/java/com/zfoo/event/manager/EventBus.java index 1c1783e5..6f2753b8 100644 --- a/event/src/main/java/com/zfoo/event/manager/EventBus.java +++ b/event/src/main/java/com/zfoo/event/manager/EventBus.java @@ -91,7 +91,7 @@ public abstract class EventBus { return executors[RandomUtils.randomInt(EXECUTORS_SIZE)]; } - public static Executor asyncExecute(int hashcode) { + public static Executor execute(int hashcode) { return executors[Math.abs(hashcode % EXECUTORS_SIZE)]; } diff --git a/net/src/main/java/com/zfoo/net/NetContext.java b/net/src/main/java/com/zfoo/net/NetContext.java index f5b9f25d..1158eb45 100644 --- a/net/src/main/java/com/zfoo/net/NetContext.java +++ b/net/src/main/java/com/zfoo/net/NetContext.java @@ -21,7 +21,7 @@ import com.zfoo.net.dispatcher.manager.IPacketDispatcher; import com.zfoo.net.packet.service.IPacketService; import com.zfoo.net.session.manager.ISessionManager; import com.zfoo.net.session.model.Session; -import com.zfoo.net.task.TaskManager; +import com.zfoo.net.task.TaskBus; import com.zfoo.protocol.collection.ArrayUtils; import com.zfoo.protocol.exception.ExceptionUtils; import com.zfoo.protocol.util.IOUtils; @@ -133,10 +133,10 @@ public class NetContext implements ApplicationListener, // 关闭TaskManager try { - Field field = TaskManager.class.getDeclaredField("executors"); + Field field = TaskBus.class.getDeclaredField("executors"); ReflectionUtils.makeAccessible(field); - var executors = (ExecutorService[]) ReflectionUtils.getField(field, TaskManager.getInstance()); + var executors = (ExecutorService[]) ReflectionUtils.getField(field, null); for (ExecutorService executor : executors) { ThreadUtils.shutdown(executor); } diff --git a/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketDispatcher.java b/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketDispatcher.java index ce82b3d1..604f4992 100644 --- a/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketDispatcher.java +++ b/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketDispatcher.java @@ -31,7 +31,7 @@ import com.zfoo.net.packet.model.IPacketAttachment; import com.zfoo.net.packet.model.SignalPacketAttachment; import com.zfoo.net.session.model.AttributeType; import com.zfoo.net.session.model.Session; -import com.zfoo.net.task.TaskManager; +import com.zfoo.net.task.TaskBus; import com.zfoo.net.task.model.ReceiveTask; import com.zfoo.protocol.IPacket; import com.zfoo.protocol.exception.ExceptionUtils; @@ -136,7 +136,7 @@ public class PacketDispatcher implements IPacketDispatcher { } // 正常发送消息的接收 - TaskManager.getInstance().addTask(new ReceiveTask(session, packet, packetAttachment)); + TaskBus.submit(new ReceiveTask(session, packet, packetAttachment)); } @Override @@ -265,7 +265,7 @@ public class PacketDispatcher implements IPacketDispatcher { } } - }, TaskManager.getInstance().getExecutorByConsistentHash(executorConsistentHash)); + }, TaskBus.executor(executorConsistentHash)); PacketSignal.addSignalAttachment(clientAttachment); diff --git a/net/src/main/java/com/zfoo/net/task/ITaskManager.java b/net/src/main/java/com/zfoo/net/task/ITaskManager.java deleted file mode 100644 index 55ed1245..00000000 --- a/net/src/main/java/com/zfoo/net/task/ITaskManager.java +++ /dev/null @@ -1,29 +0,0 @@ -/* - * Copyright (C) 2020 The zfoo 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 - * - * http://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 com.zfoo.net.task; - -import com.zfoo.net.task.model.ReceiveTask; - -import java.util.concurrent.ExecutorService; - -/** - * @author jaysunxiao - * @version 3.0 - */ -public interface ITaskManager { - - void addTask(ReceiveTask task); - - ExecutorService getExecutorByConsistentHash(int hash); -} diff --git a/net/src/main/java/com/zfoo/net/task/TaskManager.java b/net/src/main/java/com/zfoo/net/task/TaskBus.java similarity index 54% rename from net/src/main/java/com/zfoo/net/task/TaskManager.java rename to net/src/main/java/com/zfoo/net/task/TaskBus.java index 2795eace..58d9b5aa 100644 --- a/net/src/main/java/com/zfoo/net/task/TaskManager.java +++ b/net/src/main/java/com/zfoo/net/task/TaskBus.java @@ -14,33 +14,28 @@ package com.zfoo.net.task; import com.zfoo.net.NetContext; -import com.zfoo.net.task.model.AbstractTaskDispatch; -import com.zfoo.net.task.model.ITaskDispatch; import com.zfoo.net.task.model.ReceiveTask; +import com.zfoo.net.task.route.AbstractTaskRoute; +import com.zfoo.net.task.route.ITaskRoute; import com.zfoo.protocol.util.StringUtils; -import io.netty.util.concurrent.FastThreadLocalThread; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -import java.util.concurrent.ThreadFactory; -import java.util.concurrent.atomic.AtomicInteger; /** * @author jaysunxiao * @version 3.0 */ -public final class TaskManager implements ITaskManager { +public final class TaskBus { - private static final Logger logger = LoggerFactory.getLogger(TaskManager.class); - - private static final TaskManager INSTANCE = new TaskManager(); + private static final Logger logger = LoggerFactory.getLogger(TaskBus.class); // 线程池的大小 public static final int EXECUTOR_SIZE; - private static final ITaskDispatch taskDispatch; + private static final ITaskRoute taskRoute; /** @@ -57,7 +52,7 @@ public final class TaskManager implements ITaskManager { ? "default" : providerConfig.getDispatchThread(); EXECUTOR_SIZE = "default".equals(dispatchThread) ? (Runtime.getRuntime().availableProcessors() + 1) : Integer.parseInt(dispatchThread); - taskDispatch = AbstractTaskDispatch.valueOf(dispatch); + taskRoute = AbstractTaskRoute.valueOf(dispatch); executors = new ExecutorService[EXECUTOR_SIZE]; for (int i = 0; i < executors.length; i++) { @@ -66,42 +61,12 @@ public final class TaskManager implements ITaskManager { } } - private static class TaskThreadFactory implements ThreadFactory { - private static final AtomicInteger poolNumber = new AtomicInteger(1); - private final ThreadGroup group; - private final AtomicInteger threadNumber = new AtomicInteger(1); - private final String namePrefix; - TaskThreadFactory() { - var s = System.getSecurityManager(); - group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup(); - namePrefix = "task-p" + poolNumber.getAndIncrement() + "-t"; - } - - @Override - public Thread newThread(Runnable runnable) { - var t = new FastThreadLocalThread(group, runnable, namePrefix + threadNumber.getAndIncrement(), 0); - t.setDaemon(false); - t.setPriority(Thread.NORM_PRIORITY); - t.setUncaughtExceptionHandler((thread, e) -> logger.error(thread.toString(), e)); - return t; - } + public static void submit(ReceiveTask task) { + taskRoute.getExecutor(task).execute(task); } - private TaskManager() { - } - - public static TaskManager getInstance() { - return INSTANCE; - } - - @Override - public void addTask(ReceiveTask task) { - taskDispatch.getExecutor(task).execute(task); - } - - @Override - public ExecutorService getExecutorByConsistentHash(int executorConsistentHash) { + public static ExecutorService executor(int executorConsistentHash) { return executors[Math.abs(executorConsistentHash % EXECUTOR_SIZE)]; } } diff --git a/net/src/main/java/com/zfoo/net/task/TaskThreadFactory.java b/net/src/main/java/com/zfoo/net/task/TaskThreadFactory.java new file mode 100644 index 00000000..9b5f71cc --- /dev/null +++ b/net/src/main/java/com/zfoo/net/task/TaskThreadFactory.java @@ -0,0 +1,51 @@ +/* + * Copyright (C) 2020 The zfoo 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 + * + * http://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 com.zfoo.net.task; + +import io.netty.util.concurrent.FastThreadLocalThread; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * @author jaysunxiao + * @version 3.0 + */ +public class TaskThreadFactory implements ThreadFactory { + + private static final Logger logger = LoggerFactory.getLogger(TaskThreadFactory.class); + + private static final AtomicInteger poolNumber = new AtomicInteger(1); + private final ThreadGroup group; + private final AtomicInteger threadNumber = new AtomicInteger(1); + private final String namePrefix; + + TaskThreadFactory() { + var s = System.getSecurityManager(); + group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup(); + namePrefix = "task-p" + poolNumber.getAndIncrement() + "-t"; + } + + @Override + public Thread newThread(Runnable runnable) { + var t = new FastThreadLocalThread(group, runnable, namePrefix + threadNumber.getAndIncrement(), 0); + t.setDaemon(false); + t.setPriority(Thread.NORM_PRIORITY); + t.setUncaughtExceptionHandler((thread, e) -> logger.error(thread.toString(), e)); + return t; + } + +} diff --git a/net/src/main/java/com/zfoo/net/task/model/AbstractTaskDispatch.java b/net/src/main/java/com/zfoo/net/task/route/AbstractTaskRoute.java similarity index 74% rename from net/src/main/java/com/zfoo/net/task/model/AbstractTaskDispatch.java rename to net/src/main/java/com/zfoo/net/task/route/AbstractTaskRoute.java index 41ea4de2..176991e8 100644 --- a/net/src/main/java/com/zfoo/net/task/model/AbstractTaskDispatch.java +++ b/net/src/main/java/com/zfoo/net/task/route/AbstractTaskRoute.java @@ -1,6 +1,5 @@ /* * Copyright (C) 2020 The zfoo 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 * @@ -9,9 +8,10 @@ * 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 com.zfoo.net.task.model; +package com.zfoo.net.task.route; import com.zfoo.protocol.util.StringUtils; @@ -19,16 +19,16 @@ import com.zfoo.protocol.util.StringUtils; * @author jaysunxiao * @version 3.0 */ -public abstract class AbstractTaskDispatch implements ITaskDispatch { +public abstract class AbstractTaskRoute implements ITaskRoute { - public static ITaskDispatch valueOf(String taskDispatchName) { + public static ITaskRoute valueOf(String taskDispatchName) { switch (taskDispatchName) { case "random": - return new RandomTaskDispatch(); + return new RandomTaskRoute(); case "sessionId": - return new SessionIdTaskDispatch(); + return new SessionIdTaskRoute(); case "consistent-hash": - return new ConsistentHashTaskDispatch(); + return new ConsistentHashTaskRoute(); default: throw new RuntimeException(StringUtils.format("没有找到对应的taskDispatch[{}]", taskDispatchName)); } diff --git a/net/src/main/java/com/zfoo/net/task/model/ConsistentHashTaskDispatch.java b/net/src/main/java/com/zfoo/net/task/route/ConsistentHashTaskRoute.java similarity index 65% rename from net/src/main/java/com/zfoo/net/task/model/ConsistentHashTaskDispatch.java rename to net/src/main/java/com/zfoo/net/task/route/ConsistentHashTaskRoute.java index 3fc72bae..c66212f4 100644 --- a/net/src/main/java/com/zfoo/net/task/model/ConsistentHashTaskDispatch.java +++ b/net/src/main/java/com/zfoo/net/task/route/ConsistentHashTaskRoute.java @@ -1,6 +1,5 @@ /* * Copyright (C) 2020 The zfoo 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 * @@ -9,11 +8,13 @@ * 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 com.zfoo.net.task.model; +package com.zfoo.net.task.route; -import com.zfoo.net.task.TaskManager; +import com.zfoo.net.task.TaskBus; +import com.zfoo.net.task.model.ReceiveTask; import java.util.concurrent.ExecutorService; @@ -21,11 +22,11 @@ import java.util.concurrent.ExecutorService; * @author jaysunxiao * @version 3.0 */ -public class ConsistentHashTaskDispatch extends AbstractTaskDispatch { +public class ConsistentHashTaskRoute extends AbstractTaskRoute { - private static ConsistentHashTaskDispatch INSTANCE = new ConsistentHashTaskDispatch(); + private static ConsistentHashTaskRoute INSTANCE = new ConsistentHashTaskRoute(); - public static ConsistentHashTaskDispatch getINSTANCE() { + public static ConsistentHashTaskRoute getINSTANCE() { return INSTANCE; } @@ -34,10 +35,10 @@ public class ConsistentHashTaskDispatch extends AbstractTaskDispatch { var packetAttachment = receiveTask.getPacketAttachment(); if (packetAttachment == null) { - return SessionIdTaskDispatch.getInstance().getExecutor(receiveTask); + return SessionIdTaskRoute.getInstance().getExecutor(receiveTask); } - return TaskManager.getInstance().getExecutorByConsistentHash(packetAttachment.executorConsistentHash()); + return TaskBus.executor(packetAttachment.executorConsistentHash()); } } diff --git a/net/src/main/java/com/zfoo/net/task/model/ITaskDispatch.java b/net/src/main/java/com/zfoo/net/task/route/ITaskRoute.java similarity index 87% rename from net/src/main/java/com/zfoo/net/task/model/ITaskDispatch.java rename to net/src/main/java/com/zfoo/net/task/route/ITaskRoute.java index ca8d36ad..4ef10632 100644 --- a/net/src/main/java/com/zfoo/net/task/model/ITaskDispatch.java +++ b/net/src/main/java/com/zfoo/net/task/route/ITaskRoute.java @@ -1,6 +1,5 @@ /* * Copyright (C) 2020 The zfoo 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 * @@ -9,9 +8,12 @@ * 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 com.zfoo.net.task.model; +package com.zfoo.net.task.route; + +import com.zfoo.net.task.model.ReceiveTask; import java.util.concurrent.ExecutorService; @@ -19,7 +21,7 @@ import java.util.concurrent.ExecutorService; * @author jaysunxiao * @version 3.0 */ -public interface ITaskDispatch { +public interface ITaskRoute { ExecutorService getExecutor(ReceiveTask receiveTask); diff --git a/net/src/main/java/com/zfoo/net/task/model/RandomTaskDispatch.java b/net/src/main/java/com/zfoo/net/task/route/RandomTaskRoute.java similarity index 70% rename from net/src/main/java/com/zfoo/net/task/model/RandomTaskDispatch.java rename to net/src/main/java/com/zfoo/net/task/route/RandomTaskRoute.java index 9c4ea2a0..81617520 100644 --- a/net/src/main/java/com/zfoo/net/task/model/RandomTaskDispatch.java +++ b/net/src/main/java/com/zfoo/net/task/route/RandomTaskRoute.java @@ -1,6 +1,5 @@ /* * Copyright (C) 2020 The zfoo 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 * @@ -9,11 +8,13 @@ * 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 com.zfoo.net.task.model; +package com.zfoo.net.task.route; -import com.zfoo.net.task.TaskManager; +import com.zfoo.net.task.TaskBus; +import com.zfoo.net.task.model.ReceiveTask; import java.util.concurrent.ExecutorService; @@ -21,17 +22,17 @@ import java.util.concurrent.ExecutorService; * @author jaysunxiao * @version 3.0 */ -public class RandomTaskDispatch extends AbstractTaskDispatch { +public class RandomTaskRoute extends AbstractTaskRoute { - private static final RandomTaskDispatch INSTANCE = new RandomTaskDispatch(); + private static final RandomTaskRoute INSTANCE = new RandomTaskRoute(); - public static RandomTaskDispatch getInstance() { + public static RandomTaskRoute getInstance() { return INSTANCE; } @Override public ExecutorService getExecutor(ReceiveTask receiveTask) { - return TaskManager.getInstance().getExecutorByConsistentHash(-1); + return TaskBus.executor(-1); } } diff --git a/net/src/main/java/com/zfoo/net/task/model/SessionIdTaskDispatch.java b/net/src/main/java/com/zfoo/net/task/route/SessionIdTaskRoute.java similarity index 71% rename from net/src/main/java/com/zfoo/net/task/model/SessionIdTaskDispatch.java rename to net/src/main/java/com/zfoo/net/task/route/SessionIdTaskRoute.java index 83b05413..fff1dd66 100644 --- a/net/src/main/java/com/zfoo/net/task/model/SessionIdTaskDispatch.java +++ b/net/src/main/java/com/zfoo/net/task/route/SessionIdTaskRoute.java @@ -1,6 +1,5 @@ /* * Copyright (C) 2020 The zfoo 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 * @@ -9,11 +8,13 @@ * 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 com.zfoo.net.task.model; +package com.zfoo.net.task.route; -import com.zfoo.net.task.TaskManager; +import com.zfoo.net.task.TaskBus; +import com.zfoo.net.task.model.ReceiveTask; import com.zfoo.util.math.HashUtils; import java.util.concurrent.ExecutorService; @@ -24,18 +25,18 @@ import java.util.concurrent.ExecutorService; * @author jaysunxiao * @version 3.0 */ -public class SessionIdTaskDispatch extends AbstractTaskDispatch { +public class SessionIdTaskRoute extends AbstractTaskRoute { - private static final SessionIdTaskDispatch INSTANCE = new SessionIdTaskDispatch(); + private static final SessionIdTaskRoute INSTANCE = new SessionIdTaskRoute(); - public static SessionIdTaskDispatch getInstance() { + public static SessionIdTaskRoute getInstance() { return INSTANCE; } @Override public ExecutorService getExecutor(ReceiveTask receiveTask) { var session = receiveTask.getSession(); - return TaskManager.getInstance().getExecutorByConsistentHash(HashUtils.fnvHash(session.getSid())); + return TaskBus.executor(HashUtils.fnvHash(session.getSid())); } } diff --git a/net/src/test/java/com/zfoo/net/dispatcher/PacketSignalTest.java b/net/src/test/java/com/zfoo/net/dispatcher/PacketSignalTest.java index 366dd90e..56361134 100644 --- a/net/src/test/java/com/zfoo/net/dispatcher/PacketSignalTest.java +++ b/net/src/test/java/com/zfoo/net/dispatcher/PacketSignalTest.java @@ -54,7 +54,7 @@ public class PacketSignalTest { var countDownLatch = new CountDownLatch(executorSize); for (var i = 0; i < executorSize; i++) { - EventBus.asyncExecute(i).execute(new Runnable() { + EventBus.execute(i).execute(new Runnable() { @Override public void run() { addAndRemoveArray(); @@ -72,7 +72,7 @@ public class PacketSignalTest { var countDownLatch = new CountDownLatch(executorSize); for (int i = 0; i < executorSize; i++) { - EventBus.asyncExecute(i).execute(new Runnable() { + EventBus.execute(i).execute(new Runnable() { @Override public void run() { addAndRemoveMap(); diff --git a/orm/src/main/java/com/zfoo/orm/model/persister/CronOrmPersister.java b/orm/src/main/java/com/zfoo/orm/model/persister/CronOrmPersister.java index 679cc0f6..8a09cb33 100644 --- a/orm/src/main/java/com/zfoo/orm/model/persister/CronOrmPersister.java +++ b/orm/src/main/java/com/zfoo/orm/model/persister/CronOrmPersister.java @@ -78,7 +78,7 @@ public class CronOrmPersister extends AbstractOrmPersister { @Override public void run() { if (!OrmContext.isStop()) { - EventBus.asyncExecute(entityDef.getClazz().hashCode()).execute(new Runnable() { + EventBus.execute(entityDef.getClazz().hashCode()).execute(new Runnable() { @Override public void run() { entityCaches.persistAll(); diff --git a/orm/src/main/java/com/zfoo/orm/model/persister/TimeOrmPersister.java b/orm/src/main/java/com/zfoo/orm/model/persister/TimeOrmPersister.java index 47d8c32a..d0ded106 100644 --- a/orm/src/main/java/com/zfoo/orm/model/persister/TimeOrmPersister.java +++ b/orm/src/main/java/com/zfoo/orm/model/persister/TimeOrmPersister.java @@ -47,7 +47,7 @@ public class TimeOrmPersister extends AbstractOrmPersister { @Override public void run() { if (!OrmContext.isStop()) { - EventBus.asyncExecute(entityDef.getClazz().hashCode()).execute(new Runnable() { + EventBus.execute(entityDef.getClazz().hashCode()).execute(new Runnable() { @Override public void run() { entityCaches.persistAll();