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

feat[event]: support virtual thread api

Browse Source
This commit is contained in:
godotg
2023-03-10 14:27:41 +08:00
parent da9cb5f608
commit 4c98f97684
8 changed files with 86 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
event/src/main/java/com/zfoo/event/manager/EventBus.java
+26 -36
View File
@@ -10,7 +10,6 @@
* 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.event.manager;
import com.zfoo.event.model.event.IEvent;
@@ -54,13 +53,9 @@ public abstract class EventBus {
private static final CopyOnWriteHashMapLongObject<ExecutorService> threadMap = new CopyOnWriteHashMapLongObject<>(EXECUTORS_SIZE);
/**
* Synchronous event mapping, synchronize observers
* event mapping
*/
private static final Map<Class<? extends IEvent>, List<IEventReceiver>> receiverMapSync = new HashMap<>();
/**
* Asynchronous event mapping, asynchronous observer
*/
private static final Map<Class<? extends IEvent>, List<IEventReceiver>> receiverMapAsync = new HashMap<>();
private static final Map<Class<? extends IEvent>, List<IEventReceiver>> receiverMap = new HashMap<>();
static {
for (int i = 0; i < executors.length; i++) {
@@ -102,32 +97,31 @@ public abstract class EventBus {
return;
}
var clazz = event.getClass();
var listSync = receiverMapSync.get(clazz);
if (CollectionUtils.isNotEmpty(listSync)) {
for (var receiver : listSync) {
try {
receiver.invoke(event);
} catch (Exception e) {
logger.error("eventBus sync event [{}] unknown exception", clazz.getSimpleName(), e);
} catch (Throwable t) {
logger.error("eventBus sync event [{}] unknown error", clazz.getSimpleName(), t);
}
var receivers = receiverMap.get(clazz);
if (CollectionUtils.isEmpty(receivers)) {
return;
}
for (var receiver : receivers) {
switch (receiver.bus()) {
case CurrentThread:
doReceiver(receiver, event);
break;
case AsyncThread:
execute(event.executorHash(), () -> doReceiver(receiver, event));
break;
case VirtualThread:
break;
}
}
}
var listAsync = receiverMapAsync.get(clazz);
if (CollectionUtils.isNotEmpty(listAsync)) {
for (var receiver : listAsync) {
execute(event.executorHash(), () -> {
try {
receiver.invoke(event);
} catch (Exception e) {
logger.error("eventBus async event [{}] unknown exception", clazz.getSimpleName(), e);
} catch (Throwable t) {
logger.error("eventBus async event [{}] unknown error", clazz.getSimpleName(), t);
}
});
}
private static void doReceiver(IEventReceiver receiver, IEvent event) {
try {
receiver.invoke(event);
} catch (Exception e) {
logger.error("eventBus {} [{}] unknown exception", receiver.bus(), event.getClass().getSimpleName(), e);
} catch (Throwable t) {
logger.error("eventBus {} [{}] unknown error", receiver.bus(), event.getClass().getSimpleName(), t);
}
}
@@ -146,12 +140,8 @@ public abstract class EventBus {
/**
* Register the event and its counterpart observer
*/
public static void registerEventReceiver(Class<? extends IEvent> eventType, IEventReceiver receiver, boolean asyncFlag) {
if (asyncFlag) {
receiverMapAsync.computeIfAbsent(eventType, it -> new ArrayList<>(1)).add(receiver);
} else {
receiverMapSync.computeIfAbsent(eventType, it -> new ArrayList<>(1)).add(receiver);
}
public static void registerEventReceiver(Class<? extends IEvent> eventType, IEventReceiver receiver) {
receiverMap.computeIfAbsent(eventType, it -> new ArrayList<>(1)).add(receiver);
}
public static Executor threadExecutor(long currentThreadId) {
event/src/main/java/com/zfoo/event/model/anno/Bus.java
+29
View File
@@ -0,0 +1,29 @@
/*
* 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.event.model.anno;
/**
* event bus thread type
*
* @author godotg
* @version 3.0
*/
public enum Bus {
CurrentThread,
AsyncThread,
VirtualThread;
}
event/src/main/java/com/zfoo/event/model/anno/EventReceiver.java
+1 -3
View File
@@ -10,7 +10,6 @@
* 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.event.model.anno;
import java.lang.annotation.*;
@@ -21,10 +20,9 @@ import java.lang.annotation.*;
* @author godotg
* @version 3.0
*/
@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD})
public @interface EventReceiver {
boolean async() default false;
Bus value() default Bus.CurrentThread;
}
event/src/main/java/com/zfoo/event/model/vo/EnhanceUtils.java
+10 -2
View File
@@ -13,6 +13,7 @@
package com.zfoo.event.model.vo;
import com.zfoo.event.model.anno.Bus;
import com.zfoo.event.model.event.IEvent;
import com.zfoo.event.schema.NamespaceHandler;
import com.zfoo.protocol.util.StringUtils;
@@ -69,13 +70,20 @@ public abstract class EnhanceUtils {
constructor.setModifiers(Modifier.PUBLIC);
enhanceClazz.addConstructor(constructor);
// 定义类实现的接口方法
// 定义类实现的接口方法invoker
CtMethod invokeMethod = new CtMethod(classPool.get(void.class.getCanonicalName()), "invoke", classPool.get(new String[]{IEvent.class.getCanonicalName()}), enhanceClazz);
invokeMethod.setModifiers(Modifier.PUBLIC + Modifier.FINAL);
String invokeMethodBody = "{this.bean." + method.getName() + "((" + clazz.getCanonicalName() + ")$1);}";// 强制类型转换，转换为具体的Event类型的类型
String invokeMethodBody = StringUtils.format("{ this.bean.{}(({})$1); }", method.getName(), clazz.getCanonicalName()); // 强制类型转换，转换为具体的Event类型的类型
invokeMethod.setBody(invokeMethodBody);
enhanceClazz.addMethod(invokeMethod);
// 定义类实现的接口方法bus
CtMethod busMethod = new CtMethod(classPool.get(Bus.class.getCanonicalName()), "bus", null, enhanceClazz);
busMethod.setModifiers(Modifier.PUBLIC + Modifier.FINAL);
String busMethodBody = StringUtils.format("{ return {}.{}; }", Bus.class.getCanonicalName(), definition.getBus());
busMethod.setBody(busMethodBody);
enhanceClazz.addMethod(busMethod);
// 释放缓存
enhanceClazz.detach();
event/src/main/java/com/zfoo/event/model/vo/EventReceiverDefinition.java
+13 -11
View File
@@ -13,6 +13,7 @@
package com.zfoo.event.model.vo;
import com.zfoo.event.model.anno.Bus;
import com.zfoo.event.model.event.IEvent;
import com.zfoo.protocol.util.ReflectionUtils;
@@ -35,11 +36,20 @@ public class EventReceiverDefinition implements IEventReceiver {
// 接收的参数Class
private Class<? extends IEvent> eventClazz;
public EventReceiverDefinition(Object bean, Method method, Class<? extends IEvent> eventClazz) {
// 事件接收方式
private Bus bus;
public EventReceiverDefinition(Object bean, Method method, Class<? extends IEvent> eventClazz, Bus bus) {
this.bean = bean;
this.method = method;
this.eventClazz = eventClazz;
ReflectionUtils.makeAccessible(this.method);
this.bus = bus;
}
@Override
public Bus bus() {
return bus;
}
@Override
@@ -51,23 +61,15 @@ public class EventReceiverDefinition implements IEventReceiver {
return bean;
}
public void setBean(Object bean) {
this.bean = bean;
}
public Method getMethod() {
return method;
}
public void setMethod(Method method) {
this.method = method;
}
public Class<? extends IEvent> getEventClazz() {
return eventClazz;
}
public void setEventClazz(Class<? extends IEvent> eventClazz) {
this.eventClazz = eventClazz;
public Bus getBus() {
return bus;
}
}
event/src/main/java/com/zfoo/event/model/vo/IEventReceiver.java
+2 -1
View File
@@ -13,6 +13,7 @@
package com.zfoo.event.model.vo;
import com.zfoo.event.model.anno.Bus;
import com.zfoo.event.model.event.IEvent;
/**
@@ -20,7 +21,7 @@ import com.zfoo.event.model.event.IEvent;
* @version 3.0
*/
public interface IEventReceiver {
Bus bus();
void invoke(IEvent event);
}
event/src/main/java/com/zfoo/event/schema/EventRegisterProcessor.java
+3 -4
View File
@@ -79,13 +79,12 @@ public class EventRegisterProcessor implements BeanPostProcessor {
, bean.getClass().getName(), methodName, eventName, expectedMethodName));
}
var receiverDefinition = new EventReceiverDefinition(bean, method, eventClazz);
var bus = method.getDeclaredAnnotation(EventReceiver.class).value();
var receiverDefinition = new EventReceiverDefinition(bean, method, eventClazz, bus);
var enhanceReceiverDefinition = EnhanceUtils.createEventReceiver(receiverDefinition);
//异步执行标志，false表示同步执行，true表示异步执行
var asyncFlag = method.getDeclaredAnnotation(EventReceiver.class).async();
// key:class类型 value:观察者 注册Event的receiverMap中
EventBus.registerEventReceiver(eventClazz, enhanceReceiverDefinition, asyncFlag);
EventBus.registerEventReceiver(eventClazz, enhanceReceiverDefinition);
}
} catch (Throwable t) {
throw new RuntimeException(t);
event/src/test/java/com/zfoo/event/MyController2.java
+2 -1
View File
@@ -13,6 +13,7 @@
package com.zfoo.event;
import com.zfoo.event.model.anno.Bus;
import com.zfoo.event.model.anno.EventReceiver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -32,7 +33,7 @@ public class MyController2 {
*
* 异步事件会被不会立刻执行，注意日志打印的线程号
*/
@EventReceiver(async = true)
@EventReceiver(Bus.AsyncThread)
public void onMyNoticeEvent(MyNoticeEvent event) {
logger.info("方法2异步执行事件：" + event.getMessage());
}