natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[3.0] Support url merge processor extension (#7932)

This commit is contained in:
huazhongming 2021-06-08 17:21:01 +08:00 committed by GitHub
parent d0454fdbef
commit e189221708
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 202 additions and 111 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/ProviderURLMergeProcessor.java → dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/ProviderURLMergeProcessor.java
View File

@ -14,16 +14,26 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dubbo.rpc.cluster.support;
package org.apache.dubbo.rpc.cluster;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.SPI;
import java.util.Map;
@SPI
@SPI("default")
public interface ProviderURLMergeProcessor {
URL mergeProviderUrl(URL providerUrl, Map<String, String> localParametersMap);
boolean accept(URL providerUrl, Map<String, String> localParametersMap);
/**
* Merging the URL parameters of provider and consumer
*
* @param remoteUrl providerUrl
* @param localParametersMap consumer url parameters
* @return
*/
URL mergeUrl(URL remoteUrl, Map<String, String> localParametersMap);
default boolean accept(URL providerUrl, Map<String, String> localParametersMap) {
return true;
}
}

118
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/ClusterUtils.java
View File

@ -18,31 +18,13 @@ package org.apache.dubbo.rpc.cluster.support;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.ExtensionLoader;
import org.apache.dubbo.remoting.Constants;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.rpc.cluster.ProviderURLMergeProcessor;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import static org.apache.dubbo.common.constants.CommonConstants.ALIVE_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.APPLICATION_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.CORE_THREADS_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_KEY_PREFIX;
import static org.apache.dubbo.common.constants.CommonConstants.DUBBO_VERSION_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.GENERIC_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.INVOKER_LISTENER_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.METHODS_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.QUEUES_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.REFERENCE_FILTER_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.RELEASE_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.REMOTE_APPLICATION_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.TAG_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.THREADPOOL_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.THREADS_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.THREAD_NAME_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.TIMESTAMP_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.VERSION_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.URL_MERGE_PROCESSOR_KEY;
/**
* ClusterUtils
@ -53,93 +35,17 @@ public class ClusterUtils {
}
public static URL mergeUrl(URL remoteUrl, Map<String, String> localMap) {
Map<String, String> map = new HashMap<String, String>();
Map<String, String> remoteMap = remoteUrl.getParameters();
if (remoteMap != null && remoteMap.size() > 0) {
map.putAll(remoteMap);
String ump = localMap.get(URL_MERGE_PROCESSOR_KEY);
ProviderURLMergeProcessor providerURLMergeProcessor;
// Remove configurations from provider, some items should be affected by provider.
map.remove(THREAD_NAME_KEY);
map.remove(DEFAULT_KEY_PREFIX + THREAD_NAME_KEY);
map.remove(THREADPOOL_KEY);
map.remove(DEFAULT_KEY_PREFIX + THREADPOOL_KEY);
map.remove(CORE_THREADS_KEY);
map.remove(DEFAULT_KEY_PREFIX + CORE_THREADS_KEY);
map.remove(THREADS_KEY);
map.remove(DEFAULT_KEY_PREFIX + THREADS_KEY);
map.remove(QUEUES_KEY);
map.remove(DEFAULT_KEY_PREFIX + QUEUES_KEY);
map.remove(ALIVE_KEY);
map.remove(DEFAULT_KEY_PREFIX + ALIVE_KEY);
map.remove(Constants.TRANSPORTER_KEY);
map.remove(DEFAULT_KEY_PREFIX + Constants.TRANSPORTER_KEY);
if (StringUtils.isNotEmpty(ump)) {
providerURLMergeProcessor = ExtensionLoader.getExtensionLoader(ProviderURLMergeProcessor.class).getExtension(ump);
} else {
providerURLMergeProcessor = ExtensionLoader.getExtensionLoader(ProviderURLMergeProcessor.class).getExtension("default");
}
if (localMap != null && localMap.size() > 0) {
Map<String, String> copyOfLocalMap = new HashMap<>(localMap);
if(map.containsKey(GROUP_KEY)){
copyOfLocalMap.remove(GROUP_KEY);
}
if(map.containsKey(VERSION_KEY)){
copyOfLocalMap.remove(VERSION_KEY);
}
if (map.containsKey(GENERIC_KEY)) {
copyOfLocalMap.remove(GENERIC_KEY);
}
copyOfLocalMap.remove(RELEASE_KEY);
copyOfLocalMap.remove(DUBBO_VERSION_KEY);
copyOfLocalMap.remove(METHODS_KEY);
copyOfLocalMap.remove(TIMESTAMP_KEY);
copyOfLocalMap.remove(TAG_KEY);
map.putAll(copyOfLocalMap);
if (remoteMap != null) {
map.put(REMOTE_APPLICATION_KEY, remoteMap.get(APPLICATION_KEY));
// Combine filters and listeners on Provider and Consumer
String remoteFilter = remoteMap.get(REFERENCE_FILTER_KEY);
String localFilter = copyOfLocalMap.get(REFERENCE_FILTER_KEY);
if (remoteFilter != null && remoteFilter.length() > 0
&& localFilter != null && localFilter.length() > 0) {
map.put(REFERENCE_FILTER_KEY, remoteFilter + "," + localFilter);
}
String remoteListener = remoteMap.get(INVOKER_LISTENER_KEY);
String localListener = copyOfLocalMap.get(INVOKER_LISTENER_KEY);
if (remoteListener != null && remoteListener.length() > 0
&& localListener != null && localListener.length() > 0) {
map.put(INVOKER_LISTENER_KEY, remoteListener + "," + localListener);
}
}
}
return remoteUrl.clearParameters().addParameters(map);
return providerURLMergeProcessor.mergeUrl(remoteUrl, localMap);
}
public static URL mergeProviderUrl(URL remoteUrl, Map<String, String> localMap) {
//urlprocessor => upc
List<ProviderURLMergeProcessor> providerURLMergeProcessors = ExtensionLoader.getExtensionLoader(ProviderURLMergeProcessor.class)
.getActivateExtension(remoteUrl, "upc");
if (providerURLMergeProcessors != null && providerURLMergeProcessors.size() > 0) {
for (ProviderURLMergeProcessor providerURLMergeProcessor : providerURLMergeProcessors) {
if (providerURLMergeProcessor.accept(remoteUrl, localMap)) {
return providerURLMergeProcessor.mergeProviderUrl(remoteUrl, localMap);
}
}
}
return mergeUrl(remoteUrl, localMap);
}
}
}

123
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/merger/DefaultProviderURLMergeProcessor.java Normal file
View File

@ -0,0 +1,123 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dubbo.rpc.cluster.support.merger;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.remoting.Constants;
import org.apache.dubbo.rpc.cluster.ProviderURLMergeProcessor;
import java.util.HashMap;
import java.util.Map;
import static org.apache.dubbo.common.constants.CommonConstants.ALIVE_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.APPLICATION_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.CORE_THREADS_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_KEY_PREFIX;
import static org.apache.dubbo.common.constants.CommonConstants.DUBBO_VERSION_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.GENERIC_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.INVOKER_LISTENER_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.METHODS_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.QUEUES_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.REFERENCE_FILTER_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.RELEASE_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.REMOTE_APPLICATION_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.TAG_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.THREADPOOL_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.THREADS_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.THREAD_NAME_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.TIMESTAMP_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.VERSION_KEY;
public class DefaultProviderURLMergeProcessor implements ProviderURLMergeProcessor {
@Override
public URL mergeUrl(URL remoteUrl, Map<String, String> localParametersMap) {
Map<String, String> map = new HashMap<>();
Map<String, String> remoteMap = remoteUrl.getParameters();
if (remoteMap != null && remoteMap.size() > 0) {
map.putAll(remoteMap);
// Remove configurations from provider, some items should be affected by provider.
map.remove(THREAD_NAME_KEY);
map.remove(DEFAULT_KEY_PREFIX + THREAD_NAME_KEY);
map.remove(THREADPOOL_KEY);
map.remove(DEFAULT_KEY_PREFIX + THREADPOOL_KEY);
map.remove(CORE_THREADS_KEY);
map.remove(DEFAULT_KEY_PREFIX + CORE_THREADS_KEY);
map.remove(THREADS_KEY);
map.remove(DEFAULT_KEY_PREFIX + THREADS_KEY);
map.remove(QUEUES_KEY);
map.remove(DEFAULT_KEY_PREFIX + QUEUES_KEY);
map.remove(ALIVE_KEY);
map.remove(DEFAULT_KEY_PREFIX + ALIVE_KEY);
map.remove(Constants.TRANSPORTER_KEY);
map.remove(DEFAULT_KEY_PREFIX + Constants.TRANSPORTER_KEY);
}
if (localParametersMap != null && localParametersMap.size() > 0) {
Map<String, String> copyOfLocalMap = new HashMap<>(localParametersMap);
if (map.containsKey(GROUP_KEY)) {
copyOfLocalMap.remove(GROUP_KEY);
}
if (map.containsKey(VERSION_KEY)) {
copyOfLocalMap.remove(VERSION_KEY);
}
if (map.containsKey(GENERIC_KEY)) {
copyOfLocalMap.remove(GENERIC_KEY);
}
copyOfLocalMap.remove(RELEASE_KEY);
copyOfLocalMap.remove(DUBBO_VERSION_KEY);
copyOfLocalMap.remove(METHODS_KEY);
copyOfLocalMap.remove(TIMESTAMP_KEY);
copyOfLocalMap.remove(TAG_KEY);
map.putAll(copyOfLocalMap);
if (remoteMap != null) {
map.put(REMOTE_APPLICATION_KEY, remoteMap.get(APPLICATION_KEY));
// Combine filters and listeners on Provider and Consumer
String remoteFilter = remoteMap.get(REFERENCE_FILTER_KEY);
String localFilter = copyOfLocalMap.get(REFERENCE_FILTER_KEY);
if (remoteFilter != null && remoteFilter.length() > 0
&& localFilter != null && localFilter.length() > 0) {
map.put(REFERENCE_FILTER_KEY, remoteFilter + "," + localFilter);
}
String remoteListener = remoteMap.get(INVOKER_LISTENER_KEY);
String localListener = copyOfLocalMap.get(INVOKER_LISTENER_KEY);
if (remoteListener != null && remoteListener.length() > 0
&& localListener != null && localListener.length() > 0) {
map.put(INVOKER_LISTENER_KEY, remoteListener + "," + localListener);
}
}
}
return remoteUrl.clearParameters().addParameters(map);
}
}

1
dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.ProviderURLMergeProcessor Normal file
View File

@ -0,0 +1 @@
default=org.apache.dubbo.rpc.cluster.support.merger.DefaultProviderURLMergeProcessor

4
dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/ClusterUtilsTest.java
View File

@ -71,6 +71,7 @@ public class ClusterUtilsTest {
.addParameter(DEFAULT_KEY_PREFIX + THREAD_NAME_KEY, "test")
.addParameter(APPLICATION_KEY, "provider")
.addParameter(REFERENCE_FILTER_KEY, "filter1,filter2")
.addParameter(TAG_KEY,"TTT")
.build();
URL consumerURL = new URLBuilder(DUBBO_PROTOCOL, "localhost", 55555)
@ -78,6 +79,7 @@ public class ClusterUtilsTest {
.addParameter(THREADPOOL_KEY, "foo")
.addParameter(APPLICATION_KEY, "consumer")
.addParameter(REFERENCE_FILTER_KEY, "filter3")
.addParameter(TAG_KEY,"UUU")
.build();
URL url = ClusterUtils.mergeUrl(providerURL, consumerURL.getParameters());
@ -107,6 +109,8 @@ public class ClusterUtilsTest {
Assertions.assertEquals(url.getApplication(), "consumer");
Assertions.assertEquals(url.getRemoteApplication(), "provider");
Assertions.assertEquals(url.getParameter(REFERENCE_FILTER_KEY), "filter1,filter2,filter3");
Assertions.assertEquals(url.getParameter(TAG_KEY), "TTT");
}
@Test

5
dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java
View File

@ -417,4 +417,9 @@ public interface CommonConstants {
String CLASSPATH_URL_PREFIX = "classpath:";
String DEFAULT_VERSION = "0.0.0";
/**
* Url merge processor key
*/
String URL_MERGE_PROCESSOR_KEY = "url-merge-processor";
}

18
dubbo-common/src/main/java/org/apache/dubbo/config/ConsumerConfig.java
View File

@ -17,6 +17,9 @@
package org.apache.dubbo.config;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.config.support.Parameter;
import static org.apache.dubbo.common.constants.CommonConstants.URL_MERGE_PROCESSOR_KEY;
/**
* The service consumer default configuration
@ -58,6 +61,12 @@ public class ConsumerConfig extends AbstractReferenceConfig {
*/
private Integer shareconnections;
/**
* Url Merge Processor
* Used to customize the URL merge of consumer and provider
*/
private String urlMergeProcessor;
@Override
public void setTimeout(Integer timeout) {
super.setTimeout(timeout);
@ -115,4 +124,13 @@ public class ConsumerConfig extends AbstractReferenceConfig {
public void setShareconnections(Integer shareconnections) {
this.shareconnections = shareconnections;
}
@Parameter(key = URL_MERGE_PROCESSOR_KEY)
public String getUrlMergeProcessor() {
return urlMergeProcessor;
}
public void setUrlMergeProcessor(String urlMergeProcessor) {
this.urlMergeProcessor = urlMergeProcessor;
}
}

13
dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/bootstrap/builders/ConsumerBuilder.java
View File

@ -61,6 +61,12 @@ public class ConsumerBuilder extends AbstractReferenceBuilder<ConsumerConfig, Co
*/
private Integer shareconnections;
/**
* Url Merge Processor
* Used to customize the URL merge of consumer and provider
*/
private String urlMergeProcessor;
public ConsumerBuilder isDefault(Boolean isDefault) {
this.isDefault = isDefault;
return getThis();
@ -96,6 +102,12 @@ public class ConsumerBuilder extends AbstractReferenceBuilder<ConsumerConfig, Co
return getThis();
}
public ConsumerBuilder urlMergeProcessor(String urlMergeProcessor) {
this.urlMergeProcessor = urlMergeProcessor;
return getThis();
}
@Override
public ConsumerConfig build() {
ConsumerConfig consumer = new ConsumerConfig();
super.build(consumer);
@ -107,6 +119,7 @@ public class ConsumerBuilder extends AbstractReferenceBuilder<ConsumerConfig, Co
consumer.setThreads(threads);
consumer.setQueues(queues);
consumer.setShareconnections(shareconnections);
consumer.setUrlMergeProcessor(urlMergeProcessor);
return consumer;
}

5
dubbo-config/dubbo-config-spring/src/main/resources/META-INF/dubbo.xsd
View File

@ -1002,6 +1002,11 @@
<![CDATA[ The default share connections. default share one connection. ]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="url-merge-processor" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[ The Url merge processor. ]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:anyAttribute namespace="##other" processContents="lax"/>
</xsd:extension>
</xsd:complexContent>

6
dubbo-distribution/dubbo-all/pom.xml
View File

@ -764,6 +764,12 @@
META-INF/dubbo/internal/org.apache.dubbo.registry.client.migration.PreMigratingConditionChecker
</resource>
</transformer>
<transformer
implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>
META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.ProviderURLMergeProcessor
</resource>
</transformer>
</transformers>
<filters>
<filter>

2
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryDirectory.java
View File

@ -373,7 +373,7 @@ public class RegistryDirectory<T> extends DynamicDirectory<T> {
if (providerUrl instanceof ServiceAddressURL) {
providerUrl = overrideWithConfigurator(providerUrl);
} else {
providerUrl = ClusterUtils.mergeProviderUrl(providerUrl, queryMap); // Merge the consumer side parameters
providerUrl = ClusterUtils.mergeUrl(providerUrl, queryMap); // Merge the consumer side parameters
providerUrl = overrideWithConfigurator(providerUrl);
providerUrl = providerUrl.addParameter(Constants.CHECK_KEY, String.valueOf(false)); // Do not check whether the connection is successful or not, always create Invoker!
}