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

Add default Dynamic Param Key (#8028) 

* Add default Dynamic Param Key

* add ASF license header

* remove debug output

* Fix param being override

* skip null value

* introduce a serializable URLParam

* perfect comment

* add serialVersionUID
This commit is contained in:
Albumen Kevin 2021-06-11 22:19:42 +08:00 committed by GitHub
parent 4844effcb6
commit 6e9dbe9b43
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 233 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
dubbo-common/src/main/java/org/apache/dubbo/common/URL.java
View File

@ -23,6 +23,7 @@ import org.apache.dubbo.common.url.component.PathURLAddress;
import org.apache.dubbo.common.url.component.ServiceConfigURL;
import org.apache.dubbo.common.url.component.URLAddress;
import org.apache.dubbo.common.url.component.URLParam;
import org.apache.dubbo.common.url.component.URLPlainParam;
import org.apache.dubbo.common.utils.ArrayUtils;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.LRUCache;
@ -1831,4 +1832,7 @@ class URL implements Serializable {
return getMethodNumbers();
}
public URL toSerializableURL() {
return returnURL(URLPlainParam.toURLPlainParam(urlParam));
}
}

118
dubbo-common/src/main/java/org/apache/dubbo/common/url/component/URLParam.java
View File

@ -22,7 +22,6 @@ import org.apache.dubbo.common.url.component.param.DynamicParamTable;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.StringUtils;
import java.io.Serializable;
import java.util.Arrays;
import java.util.BitSet;
import java.util.Collection;
@ -49,11 +48,14 @@ import static org.apache.dubbo.common.constants.CommonConstants.TIMESTAMP_KEY;
* is not required if the real value is the default value.
* <br/>
* URLParam should operate as Copy-On-Write, each modify actions will return a new Object
* <br/>
*
* NOTE: URLParam is not support serialization! {@link DynamicParamTable} is related with
* current running environment. If you want to make URL as a parameter, please call
* {@link URL#toSerializableURL()} to create {@link URLPlainParam} instead.
* @since 3.0
*/
public class URLParam implements Serializable {
private static final long serialVersionUID = -1985165475234910535L;
public class URLParam {
/**
* Maximum size of key-pairs requested using array moving to add into URLParam.
@ -103,18 +105,24 @@ public class URLParam implements Serializable {
private transient long timestamp;
/**
* Whether to enable DynamicParamTable compression
*/
protected boolean enableCompressed;
private final static URLParam EMPTY_PARAM = new URLParam(new BitSet(0), Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap(), "");
private URLParam() {
protected URLParam() {
this.rawParam = null;
this.KEY = null;
this.DEFAULT_KEY = null;
this.VALUE = null;
this.EXTRA_PARAMS = null;
this.METHOD_PARAMETERS = null;
this.enableCompressed = true;
}
private URLParam(BitSet key, Map<Integer, Integer> value, Map<String, String> extraParams, Map<String, Map<String, String>> methodParameters, String rawParam) {
protected URLParam(BitSet key, Map<Integer, Integer> value, Map<String, String> extraParams, Map<String, Map<String, String>> methodParameters, String rawParam) {
this.KEY = key;
this.DEFAULT_KEY = new BitSet(KEY.size());
this.VALUE = new Integer[value.size()];
@ -133,9 +141,10 @@ public class URLParam implements Serializable {
this.rawParam = rawParam;
this.timestamp = System.currentTimeMillis();
this.enableCompressed = true;
}
private URLParam(BitSet key, BitSet defaultKey, Integer[] value, Map<String, String> extraParams, Map<String, Map<String, String>> methodParameters, String rawParam) {
protected URLParam(BitSet key, BitSet defaultKey, Integer[] value, Map<String, String> extraParams, Map<String, Map<String, String>> methodParameters, String rawParam) {
this.KEY = key;
this.DEFAULT_KEY = defaultKey;
@ -146,6 +155,7 @@ public class URLParam implements Serializable {
this.rawParam = rawParam;
this.timestamp = System.currentTimeMillis();
this.enableCompressed = true;
}
/**
@ -559,7 +569,10 @@ public class URLParam implements Serializable {
if (skipIfPresent && hasParameter(entry.getKey())) {
continue;
}
Integer keyIndex = DynamicParamTable.getKeyIndex(entry.getKey());
if (entry.getKey() == null || entry.getValue() == null) {
continue;
}
Integer keyIndex = DynamicParamTable.getKeyIndex(enableCompressed, entry.getKey());
if (keyIndex == null) {
// entry key is not present in DynamicParamTable, add it to EXTRA_PARAMS
if (newExtraParams == null) {
@ -575,26 +588,58 @@ public class URLParam implements Serializable {
methodMap.put(methodSplit[0], entry.getValue());
}
} else {
if (newKey == null) {
newKey = (BitSet) KEY.clone();
}
newKey.set(keyIndex);
if (parameters.size() > ADD_PARAMETER_ON_MOVE_THRESHOLD) {
// recover VALUE back to Map
if (newValueMap == null) {
newValueMap = recoverCompressedValue();
if (KEY.get(keyIndex)) {
// contains key, replace value
if (parameters.size() > ADD_PARAMETER_ON_MOVE_THRESHOLD) {
// recover VALUE back to Map, use map to replace key pair
if (newValueMap == null) {
newValueMap = recoverCompressedValue();
}
newValueMap.put(keyIndex, DynamicParamTable.getValueIndex(entry.getKey(), entry.getValue()));
} else if (!DynamicParamTable.isDefaultValue(entry.getKey(), entry.getValue())) {
// new value is not the default key
if (DEFAULT_KEY.get(keyIndex)) {
// old value is the default value
// value is default value, add to defaultKey directly
if (defaultKey == null) {
defaultKey = (BitSet) DEFAULT_KEY.clone();
}
defaultKey.set(keyIndex, false);
newValueArray = addByMove(VALUE, keyIndexToCompressIndex(KEY, DEFAULT_KEY, keyIndex), DynamicParamTable.getValueIndex(entry.getKey(), entry.getValue()));
} else {
// old value is not the default key, replace offset in VALUE array
newValueArray = replaceOffset(VALUE, keyIndexToCompressIndex(KEY, DEFAULT_KEY, keyIndex), DynamicParamTable.getValueIndex(entry.getKey(), entry.getValue()));
}
} else {
// value is default value, add to defaultKey directly
if (defaultKey == null) {
defaultKey = (BitSet) DEFAULT_KEY.clone();
}
defaultKey.set(keyIndex);
}
newValueMap.put(keyIndex, DynamicParamTable.getValueIndex(entry.getKey(), entry.getValue()));
} else if (!DynamicParamTable.isDefaultValue(entry.getKey(), entry.getValue())) {
// add parameter by moving array, only support for adding once
newValueArray = addByMove(VALUE, keyIndexToCompressIndex(newKey, DEFAULT_KEY, keyIndex), DynamicParamTable.getValueIndex(entry.getKey(), entry.getValue()));
} else {
// value is default value, add to defaultKey directly
if (defaultKey == null) {
defaultKey = (BitSet) DEFAULT_KEY.clone();
// key is absent, add it
if (newKey == null) {
newKey = (BitSet) KEY.clone();
}
newKey.set(keyIndex);
if (parameters.size() > ADD_PARAMETER_ON_MOVE_THRESHOLD) {
// recover VALUE back to Map
if (newValueMap == null) {
newValueMap = recoverCompressedValue();
}
newValueMap.put(keyIndex, DynamicParamTable.getValueIndex(entry.getKey(), entry.getValue()));
} else if (!DynamicParamTable.isDefaultValue(entry.getKey(), entry.getValue())) {
// add parameter by moving array, only support for adding once
newValueArray = addByMove(VALUE, keyIndexToCompressIndex(newKey, DEFAULT_KEY, keyIndex), DynamicParamTable.getValueIndex(entry.getKey(), entry.getValue()));
} else {
// value is default value, add to defaultKey directly
if (defaultKey == null) {
defaultKey = (BitSet) DEFAULT_KEY.clone();
}
defaultKey.set(keyIndex);
}
defaultKey.set(keyIndex);
}
}
}
@ -644,6 +689,19 @@ public class URLParam implements Serializable {
return result;
}
private Integer[] replaceOffset(Integer[] array, int index, Integer value) {
if (index < 0 || index > array.length) {
throw new IllegalArgumentException();
}
// copy-on-write
Integer[] result = new Integer[array.length];
System.arraycopy(array, 0, result, 0, array.length);
result[index] = value;
return result;
}
/**
* remove specified parameters in URLParam
*
@ -661,7 +719,7 @@ public class URLParam implements Serializable {
Map<String, String> newExtraParams = null;
Map<String, Map<String, String>> newMethodParams = null;
for (String key : keys) {
Integer keyIndex = DynamicParamTable.getKeyIndex(key);
Integer keyIndex = DynamicParamTable.getKeyIndex(enableCompressed, key);
if (keyIndex != null && KEY.get(keyIndex)) {
if (newKey == null) {
newKey = (BitSet) KEY.clone();
@ -765,7 +823,7 @@ public class URLParam implements Serializable {
* @return present or not
*/
public boolean hasParameter(String key) {
Integer keyIndex = DynamicParamTable.getKeyIndex(key);
Integer keyIndex = DynamicParamTable.getKeyIndex(enableCompressed, key);
if (keyIndex == null) {
return EXTRA_PARAMS.containsKey(key);
}
@ -779,7 +837,7 @@ public class URLParam implements Serializable {
* @return value, null if key is absent
*/
public String getParameter(String key) {
Integer keyIndex = DynamicParamTable.getKeyIndex(key);
Integer keyIndex = DynamicParamTable.getKeyIndex(enableCompressed, key);
if (keyIndex == null) {
if (EXTRA_PARAMS.containsKey(key)) {
return EXTRA_PARAMS.get(key);
@ -837,6 +895,10 @@ public class URLParam implements Serializable {
}
}
protected Map<String, Map<String, String>> getMethodParameters() {
return METHOD_PARAMETERS;
}
public long getTimestamp() {
return timestamp;
}
@ -1018,7 +1080,7 @@ public class URLParam implements Serializable {
private static void addParameter(BitSet keyBit, Map<Integer, Integer> valueMap, Map<String, String> extraParam,
Map<String, Map<String, String>> methodParameters, String key, String value, boolean skipIfPresent) {
Integer keyIndex = DynamicParamTable.getKeyIndex(key);
Integer keyIndex = DynamicParamTable.getKeyIndex(true, key);
if (skipIfPresent) {
if (keyIndex == null) {
if (extraParam.containsKey(key)) {

44
dubbo-common/src/main/java/org/apache/dubbo/common/url/component/URLPlainParam.java Normal file
View File

@ -0,0 +1,44 @@
/*
* 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.common.url.component;
import java.io.Serializable;
import java.util.BitSet;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
/**
* Act like URLParam, will not use DynamicParamTable to compress parameters,
* which can support serializer serialization and deserialization.
* DynamicParamTable is environment hard related.
*/
public class URLPlainParam extends URLParam implements Serializable {
private static final long serialVersionUID = 4722019979665434393L;
protected URLPlainParam(BitSet key, BitSet defaultKey, Integer[] value, Map<String, String> extraParams, Map<String, Map<String, String>> methodParameters, String rawParam) {
super(key, defaultKey, value, extraParams, methodParameters, rawParam);
this.enableCompressed = false;
}
public static URLPlainParam toURLPlainParam(URLParam urlParam) {
Map<String, String> params = Collections.unmodifiableMap(new HashMap<>(urlParam.getParameters()));
return new URLPlainParam(new BitSet(), new BitSet(), new Integer[0], params, urlParam.getMethodParameters(), urlParam.getRawParam());
}
}

65
dubbo-common/src/main/java/org/apache/dubbo/common/url/component/param/DefaultDynamicParamSource.java Normal file
View File

@ -0,0 +1,65 @@
/*
* 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.common.url.component.param;
import org.apache.dubbo.common.constants.CommonConstants;
import java.util.List;
public class DefaultDynamicParamSource implements DynamicParamSource {
@Override
public void init(List<String> keys, List<ParamValue> values) {
keys.add(CommonConstants.VERSION_KEY);
values.add(new DynamicValues(null));
keys.add(CommonConstants.SIDE_KEY);
values.add(new FixedParamValue(CommonConstants.CONSUMER_SIDE, CommonConstants.PROVIDER_SIDE));
keys.add(CommonConstants.INTERFACE_KEY);
values.add(new DynamicValues(null));
keys.add(CommonConstants.PID_KEY);
values.add(new DynamicValues(null));
keys.add(CommonConstants.THREADPOOL_KEY);
values.add(new DynamicValues(null));
keys.add(CommonConstants.GROUP_KEY);
values.add(new DynamicValues(null));
keys.add(CommonConstants.VERSION_KEY);
values.add(new DynamicValues(null));
keys.add(CommonConstants.METADATA_KEY);
values.add(new DynamicValues(null));
keys.add(CommonConstants.APPLICATION_KEY);
values.add(new DynamicValues(null));
keys.add(CommonConstants.DUBBO_VERSION_KEY);
values.add(new DynamicValues(null));
keys.add(CommonConstants.RELEASE_KEY);
values.add(new DynamicValues(null));
keys.add(CommonConstants.PATH_KEY);
values.add(new DynamicValues(null));
keys.add(CommonConstants.ANYHOST_KEY);
values.add(new DynamicValues(null));
}
}

3
dubbo-common/src/main/java/org/apache/dubbo/common/url/component/param/DynamicParamSource.java
View File

@ -19,8 +19,9 @@ package org.apache.dubbo.common.url.component.param;
import org.apache.dubbo.common.extension.SPI;
import java.util.List;
@SPI
public interface DynamicParamSource {
void init(List<String> KEYS, List<ParamValue> VALUES);
void init(List<String> keys, List<ParamValue> values);
}

28
dubbo-common/src/main/java/org/apache/dubbo/common/url/component/param/DynamicParamTable.java
View File

@ -42,12 +42,15 @@ public final class DynamicParamTable {
init();
}
public static Integer getKeyIndex(String key) {
public static Integer getKeyIndex(boolean enabled, String key) {
if (!enabled) {
return null;
}
return KEY2INDEX.get(key);
}
public static Integer getValueIndex(String key, String value) {
Integer idx = getKeyIndex(key);
Integer idx = getKeyIndex(true, key);
if (idx == null) {
throw new IllegalArgumentException("Cannot found key in url param:" + key);
}
@ -60,7 +63,7 @@ public final class DynamicParamTable {
}
public static boolean isDefaultValue(String key, String value) {
return Objects.equals(value, VALUES.get(getKeyIndex(key)).defaultVal());
return Objects.equals(value, VALUES.get(getKeyIndex(true, key)).defaultVal());
}
public static String getValue(int vi, Integer offset) {
@ -78,18 +81,15 @@ public final class DynamicParamTable {
keys.add("");
values.add(new DynamicValues(null));
// Cache key and defaultValue
keys.add("version");
values.add(new DynamicValues(null));
keys.add("side");
values.add(new FixedParamValue("consumer", "provider"));
KEYS.addAll(keys);
VALUES.addAll(values);
ExtensionLoader.getExtensionLoader(DynamicParamSource.class)
.getSupportedExtensionInstances().forEach(source -> source.init(KEYS, VALUES));
.getSupportedExtensionInstances().forEach(source -> source.init(keys, values));
for (int i = 0; i < keys.size(); i++) {
if (!KEYS.contains(keys.get(i))) {
KEYS.add(keys.get(i));
VALUES.add(values.get(i));
}
}
for (int i = 0; i < KEYS.size(); i++) {
if (!KEYS.get(i).isEmpty()) {

1
dubbo-common/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.common.url.component.param.DynamicParamSource Normal file
View File

@ -0,0 +1 @@
default=org.apache.dubbo.common.url.component.param.DefaultDynamicParamSource

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

@ -767,6 +767,12 @@
META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.ProviderURLMergeProcessor
</resource>
</transformer>
<transformer
implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>
META-INF/dubbo/internal/org.apache.dubbo.common.url.component.param.DynamicParamSource
</resource>
</transformer>
</transformers>
<filters>
<filter>

2
dubbo-monitor/dubbo-monitor-api/src/main/java/org/apache/dubbo/monitor/support/MonitorFilter.java
View File

@ -131,7 +131,7 @@ public class MonitorFilter implements Filter, Filter.Listener {
return;
}
URL statisticsURL = createStatisticsUrl(invoker, invocation, result, remoteHost, start, error);
monitor.collect(statisticsURL);
monitor.collect(statisticsURL.toSerializableURL());
}
} catch (Throwable t) {
logger.warn("Failed to monitor count service " + invoker.getUrl() + ", cause: " + t.getMessage(), t);

2
dubbo-monitor/dubbo-monitor-default/src/main/java/org/apache/dubbo/monitor/dubbo/DubboMonitor.java
View File

@ -119,7 +119,7 @@ public class DubboMonitor implements Monitor {
MonitorService.MAX_CONCURRENT, String.valueOf(maxConcurrent),
DEFAULT_PROTOCOL, protocol
);
monitorService.collect(url);
monitorService.collect(url.toSerializableURL());
// reset
long[] current;

10
dubbo-monitor/dubbo-monitor-default/src/test/java/org/apache/dubbo/monitor/dubbo/DubboMonitorTest.java
View File

@ -107,7 +107,7 @@ public class DubboMonitorTest {
.addParameter(MonitorService.CONCURRENT, 1)
.addParameter(MonitorService.MAX_CONCURRENT, 1)
.build();
monitor.collect(statistics);
monitor.collect(statistics.toSerializableURL());
monitor.send();
while (lastStatistics == null) {
Thread.sleep(10);
@ -158,7 +158,7 @@ public class DubboMonitorTest {
continue;
}
try {
monitor.collect(statistics);
monitor.collect(statistics.toSerializableURL());
int i = 0;
while (monitorService.getStatistics() == null && i < 200) {
i++;
@ -211,10 +211,10 @@ public class DubboMonitorTest {
given(invoker.getUrl()).willReturn(URL.valueOf("dubbo://127.0.0.1:7070?interval=20"));
DubboMonitor dubboMonitor = new DubboMonitor(invoker, monitorService);
dubboMonitor.collect(statistics);
dubboMonitor.collect(statistics.toSerializableURL());
dubboMonitor.collect(statistics.addParameter(MonitorService.SUCCESS, 3).addParameter(MonitorService.CONCURRENT, 2)
.addParameter(MonitorService.INPUT, 1).addParameter(MonitorService.OUTPUT, 2));
dubboMonitor.collect(statistics.addParameter(MonitorService.SUCCESS, 6).addParameter(MonitorService.ELAPSED, 2));
.addParameter(MonitorService.INPUT, 1).addParameter(MonitorService.OUTPUT, 2).toSerializableURL());
dubboMonitor.collect(statistics.addParameter(MonitorService.SUCCESS, 6).addParameter(MonitorService.ELAPSED, 2).toSerializableURL());
dubboMonitor.send();