diff --git a/neutrino-core/src/main/java/fun/asgc/neutrino/core/annotation/Subscribe.java b/neutrino-core/src/main/java/fun/asgc/neutrino/core/annotation/Subscribe.java new file mode 100644 index 00000000..eb29afbb --- /dev/null +++ b/neutrino-core/src/main/java/fun/asgc/neutrino/core/annotation/Subscribe.java @@ -0,0 +1,18 @@ +package fun.asgc.neutrino.core.annotation; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * @author: aoshiguchen + * @date: 2022/10/10 + */ +@Retention(RetentionPolicy.RUNTIME) +@Target(ElementType.TYPE) +public @interface Subscribe { + boolean enable() default true; + String topic() default ""; + String[] tags() default {}; +} diff --git a/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/Publisher.java b/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/Publisher.java index d33d5dba..19eea0d1 100644 --- a/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/Publisher.java +++ b/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/Publisher.java @@ -33,14 +33,14 @@ public interface Publisher { void publish(D msg); /** - * 注册渠道 + * 绑定渠道 * @param channel 渠道 */ - void registerChannel(Ch channel); + void bindChannel(Ch channel); /** - * 注销渠道 + * 解绑渠道 * @param channel 渠道 */ - void unRegisterChannel(Ch channel); + void unbindChannel(Ch channel); } diff --git a/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventPublisher.java b/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventPublisher.java index 0f89147a..eed18869 100644 --- a/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventPublisher.java +++ b/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventPublisher.java @@ -44,7 +44,7 @@ public class ApplicationEventPublisher implements EventPublisher channel) { + public void bindChannel(ApplicationEventChannel channel) { Assert.notNull(channel, "channel不能为空!"); if (!this.channelList.contains(channel)) { this.channelList.add(channel); @@ -52,7 +52,7 @@ public class ApplicationEventPublisher implements EventPublisher channel) { + public void unbindChannel(ApplicationEventChannel channel) { Assert.notNull(channel, "channel不能为空!"); this.channelList.remove(channel); } diff --git a/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventReceiver.java b/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventReceiver.java index 312f1085..47028e3d 100644 --- a/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventReceiver.java +++ b/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventReceiver.java @@ -34,6 +34,20 @@ import java.util.Set; public class ApplicationEventReceiver implements EventReceiver> { private String topic; private Set tags; + + public ApplicationEventReceiver() { + + } + + public ApplicationEventReceiver(String topic) { + this.topic = topic; + } + + public ApplicationEventReceiver(String topic, Set tags) { + this.topic = topic; + this.tags = tags; + } + @Override public void receive(ApplicationEvent msg) { log.debug("ApplicationEventReceiver receive {}", JSONObject.toJSONString(msg)); diff --git a/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/SimpleApplicationEventManager.java b/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/SimpleApplicationEventManager.java new file mode 100644 index 00000000..02689313 --- /dev/null +++ b/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/SimpleApplicationEventManager.java @@ -0,0 +1,81 @@ +/** + * Copyright (c) 2022 aoshiguchen + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ +package fun.asgc.neutrino.core.base.event; + +import java.util.Set; + +/** + * 简单的应用事件管理器 + * @author: wen.u + * @date: 2022/10/10 + */ +public class SimpleApplicationEventManager { + private ApplicationEventChannel channel; + private ApplicationEventPublisher publisher; + private Object source; + + public SimpleApplicationEventManager() { + this.channel = new ApplicationEventChannel<>(); + this.publisher = new ApplicationEventPublisher<>(); + this.publisher.bindChannel(channel); + } + + public SimpleApplicationEventManager(Object source) { + this.channel = new ApplicationEventChannel<>(); + this.publisher = new ApplicationEventPublisher<>(); + this.publisher.bindChannel(channel); + this.source = source; + } + + public void publish(D data) { + this.publish(null, null, data); + } + + public void publish(String topic, D data) { + this.publish(topic, null, data); + } + + public void publish(String topic, Set tags, D data) { + ApplicationEvent event = new ApplicationEvent<>(); + event.setData(data); + event.context().setSource(source); + event.context().setTopic(topic); + event.context().setTags(tags); + this.publisher.publish(event); + } + + public void registerReceiver(ApplicationEventReceiver receiver) { + this.channel.registerReceiver(receiver); + } + + public ApplicationEventChannel getChannel() { + return channel; + } + + public ApplicationEventPublisher getPublisher() { + return publisher; + } + + public Object getSource() { + return source; + } +} diff --git a/neutrino-core/src/main/java/fun/asgc/neutrino/core/bean/BeanWrapper.java b/neutrino-core/src/main/java/fun/asgc/neutrino/core/bean/BeanWrapper.java index a7db4748..1163377f 100644 --- a/neutrino-core/src/main/java/fun/asgc/neutrino/core/bean/BeanWrapper.java +++ b/neutrino-core/src/main/java/fun/asgc/neutrino/core/bean/BeanWrapper.java @@ -25,7 +25,10 @@ package fun.asgc.neutrino.core.bean; import fun.asgc.neutrino.core.annotation.*; import fun.asgc.neutrino.core.context.ApplicationRunner; import fun.asgc.neutrino.core.context.LifeCycle; -import fun.asgc.neutrino.core.util.*; +import fun.asgc.neutrino.core.util.ClassUtil; +import fun.asgc.neutrino.core.util.CollectionUtil; +import fun.asgc.neutrino.core.util.LockUtil; +import fun.asgc.neutrino.core.util.ReflectUtil; import lombok.Data; import lombok.experimental.Accessors; import lombok.extern.slf4j.Slf4j; diff --git a/neutrino-core/src/main/java/fun/asgc/neutrino/core/bean/factory/AbstractBeanFactory.java b/neutrino-core/src/main/java/fun/asgc/neutrino/core/bean/factory/AbstractBeanFactory.java index 29f10d19..857b5676 100644 --- a/neutrino-core/src/main/java/fun/asgc/neutrino/core/bean/factory/AbstractBeanFactory.java +++ b/neutrino-core/src/main/java/fun/asgc/neutrino/core/bean/factory/AbstractBeanFactory.java @@ -21,7 +21,10 @@ */ package fun.asgc.neutrino.core.bean.factory; +import fun.asgc.neutrino.core.annotation.Subscribe; import fun.asgc.neutrino.core.base.CustomThreadFactory; +import fun.asgc.neutrino.core.base.event.ApplicationEventReceiver; +import fun.asgc.neutrino.core.base.event.SimpleApplicationEventManager; import fun.asgc.neutrino.core.bean.*; import fun.asgc.neutrino.core.context.Environment; import fun.asgc.neutrino.core.context.LifeCycle; @@ -31,12 +34,13 @@ import fun.asgc.neutrino.core.exception.BeanException; import fun.asgc.neutrino.core.util.*; import lombok.extern.slf4j.Slf4j; -import java.util.Comparator; -import java.util.LinkedList; -import java.util.List; -import java.util.Map; -import java.util.concurrent.*; +import java.util.*; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; +import java.util.stream.Stream; /** * 抽象的bean工厂 @@ -371,7 +375,32 @@ public abstract class AbstractBeanFactory implements BeanFactory, BeanRegistry, parent.init(); } if (CollectionUtil.notEmpty(beanCache)) { - beanCache.values().stream().filter(b -> BeanStatus.INJECT == b.getStatus()).forEach(bean -> bean.init()); + beanCache.values().stream().forEach(bean -> { + SimpleApplicationEventManager defaultApplicationEventManager = getEnvironment().getDefaultApplicationEventManager(); + if (null != defaultApplicationEventManager && ApplicationEventReceiver.class.isAssignableFrom(bean.getType())) { + boolean enable = true; + String topic = null; + Set tags = null; + Subscribe subscribe = bean.getType().getAnnotation(Subscribe.class); + if (null != subscribe) { + enable = subscribe.enable(); + topic = subscribe.topic(); + if (ArrayUtil.notEmpty(subscribe.tags())) { + tags = Stream.of(subscribe.tags()).collect(Collectors.toSet()); + } + } + if (enable) { + ApplicationEventReceiver receiver = (ApplicationEventReceiver) bean.getInstance(); + receiver.setTopic(topic); + receiver.setTags(tags); + defaultApplicationEventManager.registerReceiver(receiver); + } + } + + if (BeanStatus.INJECT == bean.getStatus()) { + bean.init(); + } + }); } log.info("bean工厂[{}]初始化.", getName()); scheduledExecutor.scheduleWithFixedDelay(this::run, 0, 1, TimeUnit.SECONDS); diff --git a/neutrino-core/src/main/java/fun/asgc/neutrino/core/bean/factory/SimpleBeanFactory.java b/neutrino-core/src/main/java/fun/asgc/neutrino/core/bean/factory/SimpleBeanFactory.java index 17f4b009..8d241285 100644 --- a/neutrino-core/src/main/java/fun/asgc/neutrino/core/bean/factory/SimpleBeanFactory.java +++ b/neutrino-core/src/main/java/fun/asgc/neutrino/core/bean/factory/SimpleBeanFactory.java @@ -28,6 +28,7 @@ import fun.asgc.neutrino.core.aop.interceptor.ExceptionHandler; import fun.asgc.neutrino.core.aop.interceptor.Filter; import fun.asgc.neutrino.core.aop.interceptor.Interceptor; import fun.asgc.neutrino.core.aop.interceptor.ResultAdvice; +import fun.asgc.neutrino.core.base.event.ApplicationEventReceiver; import fun.asgc.neutrino.core.bean.*; import fun.asgc.neutrino.core.exception.BeanException; import fun.asgc.neutrino.core.context.ApplicationRunner; @@ -374,6 +375,7 @@ public class SimpleBeanFactory extends AbstractBeanFactory { || Filter.class.isAssignableFrom(item) || ExceptionHandler.class.isAssignableFrom(item) || ResultAdvice.class.isAssignableFrom(item) + || ApplicationEventReceiver.class.isAssignableFrom(item) ) .forEach(clazz -> { String beanName = TypeUtil.getDefaultVariableName(clazz); diff --git a/neutrino-core/src/main/java/fun/asgc/neutrino/core/constant/MetaDataConstant.java b/neutrino-core/src/main/java/fun/asgc/neutrino/core/constant/MetaDataConstant.java index 948bab82..f61c7b27 100644 --- a/neutrino-core/src/main/java/fun/asgc/neutrino/core/constant/MetaDataConstant.java +++ b/neutrino-core/src/main/java/fun/asgc/neutrino/core/constant/MetaDataConstant.java @@ -80,4 +80,8 @@ public interface MetaDataConstant { * 服务版本 */ String SERVER_VS = "Neutrino-1.0"; + /** + * app生命周期主题 + */ + String TOPIC_APP_LIFE_CYCLE = "TP_APP_LIFE_CYCLE"; } diff --git a/neutrino-core/src/main/java/fun/asgc/neutrino/core/context/Environment.java b/neutrino-core/src/main/java/fun/asgc/neutrino/core/context/Environment.java index 81dd8e65..8776c264 100644 --- a/neutrino-core/src/main/java/fun/asgc/neutrino/core/context/Environment.java +++ b/neutrino-core/src/main/java/fun/asgc/neutrino/core/context/Environment.java @@ -22,6 +22,7 @@ package fun.asgc.neutrino.core.context; +import fun.asgc.neutrino.core.base.event.SimpleApplicationEventManager; import fun.asgc.neutrino.core.util.SystemUtil; import lombok.Data; import lombok.experimental.Accessors; @@ -64,4 +65,8 @@ public class Environment { * 运行上下文 */ private SystemUtil.RunContext runContext; + /** + * 默认的应用事件管理器 + */ + private SimpleApplicationEventManager defaultApplicationEventManager; } diff --git a/neutrino-core/src/main/java/fun/asgc/neutrino/core/context/NeutrinoLauncher.java b/neutrino-core/src/main/java/fun/asgc/neutrino/core/context/NeutrinoLauncher.java index 0a37e1dc..ba35e6bb 100644 --- a/neutrino-core/src/main/java/fun/asgc/neutrino/core/context/NeutrinoLauncher.java +++ b/neutrino-core/src/main/java/fun/asgc/neutrino/core/context/NeutrinoLauncher.java @@ -25,6 +25,8 @@ package fun.asgc.neutrino.core.context; import com.google.common.collect.Lists; import fun.asgc.neutrino.core.annotation.EnableJob; import fun.asgc.neutrino.core.annotation.NeutrinoApplication; +import fun.asgc.neutrino.core.base.event.SimpleApplicationEventManager; +import fun.asgc.neutrino.core.constant.AppLifeCycleStatusEnum; import fun.asgc.neutrino.core.constant.MetaDataConstant; import fun.asgc.neutrino.core.util.*; import lombok.extern.slf4j.Slf4j; @@ -52,26 +54,44 @@ public class NeutrinoLauncher { private NeutrinoLauncher(Class clazz, String[] args) { this.environment = new Environment() .setMainClass(clazz) - .setMainArgs(args); + .setMainArgs(args) + .setDefaultApplicationEventManager(new SimpleApplicationEventManager(this)) + ; } private SystemUtil.RunContext launch() { + // 已创建 + publishAppLifeCycleEvent(AppLifeCycleStatusEnum.APP_CREATE); + StopWatch stopWatch = new StopWatch(); stopWatch.start(); - environmentInit(); + + // 应用已初始化 + publishAppLifeCycleEvent(AppLifeCycleStatusEnum.APP_INIT); + ApplicationContext context = new ApplicationContext(environment); SystemUtil.RunContext runContext = SystemUtil.waitProcessDestroy(() -> { + // 应用准备销毁 + publishAppLifeCycleEvent(AppLifeCycleStatusEnum.APP_PRE_DESTROY); context.destroy(); log.info("Application already stop."); + // 应用已销毁 + publishAppLifeCycleEvent(AppLifeCycleStatusEnum.APP_DESTROY); }); environment.setRunContext(runContext); context.run(); + // 容器已初始化 + publishAppLifeCycleEvent(AppLifeCycleStatusEnum.CONTAINER_INIT); + stopWatch.stop(); printLog(environment, stopWatch); + // 应用已启动完成 + publishAppLifeCycleEvent(AppLifeCycleStatusEnum.APP_STARTUP); + return runContext; } @@ -133,4 +153,12 @@ public class NeutrinoLauncher { } log.info(environment.getBanner()); } + + /** + * 发布应用生命周期事件 + * @param appLifeCycleStatusEnum 应用生命周期事件 + */ + private void publishAppLifeCycleEvent(AppLifeCycleStatusEnum appLifeCycleStatusEnum) { + this.environment.getDefaultApplicationEventManager().publish(MetaDataConstant.TOPIC_APP_LIFE_CYCLE, appLifeCycleStatusEnum); + } } diff --git a/neutrino-core/src/test/java/fun/asgc/neutrino/core/base/event/Test1.java b/neutrino-core/src/test/java/fun/asgc/neutrino/core/base/event/Test1.java index cd5fa545..248951f1 100644 --- a/neutrino-core/src/test/java/fun/asgc/neutrino/core/base/event/Test1.java +++ b/neutrino-core/src/test/java/fun/asgc/neutrino/core/base/event/Test1.java @@ -45,7 +45,7 @@ public class Test1 { private ApplicationEventPublisher publisher = new ApplicationEventPublisher<>(); { - publisher.registerChannel(channel); + publisher.bindChannel(channel); } @Test diff --git a/neutrino-core/src/test/java/fun/asgc/neutrino/core/base/event/Test2.java b/neutrino-core/src/test/java/fun/asgc/neutrino/core/base/event/Test2.java index ac9c70f1..d93e8d68 100644 --- a/neutrino-core/src/test/java/fun/asgc/neutrino/core/base/event/Test2.java +++ b/neutrino-core/src/test/java/fun/asgc/neutrino/core/base/event/Test2.java @@ -47,7 +47,7 @@ public class Test2 { { channel.connectChannel(myTestChannel); - publisher.registerChannel(channel); + publisher.bindChannel(channel); } @Test diff --git a/neutrino-core/src/test/java/fun/asgc/neutrino/core/base/event/Test3.java b/neutrino-core/src/test/java/fun/asgc/neutrino/core/base/event/Test3.java new file mode 100644 index 00000000..a4ca0477 --- /dev/null +++ b/neutrino-core/src/test/java/fun/asgc/neutrino/core/base/event/Test3.java @@ -0,0 +1,80 @@ +/** + * Copyright (c) 2022 aoshiguchen + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ +package fun.asgc.neutrino.core.base.event; + +import com.alibaba.fastjson.JSONObject; +import fun.asgc.neutrino.core.util.SystemUtil; +import lombok.Data; +import lombok.experimental.Accessors; +import lombok.extern.slf4j.Slf4j; +import org.junit.Test; + +/** + * 应用事件测试 + * 1、异步执行 + * 2、业务解耦 + * 3、topic订阅 + * 4、支持多种模式无缝切换（本地模式、redis模式、rocketMQ模式、MQTT模式等） + * 5、不支持事务消息 + * @author: aoshiguchen + * @date: 2022/10/3 + */ +@Slf4j +public class Test3 { + private SimpleApplicationEventManager simpleApplicationEventManager = new SimpleApplicationEventManager<>(this); + + @Test + public void test1() { + ApplicationEventReceiver receiver1 = new ApplicationEventReceiver() { + @Override + public void receive(ApplicationEvent msg) { + log.info("receiver1 data:{}", JSONObject.toJSONString(msg.data())); + } + }; + receiver1.setTopic("/*"); + ApplicationEventReceiver receiver2 = new ApplicationEventReceiver() { + @Override + public void receive(ApplicationEvent msg) { + log.info("receiver2 data:{}", JSONObject.toJSONString(msg.data())); + } + }; + simpleApplicationEventManager.registerReceiver(receiver1); + simpleApplicationEventManager.registerReceiver(receiver2); + + simpleApplicationEventManager.publish("/aaa", null, new Student().setId("1").setName("张三").setAge(28).setSex("男")); + + SystemUtil.waitProcessDestroy().sync(); + } + + @Accessors(chain = true) + @Data + public static class Student { + private String id; + private String name; + private Integer age; + private String sex; + } + + public static class MyTestChannel extends ApplicationEventChannel { + + } +} diff --git a/neutrino-proxy-client/src/main/java/fun/asgc/neutrino/proxy/client/core/ApplicationLifeCycleListener.java b/neutrino-proxy-client/src/main/java/fun/asgc/neutrino/proxy/client/core/ApplicationLifeCycleListener.java new file mode 100644 index 00000000..5d830929 --- /dev/null +++ b/neutrino-proxy-client/src/main/java/fun/asgc/neutrino/proxy/client/core/ApplicationLifeCycleListener.java @@ -0,0 +1,41 @@ +/** + * Copyright (c) 2022 aoshiguchen + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ +package fun.asgc.neutrino.proxy.client.core; + +import fun.asgc.neutrino.core.base.event.ApplicationEvent; +import fun.asgc.neutrino.core.base.event.ApplicationEventReceiver; +import fun.asgc.neutrino.core.constant.AppLifeCycleStatusEnum; +import lombok.extern.slf4j.Slf4j; + +/** + * @author: aoshiguchen + * @date: 2022/10/10 + */ +@Slf4j +public class ApplicationLifeCycleListener extends ApplicationEventReceiver { + + @Override + public void receive(ApplicationEvent msg) { + log.info("ApplicationLifeCycleListener:{}", msg.data().getDesc()); + } + +}