diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcInvocation.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcInvocation.java index f5ec9fd064..156b8172d5 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcInvocation.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcInvocation.java @@ -77,7 +77,7 @@ public class RpcInvocation implements Invocation, Serializable { */ private Map attachments; - private transient Lock attachmentLock = new ReentrantLock(); + private final transient Lock attachmentLock = new ReentrantLock(); /** * Only used on the caller side, will not appear on the wire. diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/InvokerInvocationHandler.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/InvokerInvocationHandler.java index ea75da5f22..255a5ad089 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/InvokerInvocationHandler.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/InvokerInvocationHandler.java @@ -33,16 +33,18 @@ import java.lang.reflect.Method; */ public class InvokerInvocationHandler implements InvocationHandler { private static final Logger logger = LoggerFactory.getLogger(InvokerInvocationHandler.class); + private final Invoker invoker; - private ServiceModel serviceModel; - private URL url; - private String protocolServiceKey; + + private final ServiceModel serviceModel; + + private final String protocolServiceKey; public InvokerInvocationHandler(Invoker handler) { this.invoker = handler; - this.url = invoker.getUrl(); - this.protocolServiceKey = this.url.getProtocolServiceKey(); - this.serviceModel = this.url.getServiceModel(); + URL url = invoker.getUrl(); + this.protocolServiceKey = url.getProtocolServiceKey(); + this.serviceModel = url.getServiceModel(); } @Override diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java index dcff5f4692..2d35784a05 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboProtocol.java @@ -52,7 +52,6 @@ import java.util.Collection; import java.util.Collections; import java.util.List; import java.util.Map; -import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; @@ -92,26 +91,28 @@ public class DubboProtocol extends AbstractProtocol { public static final String NAME = "dubbo"; public static final int DEFAULT_PORT = 20880; + private static final String IS_CALLBACK_SERVICE_INVOKE = "_isCallBackServiceInvoke"; /** * - * {@link Map} + * Map */ private final Map referenceClientMap = new ConcurrentHashMap<>(); + private static final Object PENDING_OBJECT = new Object(); - private AtomicBoolean destroyed = new AtomicBoolean(); + private final AtomicBoolean destroyed = new AtomicBoolean(); - private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() { + private final ExchangeHandler requestHandler = new ExchangeHandlerAdapter() { @Override public CompletableFuture reply(ExchangeChannel channel, Object message) throws RemotingException { if (!(message instanceof Invocation)) { throw new RemotingException(channel, "Unsupported request: " - + (message == null ? null : (message.getClass().getName() + ": " + message)) - + ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress()); + + (message == null ? null : (message.getClass().getName() + ": " + message)) + + ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress()); } Invocation inv = (Invocation) message; @@ -138,9 +139,9 @@ public class DubboProtocol extends AbstractProtocol { } if (!hasMethod) { logger.warn(new IllegalStateException("The methodName " + inv.getMethodName() - + " not found in callback service interface ,invoke will be ignored." - + " please update the api interface. url is:" - + invoker.getUrl()) + " ,invocation is :" + inv); + + " not found in callback service interface ,invoke will be ignored." + + " please update the api interface. url is:" + + invoker.getUrl()) + " ,invocation is :" + inv); return null; } } @@ -243,17 +244,12 @@ public class DubboProtocol extends AbstractProtocol { return (DubboProtocol) scopeModel.getExtensionLoader(Protocol.class).getExtension(DubboProtocol.NAME, false); } - @Override - public Collection> getExporters() { - return Collections.unmodifiableCollection(exporterMap.values()); - } - private boolean isClientSide(Channel channel) { InetSocketAddress address = channel.getRemoteAddress(); URL url = channel.getUrl(); return url.getPort() == address.getPort() && - NetUtils.filterLocalHost(channel.getUrl().getIp()) - .equals(NetUtils.filterLocalHost(address.getAddress().getHostAddress())); + NetUtils.filterLocalHost(channel.getUrl().getIp()) + .equals(NetUtils.filterLocalHost(address.getAddress().getHostAddress())); } Invoker getInvoker(Channel channel, Invocation inv) throws RemotingException { @@ -277,16 +273,16 @@ public class DubboProtocol extends AbstractProtocol { } String serviceKey = serviceKey( - port, - path, - (String) inv.getObjectAttachmentWithoutConvert(VERSION_KEY), - (String) inv.getObjectAttachmentWithoutConvert(GROUP_KEY) + port, + path, + (String) inv.getObjectAttachmentWithoutConvert(VERSION_KEY), + (String) inv.getObjectAttachmentWithoutConvert(GROUP_KEY) ); DubboExporter exporter = (DubboExporter) exporterMap.get(serviceKey); if (exporter == null) { throw new RemotingException(channel, "Not found exported service: " + serviceKey + " in " + exporterMap.keySet() + ", may be version or group mismatch " + - ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress() + ", message:" + getInvocationWithoutData(inv)); + ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress() + ", message:" + getInvocationWithoutData(inv)); } return exporter.getInvoker(); @@ -311,14 +307,14 @@ public class DubboProtocol extends AbstractProtocol { DubboExporter exporter = new DubboExporter(invoker, key, exporterMap); //export a stub service for dispatching event - Boolean isStubSupportEvent = url.getParameter(STUB_EVENT_KEY, DEFAULT_STUB_EVENT); - Boolean isCallbackservice = url.getParameter(IS_CALLBACK_SERVICE, false); - if (isStubSupportEvent && !isCallbackservice) { + boolean isStubSupportEvent = url.getParameter(STUB_EVENT_KEY, DEFAULT_STUB_EVENT); + boolean isCallbackService = url.getParameter(IS_CALLBACK_SERVICE, false); + if (isStubSupportEvent && !isCallbackService) { String stubServiceMethods = url.getParameter(STUB_EVENT_METHODS_KEY); if (stubServiceMethods == null || stubServiceMethods.length() == 0) { if (logger.isWarnEnabled()) { logger.warn(new IllegalStateException("consumer [" + url.getParameter(INTERFACE_KEY) + - "], has set stubproxy support event ,but no stub methods founded.")); + "], has set stub proxy support event ,but no stub methods founded.")); } } @@ -343,14 +339,13 @@ public class DubboProtocol extends AbstractProtocol { server = serverMap.get(key); if (server == null) { serverMap.put(key, createServer(url)); - }else { - server.reset(url); + return; } } - } else { - // server supports reset, use together with override - server.reset(url); } + + // server supports reset, use together with override + server.reset(url); } } @@ -362,16 +357,16 @@ public class DubboProtocol extends AbstractProtocol { private ProtocolServer createServer(URL url) { url = URLBuilder.from(url) - // send readonly event when server closes, it's enabled by default - .addParameterIfAbsent(CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString()) - // enable heartbeat by default - .addParameterIfAbsent(HEARTBEAT_KEY, String.valueOf(DEFAULT_HEARTBEAT)) - .addParameter(CODEC_KEY, DubboCodec.NAME) - .build(); - String str = url.getParameter(SERVER_KEY, DEFAULT_REMOTING_SERVER); + // send readonly event when server closes, it's enabled by default + .addParameterIfAbsent(CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString()) + // enable heartbeat by default + .addParameterIfAbsent(HEARTBEAT_KEY, String.valueOf(DEFAULT_HEARTBEAT)) + .addParameter(CODEC_KEY, DubboCodec.NAME) + .build(); - if (StringUtils.isNotEmpty(str) && !url.getOrDefaultFrameworkModel().getExtensionLoader(Transporter.class).hasExtension(str)) { - throw new RpcException("Unsupported server type: " + str + ", url: " + url); + String transporter = url.getParameter(SERVER_KEY, DEFAULT_REMOTING_SERVER); + if (StringUtils.isNotEmpty(transporter) && !url.getOrDefaultFrameworkModel().getExtensionLoader(Transporter.class).hasExtension(transporter)) { + throw new RpcException("Unsupported server type: " + transporter + ", url: " + url); } ExchangeServer server; @@ -381,12 +376,9 @@ public class DubboProtocol extends AbstractProtocol { throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e); } - str = url.getParameter(CLIENT_KEY); - if (StringUtils.isNotEmpty(str)) { - Set supportedTypes = url.getOrDefaultFrameworkModel().getExtensionLoader(Transporter.class).getSupportedExtensions(); - if (!supportedTypes.contains(str)) { - throw new RpcException("Unsupported client type: " + str); - } + transporter = url.getParameter(CLIENT_KEY); + if (StringUtils.isNotEmpty(transporter) && !url.getOrDefaultFrameworkModel().getExtensionLoader(Transporter.class).hasExtension(transporter)) { + throw new RpcException("Unsupported client type: " + transporter); } DubboProtocolServer protocolServer = new DubboProtocolServer(server); @@ -429,7 +421,7 @@ public class DubboProtocol extends AbstractProtocol { */ String shareConnectionsStr = url.getParameter(SHARE_CONNECTIONS_KEY, (String) null); connections = Integer.parseInt(StringUtils.isBlank(shareConnectionsStr) ? ConfigurationUtils.getProperty(url.getOrDefaultApplicationModel(), SHARE_CONNECTIONS_KEY, - DEFAULT_SHARE_CONNECTIONS) : shareConnectionsStr); + DEFAULT_SHARE_CONNECTIONS) : shareConnectionsStr); shareClients = getSharedClient(url, connections); } @@ -610,13 +602,13 @@ public class DubboProtocol extends AbstractProtocol { // BIO is not allowed since it has severe performance issue. if (StringUtils.isNotEmpty(str) && !url.getOrDefaultFrameworkModel().getExtensionLoader(Transporter.class).hasExtension(str)) { throw new RpcException("Unsupported client type: " + str + "," + - " supported client type is " + StringUtils.join(url.getOrDefaultFrameworkModel().getExtensionLoader(Transporter.class).getSupportedExtensions(), " ")); + " supported client type is " + StringUtils.join(url.getOrDefaultFrameworkModel().getExtensionLoader(Transporter.class).getSupportedExtensions(), " ")); } ExchangeClient client; try { // Replace InstanceAddressURL with ServiceConfigURL. - url = new ServiceConfigURL(DubboCodec.NAME, url.getUsername(), url.getPassword(), url.getHost(), url.getPort(), url.getPath(), url.getAllParameters()); + url = new ServiceConfigURL(DubboCodec.NAME, url.getUsername(), url.getPassword(), url.getHost(), url.getPort(), url.getPath(), url.getAllParameters()); url = url.addParameter(CODEC_KEY, DubboCodec.NAME); // enable heartbeat by default url = url.addParameterIfAbsent(HEARTBEAT_KEY, String.valueOf(DEFAULT_HEARTBEAT)); @@ -661,7 +653,7 @@ public class DubboProtocol extends AbstractProtocol { server.close(getServerShutdownTimeout(protocolServer)); } catch (Throwable t) { - logger.warn("Close dubbo server [" + server.getLocalAddress()+ "] failed: " + t.getMessage(), t); + logger.warn("Close dubbo server [" + server.getLocalAddress() + "] failed: " + t.getMessage(), t); } } serverMap.clear();