yeyinlaoying/neutrino-proxy
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ApplicationChannel消息分配逻辑优化

Browse Source
This commit is contained in:
aoshiguchen
2022-10-05 23:55:58 +08:00
parent bd17313db8
commit b845a45d06
5 changed files with 59 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
neutrino-core/src/main/java/fun/asgc/neutrino/core/base/TagMatcher.java
-35
View File
@@ -1,35 +0,0 @@
/**
* 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;
/**
* @author: aoshiguchen
* @date: 2022/10/4
*/
public interface TagMatcher {
/**
* tag匹配
* @param tag 标签
* @return 是否匹配
*/
boolean tagMatch(String tag);
}
neutrino-core/src/main/java/fun/asgc/neutrino/core/base/TopicMatcher.java
-35
View File
@@ -1,35 +0,0 @@
/**
* 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;
/**
* @author: aoshiguchen
* @date: 2022/10/4
*/
public interface TopicMatcher {
/**
* topic匹配
* @param topic topic
* @return 是否匹配
*/
boolean topicMatch(String topic);
}
neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventChannel.java
+49 -13
View File
@@ -21,11 +21,12 @@
*/
package fun.asgc.neutrino.core.base.event;
import com.google.common.collect.Sets;
import fun.asgc.neutrino.core.base.CustomThreadFactory;
import fun.asgc.neutrino.core.base.Dispatcher;
import fun.asgc.neutrino.core.util.Assert;
import fun.asgc.neutrino.core.util.CollectionUtil;
import fun.asgc.neutrino.core.util.StringUtil;
import fun.asgc.neutrino.core.web.AntPathMatcher;
import java.util.ArrayList;
import java.util.List;
@@ -42,6 +43,7 @@ public class ApplicationEventChannel<D> implements EventChannel<D,ApplicationEve
private List<ApplicationEventReceiver<D>> receiverList;
private Dispatcher<ApplicationEventContext,ApplicationEvent<D>> dispatcher;
private ThreadPoolExecutor threadPoolExecutor;
private static final AntPathMatcher antPathMatcher = new AntPathMatcher();
public ApplicationEventChannel() {
this.receiverList = new ArrayList<>();
@@ -77,20 +79,54 @@ public class ApplicationEventChannel<D> implements EventChannel<D,ApplicationEve
public void publish(ApplicationEvent<D> msg) {
this.receiverList.forEach(receiver -> {
threadPoolExecutor.submit(() -> {
if (!receiver.topicMatch(msg.context().topic())) {
return;
}
Set<String> tags = msg.context().tags();
if (CollectionUtil.isEmpty(tags)) {
tags = Sets.newHashSet("");
}
for (String tag : tags) {
if (receiver.tagMatch(tag)) {
receiver.receive(msg);
break;
}
if (match(msg, receiver)) {
receiver.receive(msg);
}
});
});
}
/**
* 判断指定消息和指定接受者是否匹配
* @param msg 消息
* @param receiver 接受者
* @return 是否匹配
*/
private boolean match(ApplicationEvent<D> msg, ApplicationEventReceiver<D> receiver) {
if (null == msg || null == receiver) {
return false;
}
return topicMatch(msg.context().topic(), receiver.getTopic()) && tagMatch(msg.context().tags(), receiver.getTags());
}
/**
* topic匹配起
* @param eventTopic 事件主题
* @param subscriptionTopic 订阅的主题
* @return 是否匹配
*/
private boolean topicMatch(String eventTopic, String subscriptionTopic) {
if (StringUtil.isEmpty(subscriptionTopic)) {
return true;
}
return antPathMatcher.match(subscriptionTopic, eventTopic == null ? "" : eventTopic);
}
/**
* 标签匹配
* @param eventTags 事件标签
* @param subscriptionTags 关注的标签
* @return 是否匹配
*/
private boolean tagMatch(Set<String> eventTags, Set<String> subscriptionTags) {
if (CollectionUtil.isEmpty(subscriptionTags)) {
return true;
}
for (String tag : eventTags) {
if (subscriptionTags.contains(tag)) {
return true;
}
}
return false;
}
}
neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventReceiver.java
+9 -24
View File
@@ -22,11 +22,6 @@
package fun.asgc.neutrino.core.base.event;
import com.alibaba.fastjson.JSONObject;
import fun.asgc.neutrino.core.base.TagMatcher;
import fun.asgc.neutrino.core.base.TopicMatcher;
import fun.asgc.neutrino.core.util.CollectionUtil;
import fun.asgc.neutrino.core.util.StringUtil;
import fun.asgc.neutrino.core.web.AntPathMatcher;
import lombok.extern.slf4j.Slf4j;
import java.util.Set;
@@ -36,27 +31,9 @@ import java.util.Set;
* @date: 2022/9/29
*/
@Slf4j
public class ApplicationEventReceiver<D> implements EventReceiver<ApplicationEventContext,D,ApplicationEvent<D>>,TagMatcher,TopicMatcher {
public class ApplicationEventReceiver<D> implements EventReceiver<ApplicationEventContext,D,ApplicationEvent<D>> {
private String topic;
private Set<String> tags;
private static final AntPathMatcher antPathMatcher = new AntPathMatcher();
@Override
public boolean tagMatch(String tag) {
if (CollectionUtil.isEmpty(this.tags)) {
return true;
}
return this.tags.contains(tag);
}
@Override
public boolean topicMatch(String topic) {
if (StringUtil.isEmpty(this.topic)) {
return true;
}
return antPathMatcher.match(this.topic, topic == null ? "" : topic);
}
@Override
public void receive(ApplicationEvent<D> msg) {
log.debug("ApplicationEventReceiver receive {}", JSONObject.toJSONString(msg));
@@ -69,4 +46,12 @@ public class ApplicationEventReceiver<D> implements EventReceiver<ApplicationEve
public void setTags(Set<String> tags) {
this.tags = tags;
}
public String getTopic() {
return topic;
}
public Set<String> getTags() {
return tags;
}
}
neutrino-core/src/test/java/fun/asgc/neutrino/core/base/event/Test1.java
+1 -1
View File
@@ -61,7 +61,7 @@ public class Test1 {
channel.registerReceiver(receiver);
ApplicationEvent<Student> event = new ApplicationEvent<>();
event.context().setId("123");
// 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("男"));