From 4c98f97684c9cb96fb61bec2c97891acd5d4b711 Mon Sep 17 00:00:00 2001 From: godotg Date: Fri, 10 Mar 2023 14:27:41 +0800 Subject: [PATCH] feat[event]: support virtual thread api --- .../java/com/zfoo/event/manager/EventBus.java | 62 ++++++++----------- .../java/com/zfoo/event/model/anno/Bus.java | 29 +++++++++ .../zfoo/event/model/anno/EventReceiver.java | 4 +- .../com/zfoo/event/model/vo/EnhanceUtils.java | 12 +++- .../model/vo/EventReceiverDefinition.java | 24 +++---- .../zfoo/event/model/vo/IEventReceiver.java | 3 +- .../event/schema/EventRegisterProcessor.java | 7 +-- .../java/com/zfoo/event/MyController2.java | 3 +- 8 files changed, 86 insertions(+), 58 deletions(-) create mode 100644 event/src/main/java/com/zfoo/event/model/anno/Bus.java 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 a1e2821d..b9dce498 100644 --- a/event/src/main/java/com/zfoo/event/manager/EventBus.java +++ b/event/src/main/java/com/zfoo/event/manager/EventBus.java @@ -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 threadMap = new CopyOnWriteHashMapLongObject<>(EXECUTORS_SIZE); /** - * Synchronous event mapping, synchronize observers + * event mapping */ - private static final Map, List> receiverMapSync = new HashMap<>(); - /** - * Asynchronous event mapping, asynchronous observer - */ - private static final Map, List> receiverMapAsync = new HashMap<>(); + private static final Map, List> 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 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 eventType, IEventReceiver receiver) { + receiverMap.computeIfAbsent(eventType, it -> new ArrayList<>(1)).add(receiver); } public static Executor threadExecutor(long currentThreadId) { diff --git a/event/src/main/java/com/zfoo/event/model/anno/Bus.java b/event/src/main/java/com/zfoo/event/model/anno/Bus.java new file mode 100644 index 00000000..b42d3dea --- /dev/null +++ b/event/src/main/java/com/zfoo/event/model/anno/Bus.java @@ -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; + +} diff --git a/event/src/main/java/com/zfoo/event/model/anno/EventReceiver.java b/event/src/main/java/com/zfoo/event/model/anno/EventReceiver.java index 08a3f191..d81047b7 100644 --- a/event/src/main/java/com/zfoo/event/model/anno/EventReceiver.java +++ b/event/src/main/java/com/zfoo/event/model/anno/EventReceiver.java @@ -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; } diff --git a/event/src/main/java/com/zfoo/event/model/vo/EnhanceUtils.java b/event/src/main/java/com/zfoo/event/model/vo/EnhanceUtils.java index 63ac207d..60d57555 100644 --- a/event/src/main/java/com/zfoo/event/model/vo/EnhanceUtils.java +++ b/event/src/main/java/com/zfoo/event/model/vo/EnhanceUtils.java @@ -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(); diff --git a/event/src/main/java/com/zfoo/event/model/vo/EventReceiverDefinition.java b/event/src/main/java/com/zfoo/event/model/vo/EventReceiverDefinition.java index 94d8ec57..36e46644 100644 --- a/event/src/main/java/com/zfoo/event/model/vo/EventReceiverDefinition.java +++ b/event/src/main/java/com/zfoo/event/model/vo/EventReceiverDefinition.java @@ -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 eventClazz; - public EventReceiverDefinition(Object bean, Method method, Class eventClazz) { + // 事件接收方式 + private Bus bus; + + public EventReceiverDefinition(Object bean, Method method, Class 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 getEventClazz() { return eventClazz; } - public void setEventClazz(Class eventClazz) { - this.eventClazz = eventClazz; + public Bus getBus() { + return bus; } } diff --git a/event/src/main/java/com/zfoo/event/model/vo/IEventReceiver.java b/event/src/main/java/com/zfoo/event/model/vo/IEventReceiver.java index 77d39555..01a544b5 100644 --- a/event/src/main/java/com/zfoo/event/model/vo/IEventReceiver.java +++ b/event/src/main/java/com/zfoo/event/model/vo/IEventReceiver.java @@ -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); - } diff --git a/event/src/main/java/com/zfoo/event/schema/EventRegisterProcessor.java b/event/src/main/java/com/zfoo/event/schema/EventRegisterProcessor.java index 707a6c53..6e450dfa 100644 --- a/event/src/main/java/com/zfoo/event/schema/EventRegisterProcessor.java +++ b/event/src/main/java/com/zfoo/event/schema/EventRegisterProcessor.java @@ -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); diff --git a/event/src/test/java/com/zfoo/event/MyController2.java b/event/src/test/java/com/zfoo/event/MyController2.java index d0a6f459..cbdf724f 100644 --- a/event/src/test/java/com/zfoo/event/MyController2.java +++ b/event/src/test/java/com/zfoo/event/MyController2.java @@ -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()); }