diff --git a/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/ChannelConnector.java b/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/ChannelConnector.java new file mode 100644 index 00000000..e1068166 --- /dev/null +++ b/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/ChannelConnector.java @@ -0,0 +1,54 @@ +/** + * 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; + +/** + * 该接口用于将多个channel连接，实现消息广播 + * 应用场景： + * 项目初期，为了快速实现功能，不考虑多节点，可能不会接入太多的第三方依赖，如：redis、rocketMQ等。 + * 但是要保证后期需要的时候能够快速接入，而不需要对现有逻辑做太大的改动。 + * + * 那么，前期你可以使用ApplicationEventChannel，来做业务解藕。 + * 当多次迭代后需要引入RocketMQ，则直接让ApplicationEventChannel连接RocketMQChannel，从而大大减少开发工作。 + * + * 该方案适用于异步、解耦、削峰，不适用与事务消息、延时消息. + * + * 需要注意的是，该连接是单向连接。 + * 如：A连接B，则经过A的消息会广播给B，而经过B的消息不会广播给A + * 如果需要，B也需要实现该接口，并且连接A + * + * @author: aoshiguchen + * @date: 2022/10/7 + */ +public interface ChannelConnector { + /** + * 连接channel + * @param channel channel + */ + void connectChannel(C channel); + + /** + * 断开连接channel + * @param channel channel + */ + void disconnectChannel(C channel); +} diff --git a/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventChannel.java b/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventChannel.java index 50e0b446..d9284db3 100644 --- a/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventChannel.java +++ b/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventChannel.java @@ -21,6 +21,7 @@ */ package fun.asgc.neutrino.core.base.event; +import fun.asgc.neutrino.core.base.ChannelConnector; import fun.asgc.neutrino.core.base.CustomThreadFactory; import fun.asgc.neutrino.core.base.Dispatcher; import fun.asgc.neutrino.core.util.Assert; @@ -39,14 +40,16 @@ import java.util.concurrent.TimeUnit; * @author: aoshiguchen * @date: 2022/9/29 */ -public class ApplicationEventChannel implements EventChannel,ApplicationEventReceiver,Dispatcher>> { +public class ApplicationEventChannel implements EventChannel,ApplicationEventReceiver,Dispatcher>>, ChannelConnector> { private List> receiverList; private Dispatcher> dispatcher; private ThreadPoolExecutor threadPoolExecutor; private static final AntPathMatcher antPathMatcher = new AntPathMatcher(); + private List> channelList; public ApplicationEventChannel() { this.receiverList = new ArrayList<>(); + this.channelList = new ArrayList<>(); this.threadPoolExecutor = new ThreadPoolExecutor(5, 20, 10L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), new CustomThreadFactory("ApplicationEventChannel")); } @@ -77,6 +80,11 @@ public class ApplicationEventChannel implements EventChannel msg) { + if (msg.context().channelList().contains(this)) { + return; + } + msg.context().channelList().add(this); + // 将消息推送给关注该channel的接受者 this.receiverList.forEach(receiver -> { threadPoolExecutor.submit(() -> { if (match(msg, receiver)) { @@ -84,6 +92,24 @@ public class ApplicationEventChannel implements EventChannel channel.publish(msg)); + } + + @Override + public void connectChannel(ApplicationEventChannel channel) { + if (null == channel || this.channelList.contains(channel) || this == channel) { + return; + } + this.channelList.add(channel); + } + + @Override + public void disconnectChannel(ApplicationEventChannel channel) { + if (null == channel) { + return; + } + this.channelList.remove(channel); } /** diff --git a/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventContext.java b/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventContext.java index ff2bb511..617a5206 100644 --- a/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventContext.java +++ b/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventContext.java @@ -21,6 +21,7 @@ */ package fun.asgc.neutrino.core.base.event; +import fun.asgc.neutrino.core.base.Channel; import fun.asgc.neutrino.core.util.StringUtil; import java.util.*; @@ -37,10 +38,12 @@ public class ApplicationEventContext implements EventContext { private Object source; private Date happenTime; private Map attachData = new HashMap<>(); + private List channelList; public ApplicationEventContext() { this.id = StringUtil.genUUID(); this.happenTime = new Date(); + this.channelList = new ArrayList<>(); } @Override @@ -88,4 +91,9 @@ public class ApplicationEventContext implements EventContext { public void setSource(S source) { this.source = source; } + + @Override + public List channelList() { + return this.channelList; + } } diff --git a/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/EventContext.java b/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/EventContext.java index 195341e8..09cefb44 100644 --- a/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/EventContext.java +++ b/neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/EventContext.java @@ -21,7 +21,10 @@ */ package fun.asgc.neutrino.core.base.event; +import fun.asgc.neutrino.core.base.Channel; + import java.util.Date; +import java.util.List; import java.util.Map; import java.util.Set; @@ -65,4 +68,12 @@ public interface EventContext { * @return 事件发生的时间 */ Date happenTime(); + + /** + * channel列表 + * 1、如果为空，说明该事件未经过channel + * 2、该list代表事件在channel中的广播顺序 + * @return channel列表 + */ + List channelList(); } 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 new file mode 100644 index 00000000..ac9c70f1 --- /dev/null +++ b/neutrino-core/src/test/java/fun/asgc/neutrino/core/base/event/Test2.java @@ -0,0 +1,94 @@ +/** + * 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 com.google.common.collect.Sets; +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 Test2 { + private ApplicationEventChannel channel = new ApplicationEventChannel<>(); + private MyTestChannel myTestChannel = new MyTestChannel(); + private ApplicationEventPublisher publisher = new ApplicationEventPublisher<>(); + + { + channel.connectChannel(myTestChannel); + publisher.registerChannel(channel); + } + + @Test + public void test1() { + ApplicationEventReceiver receiver1 = new ApplicationEventReceiver() { + @Override + public void receive(ApplicationEvent msg) { + log.info("receiver1 data:{}", JSONObject.toJSONString(msg.data())); + } + }; + ApplicationEventReceiver receiver2 = new ApplicationEventReceiver() { + @Override + public void receive(ApplicationEvent msg) { + log.info("receiver2 data:{}", JSONObject.toJSONString(msg.data())); + } + }; +// receiver.setTopic("/**"); +// receiver.setTags(Sets.newHashSet("create")); + channel.registerReceiver(receiver1); + myTestChannel.registerReceiver(receiver2); + + ApplicationEvent event = new ApplicationEvent<>(); +// event.context().setId("123"); + event.context().setTopic("/student/create"); + event.context().setTags(Sets.newHashSet("create")); + event.setData(new Student().setId("1").setName("张三").setAge(28).setSex("男")); + publisher.publish(event); + + 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 { + + } +}