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

fix #10105, support serialization optimizer (#10147)

This commit is contained in:
earthchen 2022-06-13 18:12:54 +08:00 committed by GitHub
parent f7bf43b0c4
commit 162b98f6ba
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 50 additions and 42 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/constants/CommonConstants.java
View File

@ -539,4 +539,8 @@ public interface CommonConstants {
String METADATA = "metadata";
String IGNORE_LISTEN_SHUTDOWN_HOOK = "dubbo.shutdownHook.listenIgnore";
String OPTIMIZER_KEY = "optimizer";
}

43
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AbstractProtocol.java
View File

@ -20,7 +20,10 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.config.ConfigurationUtils;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.serialize.support.SerializableClassRegistry;
import org.apache.dubbo.common.serialize.support.SerializationOptimizer;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.remoting.Constants;
import org.apache.dubbo.rpc.Exporter;
import org.apache.dubbo.rpc.Invoker;
@ -41,6 +44,7 @@ import java.util.concurrent.ConcurrentHashMap;
import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_SERVER_SHUTDOWN_TIMEOUT;
import static org.apache.dubbo.common.constants.CommonConstants.SHUTDOWN_WAIT_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.OPTIMIZER_KEY;
/**
* abstract ProtocolSupport.
@ -61,6 +65,9 @@ public abstract class AbstractProtocol implements Protocol, ScopeModelAware {
protected FrameworkModel frameworkModel;
private final Set<String> optimizers = new ConcurrentHashSet<>();
@Override
public void setFrameworkModel(FrameworkModel frameworkModel) {
this.frameworkModel = frameworkModel;
@ -136,4 +143,40 @@ public abstract class AbstractProtocol implements Protocol, ScopeModelAware {
public Collection<Exporter<?>> getExporters() {
return Collections.unmodifiableCollection(exporterMap.values());
}
protected void optimizeSerialization(URL url) throws RpcException {
String className = url.getParameter(OPTIMIZER_KEY, "");
if (StringUtils.isEmpty(className) || optimizers.contains(className)) {
return;
}
logger.info("Optimizing the serialization process for Kryo, FST, etc...");
try {
Class clazz = Thread.currentThread().getContextClassLoader().loadClass(className);
if (!SerializationOptimizer.class.isAssignableFrom(clazz)) {
throw new RpcException("The serialization optimizer " + className + " isn't an instance of " + SerializationOptimizer.class.getName());
}
SerializationOptimizer optimizer = (SerializationOptimizer) clazz.newInstance();
if (optimizer.getSerializableClasses() == null) {
return;
}
for (Class c : optimizer.getSerializableClasses()) {
SerializableClassRegistry.registerClass(c);
}
optimizers.add(className);
} catch (ClassNotFoundException e) {
throw new RpcException("Cannot find the serialization optimizer class: " + className, e);
} catch (InstantiationException | IllegalAccessException e) {
throw new RpcException("Cannot instantiate the serialization optimizer class: " + className, e);
}
}
}

1
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/Constants.java
View File

@ -55,7 +55,6 @@ public interface Constants {
boolean DEFAULT_LAZY_REQUEST_WITH_WARNING = false;
String OPTIMIZER_KEY = "optimizer";
String ON_CONNECT_KEY = "onconnect";

39
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java
View File

@ -20,11 +20,8 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.URLBuilder;
import org.apache.dubbo.common.config.ConfigurationUtils;
import org.apache.dubbo.common.extension.ExtensionLoader;
import org.apache.dubbo.common.serialize.support.SerializableClassRegistry;
import org.apache.dubbo.common.serialize.support.SerializationOptimizer;
import org.apache.dubbo.common.url.component.ServiceConfigURL;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.remoting.Channel;
@ -84,7 +81,6 @@ import static org.apache.dubbo.rpc.protocol.dubbo.Constants.DEFAULT_SHARE_CONNEC
import static org.apache.dubbo.rpc.protocol.dubbo.Constants.IS_CALLBACK_SERVICE;
import static org.apache.dubbo.rpc.protocol.dubbo.Constants.ON_CONNECT_KEY;
import static org.apache.dubbo.rpc.protocol.dubbo.Constants.ON_DISCONNECT_KEY;
import static org.apache.dubbo.rpc.protocol.dubbo.Constants.OPTIMIZER_KEY;
import static org.apache.dubbo.rpc.protocol.dubbo.Constants.SHARE_CONNECTIONS_KEY;
@ -104,7 +100,6 @@ public class DubboProtocol extends AbstractProtocol {
*/
private final Map<String, Object> referenceClientMap = new ConcurrentHashMap<>();
private static final Object PENDING_OBJECT = new Object();
private final Set<String> optimizers = new ConcurrentHashSet<>();
private AtomicBoolean destroyed = new AtomicBoolean();
@ -399,40 +394,6 @@ public class DubboProtocol extends AbstractProtocol {
return protocolServer;
}
private void optimizeSerialization(URL url) throws RpcException {
String className = url.getParameter(OPTIMIZER_KEY, "");
if (StringUtils.isEmpty(className) || optimizers.contains(className)) {
return;
}
logger.info("Optimizing the serialization process for Kryo, FST, etc...");
try {
Class clazz = Thread.currentThread().getContextClassLoader().loadClass(className);
if (!SerializationOptimizer.class.isAssignableFrom(clazz)) {
throw new RpcException("The serialization optimizer " + className + " isn't an instance of " + SerializationOptimizer.class.getName());
}
SerializationOptimizer optimizer = (SerializationOptimizer) clazz.newInstance();
if (optimizer.getSerializableClasses() == null) {
return;
}
for (Class c : optimizer.getSerializableClasses()) {
SerializableClassRegistry.registerClass(c);
}
optimizers.add(className);
} catch (ClassNotFoundException e) {
throw new RpcException("Cannot find the serialization optimizer class: " + className, e);
} catch (InstantiationException | IllegalAccessException e) {
throw new RpcException("Cannot instantiate the serialization optimizer class: " + className, e);
}
}
@Override
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {

5
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleProtocol.java
View File

@ -66,7 +66,6 @@ public class TripleProtocol extends AbstractProtocol {
private final PathResolver pathResolver;
private final TriBuiltinService triBuiltinService;
private final ConnectionManager connectionManager;
private final FrameworkModel frameworkModel;
private final String acceptEncodings;
private boolean versionChecked = false;
@ -124,12 +123,14 @@ public class TripleProtocol extends AbstractProtocol {
url.getOrDefaultApplicationModel().getExtensionLoader(ExecutorRepository.class)
.getDefaultExtension()
.createExecutorIfAbsent(url);
PortUnificationExchanger.bind(invoker.getUrl());
PortUnificationExchanger.bind(url);
optimizeSerialization(url);
return exporter;
}
@Override
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
optimizeSerialization(url);
ExecutorService streamExecutor = getOrCreateStreamExecutor(
url.getOrDefaultApplicationModel());
TripleInvoker<T> invoker = new TripleInvoker<>(type, url, acceptEncodings,