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

✨ Enable consumers to request service providers by priority … (#10533) 

* ✨ Enable consumers to request service providers by priority using the appropriate serialization type

* ✏️ Fixing PR specification

* ✏️ Optimization of import

* 🎉 Optimization of import and comment

* 🎉 Optimization of import

* 🎉 Optimization of import

* 🎉 fix Junit

* 🎉 Default prefer serialization is `hessian2`

* fix comment
This commit is contained in:
XS 2022-10-08 11:23:01 +08:00 committed by GitHub
parent 5749ac8051
commit f2cee0ab89
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 258 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
dubbo-common/src/main/java/org/apache/dubbo/common/constants/ProviderConstants.java Normal file
View File

@ -0,0 +1,28 @@
/*
* 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.constants;
/**
* Provider Constants
*/
public interface ProviderConstants {
/**
* Default prefer serialization,multiple separated by commas
*/
String DEFAULT_PREFER_SERIALIZATION = "hessian2";
}

26
dubbo-common/src/main/java/org/apache/dubbo/config/AbstractServiceConfig.java
View File

@ -17,6 +17,7 @@
package org.apache.dubbo.config;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.config.support.Parameter;
import org.apache.dubbo.rpc.model.ModuleModel;
@ -27,6 +28,7 @@ import java.util.List;
import static org.apache.dubbo.common.constants.CommonConstants.EXPORTER_LISTENER_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.EXPORT_ASYNC_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.SERVICE_FILTER_KEY;
import static org.apache.dubbo.common.constants.ProviderConstants.DEFAULT_PREFER_SERIALIZATION;
/**
* AbstractServiceConfig
@ -121,6 +123,18 @@ public abstract class AbstractServiceConfig extends AbstractInterfaceConfig {
*/
private String serialization;
/**
* If the parameter has a value, the consumer will read the parameter first.
* If the Dubbo Sdk you are using contains the serialization type, the serialization method specified by the argument is used.
* <p>
* When this parameter is null or the serialization type specified by this parameter does not exist in the Dubbo SDK, the serialization type specified by serialization is used.
* If the Dubbo SDK if still does not exist, the default type of the Dubbo SDK is used.
* For Dubbo SDK >= 3.1, <code>preferSerialization</code> takes precedence over <code>serialization</code>
* <p>
* The configuration supports multiple, which are separated by commas.Such as:<code>fastjson2,fastjson,hessian2</code>
*/
private String preferSerialization; // default:hessian2
/**
* Weather the service is export asynchronously
* @deprecated
@ -145,6 +159,10 @@ public abstract class AbstractServiceConfig extends AbstractInterfaceConfig {
if (dynamic == null) {
dynamic = true;
}
if (StringUtils.isBlank(preferSerialization)) {
preferSerialization = DEFAULT_PREFER_SERIALIZATION;
}
}
@Override
@ -323,6 +341,14 @@ public abstract class AbstractServiceConfig extends AbstractInterfaceConfig {
this.serialization = serialization;
}
public String getPreferSerialization() {
return preferSerialization;
}
public void setPreferSerialization(String preferSerialization) {
this.preferSerialization = preferSerialization;
}
@Deprecated
@Parameter(key = EXPORT_ASYNC_KEY)
public Boolean getExportAsync() {

23
dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/bootstrap/builders/AbstractServiceBuilder.java
View File

@ -29,7 +29,7 @@ import java.util.List;
* @since 2.7
*/
public abstract class AbstractServiceBuilder<T extends AbstractServiceConfig, B extends AbstractServiceBuilder<T, B>>
extends AbstractInterfaceBuilder<T, B> {
extends AbstractInterfaceBuilder<T, B> {
/**
* The service version
@ -107,6 +107,11 @@ public abstract class AbstractServiceBuilder<T extends AbstractServiceConfig, B
*/
private String serialization;
/**
* The prefer serialization type
*/
private String preferSerialization;
public B version(String version) {
this.version = version;
return getThis();
@ -211,11 +216,22 @@ public abstract class AbstractServiceBuilder<T extends AbstractServiceConfig, B
return getThis();
}
public B serialization(String serialization) {
public B serialization(String serialization) {
this.serialization = serialization;
return getThis();
}
/**
* The prefer serialization type
*
* @param preferSerialization prefer serialization type
* @return {@link B}
*/
public B preferSerialization(String preferSerialization) {
this.preferSerialization = preferSerialization;
return getThis();
}
@Override
public void build(T instance) {
super.build(instance);
@ -268,5 +284,8 @@ public abstract class AbstractServiceBuilder<T extends AbstractServiceConfig, B
if (!StringUtils.isEmpty(serialization)) {
instance.setSerialization(serialization);
}
if (StringUtils.isNotBlank(preferSerialization)) {
instance.setPreferSerialization(preferSerialization);
}
}
}

7
dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/AbstractServiceConfigTest.java
View File

@ -177,6 +177,13 @@ public class AbstractServiceConfigTest {
assertThat(serviceConfig.getSerialization(), equalTo("serialization"));
}
@Test
public void testPreferSerialization() throws Exception {
ServiceConfig serviceConfig = new ServiceConfig();
serviceConfig.setPreferSerialization("preferSerialization");
assertThat(serviceConfig.getPreferSerialization(), equalTo("preferSerialization"));
}
private static class ServiceConfig extends AbstractServiceConfig {

5
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryProtocol.java
View File

@ -118,6 +118,7 @@ import static org.apache.dubbo.remoting.Constants.CODEC_KEY;
import static org.apache.dubbo.remoting.Constants.CONNECTIONS_KEY;
import static org.apache.dubbo.remoting.Constants.EXCHANGER_KEY;
import static org.apache.dubbo.remoting.Constants.SERIALIZATION_KEY;
import static org.apache.dubbo.remoting.Constants.PREFER_SERIALIZATION_KEY;
import static org.apache.dubbo.rpc.Constants.DEPRECATED_KEY;
import static org.apache.dubbo.rpc.Constants.GENERIC_KEY;
import static org.apache.dubbo.rpc.Constants.INTERFACES;
@ -135,7 +136,7 @@ import static org.apache.dubbo.rpc.model.ScopeModelUtil.getApplicationModel;
*/
public class RegistryProtocol implements Protocol, ScopeModelAware {
public static final String[] DEFAULT_REGISTER_PROVIDER_KEYS = {
APPLICATION_KEY, CODEC_KEY, EXCHANGER_KEY, SERIALIZATION_KEY, CLUSTER_KEY, CONNECTIONS_KEY, DEPRECATED_KEY,
APPLICATION_KEY, CODEC_KEY, EXCHANGER_KEY, SERIALIZATION_KEY, PREFER_SERIALIZATION_KEY, CLUSTER_KEY, CONNECTIONS_KEY, DEPRECATED_KEY,
GROUP_KEY, LOADBALANCE_KEY, MOCK_KEY, PATH_KEY, TIMEOUT_KEY, TOKEN_KEY, VERSION_KEY, WARMUP_KEY,
WEIGHT_KEY, DUBBO_VERSION_KEY, RELEASE_KEY, SIDE_KEY, IPV6_KEY
};
@ -487,7 +488,7 @@ public class RegistryProtocol implements Protocol, ScopeModelAware {
Map<String, Object> consumerAttribute = new HashMap<>(url.getAttributes());
consumerAttribute.remove(REFER_KEY);
String p = isEmpty(parameters.get(PROTOCOL_KEY)) ? CONSUMER : parameters.get(PROTOCOL_KEY);
URL consumerUrl = new ServiceConfigURL (
URL consumerUrl = new ServiceConfigURL(
p,
null,
null,

5
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/Constants.java
View File

@ -79,6 +79,11 @@ public interface Constants {
String SERIALIZATION_KEY = "serialization";
/**
* Prefer serialization
*/
String PREFER_SERIALIZATION_KEY = "prefer_serialization";
String DEFAULT_REMOTING_SERIALIZATION_PROPERTY_KEY = "DUBBO_DEFAULT_SERIALIZATION";
String CODEC_KEY = "codec";

39
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/CodecSupport.java
View File

@ -24,9 +24,8 @@ import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.serialize.ObjectInput;
import org.apache.dubbo.common.serialize.ObjectOutput;
import org.apache.dubbo.common.serialize.Serialization;
import org.apache.dubbo.common.serialize.support.DefaultSerializationSelector;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.remoting.Constants;
import org.apache.dubbo.remoting.utils.UrlUtils;
import org.apache.dubbo.rpc.model.FrameworkModel;
import org.apache.dubbo.rpc.model.FrameworkServiceRepository;
@ -83,11 +82,10 @@ public class CodecSupport {
}
public static Serialization getSerialization(URL url) {
return url.getOrDefaultFrameworkModel().getExtensionLoader(Serialization.class).getExtension(
url.getParameter(Constants.SERIALIZATION_KEY, DefaultSerializationSelector.getDefaultRemotingSerialization()));
return url.getOrDefaultFrameworkModel().getExtensionLoader(Serialization.class).getExtension(UrlUtils.serializationOrDefault(url));
}
public static Serialization getSerialization(URL url, Byte id) throws IOException {
public static Serialization getSerialization(Byte id) throws IOException {
Serialization result = getSerializationById(id);
if (result == null) {
throw new IOException("Unrecognized serialize type from consumer: " + id);
@ -96,7 +94,7 @@ public class CodecSupport {
}
public static ObjectInput deserialize(URL url, InputStream is, byte proto) throws IOException {
Serialization s = getSerialization(url, proto);
Serialization s = getSerialization(proto);
return s.deserialize(url, is);
}
@ -167,20 +165,29 @@ public class CodecSupport {
if (CollectionUtils.isEmpty(urls)) {
throw new IOException("Service " + path + " with version " + version + " not found, invocation rejected.");
} else {
boolean match = false;
for (URL url : urls) {
String serializationName = url.getParameter(org.apache.dubbo.remoting.Constants.SERIALIZATION_KEY, DefaultSerializationSelector.getDefaultRemotingSerialization());
Byte localId = SERIALIZATIONNAME_ID_MAP.get(serializationName);
if (localId != null && localId.equals(id)) {
match = true;
}
}
if(!match) {
boolean match = urls.stream().anyMatch(url -> isMatch(url, id));
if (!match) {
throw new IOException("Unexpected serialization id:" + id + " received from network, please check if the peer send the right id.");
}
}
}
/**
* Is Match
*
* @param url url
* @param id id
* @return boolean
*/
private static boolean isMatch(URL url, Byte id) {
Byte localId;
for (String serialization : UrlUtils.allSerializations(url)) {
localId = SERIALIZATIONNAME_ID_MAP.get(serialization);
if (id.equals(localId)) {
return true;
}
}
return false;
}
}

78
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/utils/UrlUtils.java
View File

@ -18,7 +18,21 @@
package org.apache.dubbo.remoting.utils;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.serialize.support.DefaultSerializationSelector;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.remoting.Constants;
import org.apache.dubbo.remoting.transport.CodecSupport;
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import static org.apache.dubbo.remoting.Constants.PREFER_SERIALIZATION_KEY;
import static org.apache.dubbo.remoting.Constants.SERIALIZATION_KEY;
public class UrlUtils {
public static int getIdleTimeout(URL url) {
@ -34,4 +48,68 @@ public class UrlUtils {
public static int getHeartbeat(URL url) {
return url.getParameter(Constants.HEARTBEAT_KEY, Constants.DEFAULT_HEARTBEAT);
}
/**
* Get the serialization id
*
* @param url url
* @return {@link Byte}
*/
public static Byte serializationId(URL url) {
Byte serializationId;
// Obtain the value from prefer_serialization. Such as:fastjson2,hessian2
List<String> preferSerials = preferSerialization(url);
for (String preferSerial : preferSerials) {
if ((serializationId = CodecSupport.getIDByName(preferSerial)) != null) {
return serializationId;
}
}
// Secondly, obtain the value from serialization
if ((serializationId = CodecSupport.getIDByName(url.getParameter(SERIALIZATION_KEY))) != null) {
return serializationId;
}
// Finally, use the default serialization type
return CodecSupport.getIDByName(DefaultSerializationSelector.getDefaultRemotingSerialization());
}
/**
* Get the serialization or default serialization
*
* @param url url
* @return {@link String}
*/
public static String serializationOrDefault(URL url) {
return allSerializations(url).stream().findFirst().get();
}
/**
* Get the all serializations,ensure insertion order
*
* @param url url
* @return {@link List}<{@link String}>
*/
public static Collection<String> allSerializations(URL url) {
// preferSerialization -> serialization -> default serialization
Set<String> serializations = new LinkedHashSet<>();
UrlUtils.preferSerialization(url).forEach(serializations::add);
Optional.ofNullable(url.getParameter(SERIALIZATION_KEY)).filter(StringUtils::isNotBlank).ifPresent(serializations::add);
serializations.add(DefaultSerializationSelector.getDefaultRemotingSerialization());
return Collections.unmodifiableSet(serializations);
}
/**
* Prefer Serialization
*
* @param url url
* @return {@link List}<{@link String}>
*/
public static List<String> preferSerialization(URL url) {
String preferSerialization = url.getParameter(PREFER_SERIALIZATION_KEY);
if (StringUtils.isNotBlank(preferSerialization)) {
return Collections.unmodifiableList(StringUtils.splitToList(preferSerialization, ','));
}
return Collections.emptyList();
}
}

5
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/CompatibleFilter.java
View File

@ -20,6 +20,7 @@ import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CompatibleTypeUtils;
import org.apache.dubbo.common.utils.PojoUtils;
import org.apache.dubbo.remoting.utils.UrlUtils;
import org.apache.dubbo.rpc.Filter;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
@ -29,8 +30,6 @@ import org.apache.dubbo.rpc.RpcException;
import java.lang.reflect.Method;
import java.lang.reflect.Type;
import static org.apache.dubbo.remoting.Constants.SERIALIZATION_KEY;
/**
* CompatibleFilter make the remote method's return value compatible to invoker's version of object.
* To make return object compatible it does
@ -63,7 +62,7 @@ public class CompatibleFilter implements Filter, Filter.Listener {
Method method = invoker.getInterface().getMethod(invocation.getMethodName(), invocation.getParameterTypes());
Class<?> type = method.getReturnType();
Object newValue;
String serialization = invoker.getUrl().getParameter(SERIALIZATION_KEY);
String serialization = UrlUtils.serializationOrDefault(invoker.getUrl());
if ("json".equals(serialization) || "fastjson".equals(serialization)) {
// If the serialization key is json or fastjson
Type gtype = method.getGenericReturnType();

49
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractInvoker.java
View File

@ -22,7 +22,6 @@ import org.apache.dubbo.common.Version;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.serialize.support.DefaultSerializationSelector;
import org.apache.dubbo.common.threadpool.ThreadlessExecutor;
import org.apache.dubbo.common.threadpool.manager.ExecutorRepository;
import org.apache.dubbo.common.utils.ArrayUtils;
@ -30,7 +29,7 @@ import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.remoting.RemotingException;
import org.apache.dubbo.remoting.TimeoutException;
import org.apache.dubbo.remoting.transport.CodecSupport;
import org.apache.dubbo.remoting.utils.UrlUtils;
import org.apache.dubbo.rpc.AsyncRpcResult;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.InvokeMode;
@ -51,7 +50,6 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY;
import static org.apache.dubbo.remoting.Constants.SERIALIZATION_KEY;
import static org.apache.dubbo.rpc.Constants.SERIALIZATION_ID_KEY;
/**
@ -110,9 +108,7 @@ public abstract class AbstractInvoker<T> implements Invoker<T> {
}
this.type = type;
this.url = url;
this.attachment = attachment == null
? null
: Collections.unmodifiableMap(attachment);
this.attachment = attachment == null ? null : Collections.unmodifiableMap(attachment);
}
private static Map<String, Object> convertAttachment(URL url, String[] keys) {
@ -169,8 +165,7 @@ public abstract class AbstractInvoker<T> implements Invoker<T> {
public Result invoke(Invocation inv) throws RpcException {
// if invoker is destroyed due to address refresh from registry, let's allow the current invoke to proceed
if (isDestroyed()) {
logger.warn("Invoker for service " + this + " on consumer " + NetUtils.getLocalHost() + " is destroyed, "
+ ", dubbo version is " + Version.getVersion() + ", this invoker should not be used any longer");
logger.warn("Invoker for service " + this + " on consumer " + NetUtils.getLocalHost() + " is destroyed, " + ", dubbo version is " + Version.getVersion() + ", this invoker should not be used any longer");
}
RpcInvocation invocation = (RpcInvocation) inv;
@ -196,7 +191,24 @@ public abstract class AbstractInvoker<T> implements Invoker<T> {
RpcUtils.attachInvocationIdIfAsync(getUrl(), inv);
Byte serializationId = CodecSupport.getIDByName(getUrl().getParameter(SERIALIZATION_KEY, DefaultSerializationSelector.getDefaultRemotingSerialization()));
attachInvocationSerializationId(inv);
}
/**
* Attach Invocation Serialization id
* <p>
* <ol>
* <li>Obtain the value from <code>prefer_serialization</code></li>
* <li>If the preceding information is not obtained, obtain the value from <code>serialization</code></li>
* <li>If neither is obtained, use the default value</li>
* </ol>
* </p>
*
* @param inv inv
*/
private void attachInvocationSerializationId(RpcInvocation inv) {
Byte serializationId = UrlUtils.serializationId(getUrl());
if (serializationId != null) {
inv.put(SERIALIZATION_ID_KEY, serializationId);
}
@ -267,23 +279,18 @@ public abstract class AbstractInvoker<T> implements Invoker<T> {
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RpcException("Interrupted unexpectedly while waiting for remote result to return! method: " +
invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
throw new RpcException("Interrupted unexpectedly while waiting for remote result to return! method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
} catch (ExecutionException e) {
Throwable rootCause = e.getCause();
if (rootCause instanceof TimeoutException) {
throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " +
invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
} else if (rootCause instanceof RemotingException) {
throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " +
invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
} else {
throw new RpcException(RpcException.UNKNOWN_EXCEPTION, "Fail to invoke remote method: " +
invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
throw new RpcException(RpcException.UNKNOWN_EXCEPTION, "Fail to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
}
} catch (java.util.concurrent.TimeoutException e) {
throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " +
invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
} catch (Throwable e) {
throw new RpcException(e.getMessage(), e);
}
@ -295,9 +302,7 @@ public abstract class AbstractInvoker<T> implements Invoker<T> {
if (InvokeMode.SYNC == RpcUtils.getInvokeMode(getUrl(), inv)) {
return new ThreadlessExecutor();
}
return url.getOrDefaultApplicationModel().getExtensionLoader(ExecutorRepository.class)
.getDefaultExtension()
.getExecutor(url);
return url.getOrDefaultApplicationModel().getExtensionLoader(ExecutorRepository.class).getDefaultExtension().getExecutor(url);
}
/**

2
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DecodeableRpcInvocation.java
View File

@ -113,7 +113,7 @@ public class DecodeableRpcInvocation extends RpcInvocation implements Codec, Dec
@Override
public Object decode(Channel channel, InputStream input) throws IOException {
ObjectInput in = CodecSupport.getSerialization(channel.getUrl(), serializationType)
ObjectInput in = CodecSupport.getSerialization(serializationType)
.deserialize(channel.getUrl(), input);
this.put(SERIALIZATION_ID_KEY, serializationType);

4
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DecodeableRpcResult.java
View File

@ -86,8 +86,8 @@ public class DecodeableRpcResult extends AppResponse implements Codec, Decodeabl
if (invocation != null && invocation.getServiceModel() != null) {
Thread.currentThread().setContextClassLoader(invocation.getServiceModel().getClassLoader());
}
ObjectInput in = CodecSupport.getSerialization(channel.getUrl(), serializationType)
.deserialize(channel.getUrl(), input);
ObjectInput in = CodecSupport.getSerialization(serializationType)
.deserialize(channel.getUrl(), input);
byte flag = in.readByte();
switch (flag) {

9
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCodecSupport.java
View File

@ -18,9 +18,8 @@ package org.apache.dubbo.rpc.protocol.dubbo;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.serialize.Serialization;
import org.apache.dubbo.common.serialize.support.DefaultSerializationSelector;
import org.apache.dubbo.remoting.Constants;
import org.apache.dubbo.remoting.transport.CodecSupport;
import org.apache.dubbo.remoting.utils.UrlUtils;
import org.apache.dubbo.rpc.AppResponse;
import org.apache.dubbo.rpc.Invocation;
@ -34,8 +33,7 @@ public class DubboCodecSupport {
if (serializationTypeObj != null) {
return CodecSupport.getSerializationById((byte) serializationTypeObj);
}
return url.getOrDefaultFrameworkModel().getExtensionLoader(Serialization.class).getExtension(
url.getParameter(org.apache.dubbo.remoting.Constants.SERIALIZATION_KEY, DefaultSerializationSelector.getDefaultRemotingSerialization()));
return url.getOrDefaultFrameworkModel().getExtensionLoader(Serialization.class).getExtension(UrlUtils.serializationOrDefault(url));
}
public static Serialization getResponseSerialization(URL url, AppResponse appResponse) {
@ -47,7 +45,6 @@ public class DubboCodecSupport {
return CodecSupport.getSerializationById((byte) serializationTypeObj);
}
}
return url.getOrDefaultFrameworkModel().getExtensionLoader(Serialization.class).getExtension(
url.getParameter(Constants.SERIALIZATION_KEY, DefaultSerializationSelector.getDefaultRemotingSerialization()));
return url.getOrDefaultFrameworkModel().getExtensionLoader(Serialization.class).getExtension(UrlUtils.serializationOrDefault(url));
}
}

20
dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/DubboInvokerAvailableTest.java
View File

@ -151,6 +151,26 @@ public class DubboInvokerAvailableTest {
Assertions.assertFalse(invoker.isAvailable());
}
/**
* The test prefer serialization
*
* @throws Exception Exception
*/
@Test
public void testPreferSerialization() throws Exception {
int port = NetUtils.getAvailablePort();
URL url = URL.valueOf("dubbo://127.0.0.1:" + port + "/org.apache.dubbo.rpc.protocol.dubbo.IDemoService?lazy=true&connections=1&timeout=10000&serialization=fastjson&prefer_serialization=fastjson2,hessian2");
ProtocolUtils.export(new DemoServiceImpl(), IDemoService.class, url);
Invoker<?> invoker = protocol.refer(IDemoService.class, url);
Assertions.assertTrue(invoker.isAvailable());
ExchangeClient exchangeClient = getClients((DubboInvoker<?>) invoker)[0];
Assertions.assertFalse(exchangeClient.isClosed());
//invoke method --> init client
IDemoService service = (IDemoService) proxy.getProxy(invoker);
Assertions.assertEquals("ok", service.get());
}
private ExchangeClient[] getClients(DubboInvoker<?> invoker) throws Exception {
Field field = DubboInvoker.class.getDeclaredField("clients");
field.setAccessible(true);

5
dubbo-rpc/dubbo-rpc-injvm/src/main/java/org/apache/dubbo/rpc/protocol/injvm/DefaultParamDeepCopyUtil.java
View File

@ -22,8 +22,7 @@ import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.serialize.ObjectInput;
import org.apache.dubbo.common.serialize.ObjectOutput;
import org.apache.dubbo.common.serialize.Serialization;
import org.apache.dubbo.common.serialize.support.DefaultSerializationSelector;
import org.apache.dubbo.remoting.Constants;
import org.apache.dubbo.remoting.utils.UrlUtils;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
@ -38,7 +37,7 @@ public class DefaultParamDeepCopyUtil implements ParamDeepCopyUtil {
@SuppressWarnings({"unchecked"})
public <T> T copy(URL url, Object src, Class<T> targetClass) {
Serialization serialization = url.getOrDefaultFrameworkModel().getExtensionLoader(Serialization.class).getExtension(
url.getParameter(Constants.SERIALIZATION_KEY, DefaultSerializationSelector.getDefaultRemotingSerialization()));
UrlUtils.serializationOrDefault(url));
try (ByteArrayOutputStream outputStream = new ByteArrayOutputStream()) {
ObjectOutput objectOutput = serialization.serialize(url, outputStream);

23
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ReflectionPackableMethod.java
View File

@ -17,19 +17,18 @@
package org.apache.dubbo.rpc.protocol.tri;
import com.google.protobuf.ByteString;
import com.google.protobuf.Message;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.serialize.MultipleSerialization;
import org.apache.dubbo.common.serialize.support.DefaultSerializationSelector;
import org.apache.dubbo.common.stream.StreamObserver;
import org.apache.dubbo.config.Constants;
import org.apache.dubbo.remoting.utils.UrlUtils;
import org.apache.dubbo.rpc.model.MethodDescriptor;
import org.apache.dubbo.rpc.model.PackableMethod;
import org.apache.dubbo.triple.TripleWrapper;
import com.google.protobuf.ByteString;
import com.google.protobuf.Message;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
@ -39,7 +38,6 @@ import java.util.stream.Stream;
import static org.apache.dubbo.common.constants.CommonConstants.$ECHO;
import static org.apache.dubbo.common.constants.CommonConstants.PROTOBUF_MESSAGE_CLASS_NAME;
import static org.apache.dubbo.remoting.Constants.SERIALIZATION_KEY;
import static org.apache.dubbo.rpc.protocol.tri.TripleProtocol.METHOD_ATTR_PACK;
public class ReflectionPackableMethod implements PackableMethod {
@ -104,8 +102,7 @@ public class ReflectionPackableMethod implements PackableMethod {
}
public static ReflectionPackableMethod init(MethodDescriptor methodDescriptor, URL url) {
final String serializeName = url.getParameter(SERIALIZATION_KEY,
DefaultSerializationSelector.getDefaultRemotingSerialization());
final String serializeName = UrlUtils.serializationOrDefault(url);
Object stored = methodDescriptor.getAttribute(METHOD_ATTR_PACK);
if (stored != null) {
return (ReflectionPackableMethod) stored;
@ -127,7 +124,7 @@ public class ReflectionPackableMethod implements PackableMethod {
* @return true if the request and response object is not generated by protobuf
*/
static boolean needWrap(MethodDescriptor methodDescriptor, Class<?>[] parameterClasses,
Class<?> returnClass) {
Class<?> returnClass) {
String methodName = methodDescriptor.getMethodName();
// generic call must be wrapped
if (CommonConstants.$INVOKE.equals(methodName) || CommonConstants.$INVOKE_ASYNC.equals(
@ -316,7 +313,7 @@ public class ReflectionPackableMethod implements PackableMethod {
String serialize;
private WrapResponsePack(MultipleSerialization multipleSerialization, URL url,
String returnType) {
String returnType) {
this.multipleSerialization = multipleSerialization;
this.url = url;
this.returnType = returnType;
@ -364,10 +361,10 @@ public class ReflectionPackableMethod implements PackableMethod {
private final boolean singleArgument;
private WrapRequestPack(MultipleSerialization multipleSerialization,
URL url,
String serialize,
String[] argumentsType,
boolean singleArgument) {
URL url,
String serialize,
String[] argumentsType,
boolean singleArgument) {
this.url = url;
this.serialize = convertHessianToWrapper(serialize);
this.multipleSerialization = multipleSerialization;