为ApplicationChannel增加channel连接功能

2022-10-07 19:09:05 +08:00
parent b845a45d06
commit 11742bb1c2
5 changed files with 194 additions and 1 deletions
@@ -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<C extends Channel> {
+    /**
+     * 连接channel
+     * @param channel channel
+     */
+    void connectChannel(C channel);
+
+    /**
+     * 断开连接channel
+     * @param channel channel
+     */
+    void disconnectChannel(C channel);
+}
@@ -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<D> implements EventChannel<D,ApplicationEventContext,ApplicationEvent<D>,ApplicationEventReceiver<D>,Dispatcher<ApplicationEventContext,ApplicationEvent<D>>> {
+public class ApplicationEventChannel<D> implements EventChannel<D,ApplicationEventContext,ApplicationEvent<D>,ApplicationEventReceiver<D>,Dispatcher<ApplicationEventContext,ApplicationEvent<D>>>, ChannelConnector<ApplicationEventChannel<D>> {
    private List<ApplicationEventReceiver<D>> receiverList;
    private Dispatcher<ApplicationEventContext,ApplicationEvent<D>> dispatcher;
    private ThreadPoolExecutor threadPoolExecutor;
    private static final AntPathMatcher antPathMatcher = new AntPathMatcher();
+    private List<ApplicationEventChannel<D>> 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<D> implements EventChannel<D,ApplicationEve

    @Override
    public void publish(ApplicationEvent<D> 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<D> implements EventChannel<D,ApplicationEve
                }
            });
        });
+        // 将消息广播到所有关联的channel中去
+        this.channelList.forEach(channel -> channel.publish(msg));
+    }
+
+    @Override
+    public void connectChannel(ApplicationEventChannel<D> channel) {
+        if (null == channel || this.channelList.contains(channel) || this == channel) {
+            return;
+        }
+        this.channelList.add(channel);
+    }
+
+    @Override
+    public void disconnectChannel(ApplicationEventChannel<D> channel) {
+        if (null == channel) {
+            return;
+        }
+        this.channelList.remove(channel);
    }

    /**
@@ -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<String, Object> attachData = new HashMap<>();
+    private List<Channel> 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 <S> void setSource(S source) {
        this.source = source;
    }
+
+    @Override
+    public List<Channel> channelList() {
+        return this.channelList;
+    }
 }
@@ -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<Channel> channelList();
 }
@@ -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<Student> channel = new ApplicationEventChannel<>();
+    private MyTestChannel myTestChannel = new MyTestChannel();
+    private ApplicationEventPublisher<Student> publisher = new ApplicationEventPublisher<>();
+
+    {
+        channel.connectChannel(myTestChannel);
+        publisher.registerChannel(channel);
+    }
+
+    @Test
+    public void test1() {
+        ApplicationEventReceiver<Student> receiver1 = new ApplicationEventReceiver<Student>() {
+            @Override
+            public void receive(ApplicationEvent<Student> msg) {
+                log.info("receiver1 data:{}", JSONObject.toJSONString(msg.data()));
+            }
+        };
+        ApplicationEventReceiver<Student> receiver2 = new ApplicationEventReceiver<Student>() {
+            @Override
+            public void receive(ApplicationEvent<Student> msg) {
+                log.info("receiver2 data:{}", JSONObject.toJSONString(msg.data()));
+            }
+        };
+//        receiver.setTopic("/**");
+//        receiver.setTags(Sets.newHashSet("create"));
+        channel.registerReceiver(receiver1);
+        myTestChannel.registerReceiver(receiver2);
+
+        ApplicationEvent<Student> 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<Student> {
+
+    }
+}