MessageAgent增加rpcFirst配置

This commit is contained in:
redkale
2023-10-17 10:20:06 +08:00
parent c3928fdc2b
commit 2155c58989
3 changed files with 17 additions and 6 deletions

View File

@@ -70,11 +70,12 @@
name: 服务的名称用于监控识别多个mq节点时只能有一个name为空的节点mq.name不能重复,命名规则: 字母、数字、下划线
type 实现类名必须是org.redkale.mq.MessageAgent的子类
threads线程数为0表示使用workExecutor。默认: CPU核数, 核数=1的情况下默认值为2JDK 21以上版本默认使用虚拟线程池
rpcfirstcluster和mq同名组件时HttpRpcClient优先使用MQ默认不优先走MQ。
coder: MessageRecord的解析器类必须是org.redkale.mq.MessageCoder<MessageRecord>的实现类,
可对数据包进行加密解密默认值org.redkale.mq.MessageRecordCoder
MQ节点下的子节点配置没有固定格式, 根据MessageAgent实现方的定义来配置
-->
<mq name="" type="org.redkalex.mq.kafka.KafkaMessageAgent" threads="4">
<mq name="" type="org.redkalex.mq.kafka.KafkaMessageAgent" rpcfirst="false" threads="4">
<servers value="127.0.0.1:9101"/>
<!--
加载所有的MessageConsumer实例;

View File

@@ -1212,11 +1212,14 @@ public final class Application {
if (this.messageAgents != null) {
MessageAgent messageAgent = this.resourceFactory.find(resourceName, MessageAgent.class);
if (messageAgent != null) {
HttpRpcClient rpcClient = messageAgent.getHttpRpcClient();
field.set(srcObj, rpcClient);
rf.inject(resourceName, rpcClient, null); // 给其可能包含@Resource的字段赋值;
rf.register(resourceName, HttpRpcClient.class, rpcClient);
return rpcClient;
if (clusterAgent == null || !Objects.equals(clusterAgent.getName(), resourceName)
|| messageAgent.isRpcFirst()) {
HttpRpcClient rpcClient = messageAgent.getHttpRpcClient();
field.set(srcObj, rpcClient);
rf.inject(resourceName, rpcClient, null); // 给其可能包含@Resource的字段赋值;
rf.register(resourceName, HttpRpcClient.class, rpcClient);
return rpcClient;
}
}
}
if (clusterAgent == null) {

View File

@@ -78,6 +78,8 @@ public abstract class MessageAgent implements Resourcable {
protected final Map<String, Map<String, MessageConsumerWrapper>> messageConsumerMap = new HashMap<>();
//-------------------------- HttpRpcClient、SncpMessageClient --------------------------
private boolean rpcFirst;
private HttpRpcMessageClient httpRpcClient;
private String httpAppRespTopic;
@@ -100,6 +102,7 @@ public abstract class MessageAgent implements Resourcable {
public void init(AnyValue config) {
this.name = checkName(config.getValue("name", ""));
this.rpcFirst = config.getBoolValue("rpcfirst", false);
this.httpAppRespTopic = generateHttpAppRespTopic();
this.sncpAppRespTopic = generateSncpAppRespTopic();
int threads = config.getIntValue("threads", application.isVirtualWorkExecutor() ? 0 : -1);
@@ -344,6 +347,10 @@ public abstract class MessageAgent implements Resourcable {
return sncpMessageClient;
}
public boolean isRpcFirst() {
return rpcFirst;
}
protected String checkName(String name) { //不能含特殊字符
if (name.isEmpty()) {
return name;