tiennm99/zfoo
1
0
Fork 0
mirror of https://github.com/tiennm99/zfoo.git synced 2026-08-09 02:25:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

revert[event]: revert vent

Browse Source
This commit is contained in:
godotg
2024-04-10 19:27:07 +08:00
parent ad67c2f5ec
commit 966fe14d89
7 changed files with 55 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
event/src/main/java/com/zfoo/event/anno/Bus.java
+1 -4
View File
@@ -22,9 +22,6 @@ public enum Bus {
AsyncThread,
VirtualThread,
ManualThread
;
VirtualThread;
}
event/src/main/java/com/zfoo/event/manager/EventBus.java
+50 -14
View File
@@ -16,8 +16,11 @@ import com.zfoo.event.enhance.IEventReceiver;
import com.zfoo.event.model.IEvent;
import com.zfoo.protocol.collection.CollectionUtils;
import com.zfoo.protocol.collection.concurrent.CopyOnWriteHashMapLongObject;
import com.zfoo.protocol.util.AssertionUtils;
import com.zfoo.protocol.util.RandomUtils;
import com.zfoo.protocol.util.StringUtils;
import com.zfoo.protocol.util.ThreadUtils;
import io.netty.util.concurrent.FastThreadLocalThread;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -27,6 +30,9 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
@@ -37,22 +43,56 @@ public abstract class EventBus {
private static final Logger logger = LoggerFactory.getLogger(EventBus.class);
/**
* EN: The size of the thread pool. Event's thread pool is often used to do time-consuming operations, so set it a little bigger
* CN: 线程池的大小. event的线程池经常用来做一些耗时的操作，所以要设置大一点
*/
private static final int EXECUTORS_SIZE = Math.max(Runtime.getRuntime().availableProcessors(), 4) * 2 + 1;
private static final ExecutorService[] executors = new ExecutorService[EXECUTORS_SIZE];
private static final CopyOnWriteHashMapLongObject<ExecutorService> threadMap = new CopyOnWriteHashMapLongObject<>(EXECUTORS_SIZE);
/**
* event mapping
*/
private static final Map<Class<? extends IEvent>, List<IEventReceiver>> receiverMap = new HashMap<>();
/**
* custom thread event receiver
*/
public static BiConsumer<IEventReceiver, IEvent> manualThreadHandler = EventBus::doReceiver;
/**
* event exception handler
*/
public static BiConsumer<IEventReceiver, IEvent> exceptionHandler = null;
public static Consumer<IEvent> noEventReceiverHandler = null;
static {
for (int i = 0; i < executors.length; i++) {
var namedThreadFactory = new EventThreadFactory(i);
var executor = Executors.newSingleThreadExecutor(namedThreadFactory);
executors[i] = executor;
}
}
public static class EventThreadFactory implements ThreadFactory {
private final int poolNumber;
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final ThreadGroup group;
public EventThreadFactory(int poolNumber) {
this.group = Thread.currentThread().getThreadGroup();
this.poolNumber = poolNumber;
}
@Override
public Thread newThread(Runnable runnable) {
var threadName = StringUtils.format("event-p{}-t{}", poolNumber + 1, threadNumber.getAndIncrement());
var thread = new FastThreadLocalThread(group, runnable, threadName);
thread.setDaemon(false);
thread.setPriority(Thread.NORM_PRIORITY);
thread.setUncaughtExceptionHandler((t, e) -> logger.error(t.toString(), e));
var executor = executors[poolNumber];
AssertionUtils.notNull(executor);
threadMap.put(thread.getId(), executor);
return thread;
}
}
/**
* Publish the event
@@ -74,9 +114,8 @@ public abstract class EventBus {
for (var receiver : receivers) {
switch (receiver.bus()) {
case CurrentThread -> doReceiver(receiver, event);
case AsyncThread -> asyncExecute(event.executorHash(), () -> doReceiver(receiver, event));
case AsyncThread -> execute(event.executorHash(), () -> doReceiver(receiver, event));
// case VirtualThread -> Thread.ofVirtual().name("virtual-on" + clazz.getSimpleName()).start(() -> doReceiver(receiver, event));
case ManualThread -> manualThreadHandler.accept(receiver, event);
}
}
}
@@ -93,14 +132,14 @@ public abstract class EventBus {
}
public static void asyncExecute(Runnable runnable) {
asyncExecute(RandomUtils.randomInt(), runnable);
execute(RandomUtils.randomInt(), runnable);
}
/**
* Use the event thread specified by the hashcode to execute the task
*/
public static void asyncExecute(int executorHash, Runnable runnable) {
EventExecutors.execute(executorHash, ThreadUtils.safeRunnable(runnable));
public static void execute(int executorHash, Runnable runnable) {
executors[Math.abs(executorHash % EXECUTORS_SIZE)].execute(ThreadUtils.safeRunnable(runnable));
}
/**
@@ -110,9 +149,6 @@ public abstract class EventBus {
receiverMap.computeIfAbsent(eventType, it -> new ArrayList<>(1)).add(receiver);
}
// ------------------------------------------------------------------------------------------------------------------
static final CopyOnWriteHashMapLongObject<ExecutorService> threadMap = new CopyOnWriteHashMapLongObject<>();
public static Executor threadExecutor(long currentThreadId) {
return threadMap.getPrimitive(currentThreadId);
}
event/src/main/java/com/zfoo/event/manager/EventExecutors.java
-70
View File
@@ -1,70 +0,0 @@
package com.zfoo.event.manager;
import com.zfoo.protocol.util.AssertionUtils;
import com.zfoo.protocol.util.StringUtils;
import com.zfoo.protocol.util.ThreadUtils;
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 godotg
*/
public abstract class EventExecutors {
private static final Logger logger = LoggerFactory.getLogger(EventExecutors.class);
/**
* EN: The size of the thread pool. Event's thread pool is often used to do time-consuming operations, so set it a little bigger
* CN: 线程池的大小. event的线程池经常用来做一些耗时的操作，所以要设置大一点
*/
private static final int EXECUTORS_SIZE = Math.max(Runtime.getRuntime().availableProcessors(), 4) * 2 + 1;
private static final ExecutorService[] executors = new ExecutorService[EXECUTORS_SIZE];
static {
for (int i = 0; i < executors.length; i++) {
var namedThreadFactory = new EventThreadFactory(i);
var executor = Executors.newSingleThreadExecutor(namedThreadFactory);
executors[i] = executor;
}
}
public static class EventThreadFactory implements ThreadFactory {
private final int poolNumber;
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final ThreadGroup group;
public EventThreadFactory(int poolNumber) {
this.group = Thread.currentThread().getThreadGroup();
this.poolNumber = poolNumber;
}
@Override
public Thread newThread(Runnable runnable) {
var threadName = StringUtils.format("event-p{}-t{}", poolNumber + 1, threadNumber.getAndIncrement());
var thread = new FastThreadLocalThread(group, runnable, threadName);
thread.setDaemon(false);
thread.setPriority(Thread.NORM_PRIORITY);
thread.setUncaughtExceptionHandler((t, e) -> logger.error(t.toString(), e));
var executor = executors[poolNumber];
AssertionUtils.notNull(executor);
EventBus.threadMap.put(thread.getId(), executor);
return thread;
}
}
/**
* Use the event thread specified by the hashcode to execute the task
*/
public static void execute(int executorHash, Runnable runnable) {
executors[Math.abs(executorHash % EXECUTORS_SIZE)].execute(ThreadUtils.safeRunnable(runnable));
}
}
net/src/test/java/com/zfoo/net/router/SignalBridgeTest.java
+1 -1
View File
@@ -45,7 +45,7 @@ public class SignalBridgeTest {
var countDownLatch = new CountDownLatch(executorSize);
for (var i = 0; i < executorSize; i++) {
EventBus.asyncExecute(i, new Runnable() {
EventBus.execute(i, new Runnable() {
@Override
public void run() {
addAndRemoveArray();
orm/src/main/java/com/zfoo/orm/cache/EntityCache.java
Vendored
+1 -1
View File
@@ -69,7 +69,7 @@ public class EntityCache<PK extends Comparable<PK>, E extends IEntity<PK>> imple
var entity = pnode.getEntity();
@SuppressWarnings("unchecked")
var entityClass = (Class<E>) entityDef.getClazz();
EventBus.asyncExecute(entityClass.hashCode(), new Runnable() {
EventBus.execute(entityClass.hashCode(), new Runnable() {
@Override
public void run() {
var collection = OrmContext.getOrmManager().getCollection(entityClass);
orm/src/main/java/com/zfoo/orm/cache/persister/CronOrmPersister.java
Vendored
+1 -1
View File
@@ -75,7 +75,7 @@ public class CronOrmPersister extends AbstractOrmPersister {
if (!OrmContext.isStop()) {
SchedulerBus.schedule(() -> {
if (!OrmContext.isStop()) {
EventBus.asyncExecute(entityDef.getClazz().hashCode(), () -> {
EventBus.execute(entityDef.getClazz().hashCode(), () -> {
entityCaches.persistAll();
schedulePersist();
});
orm/src/main/java/com/zfoo/orm/cache/persister/TimeOrmPersister.java
Vendored
+1 -1
View File
@@ -44,7 +44,7 @@ public class TimeOrmPersister extends AbstractOrmPersister {
public void start() {
SchedulerBus.scheduleAtFixedRate(() -> {
if (!OrmContext.isStop()) {
EventBus.asyncExecute(entityDef.getClazz().hashCode(), () -> entityCaches.persistAll());
EventBus.execute(entityDef.getClazz().hashCode(), () -> entityCaches.persistAll());
}
}, rate, TimeUnit.MILLISECONDS);
}