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

Merge branch 'apache-3.2' into apache-3.3

This commit is contained in:
Albumen Kevin 2023-07-03 16:07:41 +08:00
parent 0d4e8304bc 5f83f10eac
commit 9bbf7625a8
20 changed files with 213 additions and 118 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvoker.java
View File

@ -379,7 +379,7 @@ public abstract class AbstractClusterInvoker<T> implements ClusterInvoker<T> {
if (ProfilerSwitch.isEnableSimpleProfiler()) {
InvocationProfilerUtils.enterProfiler(invocation, "Invoker invoke. Target Address: " + invoker.getUrl().getAddress());
}
invocation.addInvokedInvoker(invoker);
setRemote(invoker, invocation);
result = invoker.invoke(invocation);
} finally {
clearContext(originInvoker);
@ -388,6 +388,17 @@ public abstract class AbstractClusterInvoker<T> implements ClusterInvoker<T> {
return result;
}
/**
* Set the remoteAddress and remoteApplicationName so that filter can get them.
*
*/
private void setRemote(Invoker<?> invoker, Invocation invocation) {
invocation.addInvokedInvoker(invoker);
RpcServiceContext serviceContext = RpcContext.getServiceContext();
serviceContext.setRemoteAddress(invoker.getUrl().toInetSocketAddress());
serviceContext.setRemoteApplicationName(invoker.getUrl().getRemoteApplication());
}
/**
* When using a thread pool to fork a child thread, ThreadLocal cannot be passed.
* In this scenario, please use the invokeWithContextAsync method.

29
dubbo-common/src/main/java/org/apache/dubbo/common/config/ConfigurationUtils.java
View File

@ -60,6 +60,8 @@ public final class ConfigurationUtils {
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(ConfigurationUtils.class);
private static final List<String> securityKey;
private static volatile long expectedShutdownTime = Long.MAX_VALUE;
static {
List<String> keys = new LinkedList<>();
keys.add("accesslog");
@ -113,6 +115,9 @@ public final class ConfigurationUtils {
*/
@SuppressWarnings("deprecation")
public static int getServerShutdownTimeout(ScopeModel scopeModel) {
if (expectedShutdownTime < System.currentTimeMillis()) {
return 1;
}
int timeout = DEFAULT_SERVER_SHUTDOWN_TIMEOUT;
Configuration configuration = getGlobalConfiguration(scopeModel);
String value = StringUtils.trim(configuration.getString(SHUTDOWN_WAIT_KEY));
@ -133,9 +138,33 @@ public final class ConfigurationUtils {
}
}
}
if (expectedShutdownTime - System.currentTimeMillis() < timeout) {
return (int) Math.max(1, expectedShutdownTime - System.currentTimeMillis());
}
return timeout;
}
public static int reCalShutdownTime(int expected) {
// already timeout
if (expectedShutdownTime < System.currentTimeMillis()) {
return 1;
}
if (expectedShutdownTime - System.currentTimeMillis() < expected) {
// the shutdown time rest is less than expected
return (int) Math.max(1, expectedShutdownTime - System.currentTimeMillis());
}
// return the expected
return expected;
}
public static void setExpectedShutdownTime(long expectedShutdownTime) {
ConfigurationUtils.expectedShutdownTime = expectedShutdownTime;
}
public static String getCachedDynamicProperty(ScopeModel realScopeModel, String key, String defaultValue) {
ScopeModel scopeModel = getScopeModelOrDefaultApplicationModel(realScopeModel);
ConfigurationCache configurationCache = scopeModel.getBeanFactory().getBean(ConfigurationCache.class);

4
dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/DubboShutdownHook.java
View File

@ -82,6 +82,9 @@ public class DubboShutdownHook extends Thread {
}
private void doDestroy() {
int timeout = ConfigurationUtils.getServerShutdownTimeout(applicationModel);
ConfigurationUtils.setExpectedShutdownTime(System.currentTimeMillis() + timeout);
// send readonly for shutdown hook
List<GracefulShutdown> gracefulShutdowns = GracefulShutdown.getGracefulShutdowns(applicationModel.getFrameworkModel());
for (GracefulShutdown gracefulShutdown : gracefulShutdowns) {
@ -97,7 +100,6 @@ public class DubboShutdownHook extends Thread {
}
}
if (hasModuleBindSpring) {
int timeout = ConfigurationUtils.getServerShutdownTimeout(applicationModel);
if (timeout > 0) {
long start = System.currentTimeMillis();
/*

105
dubbo-plugin/dubbo-plugin-context/src/main/java/org/apache/dubbo/rpc/cluster/filter/support/ConsumerContextFilter.java
View File

@ -62,63 +62,58 @@ public class ConsumerContextFilter implements ClusterFilter, ClusterFilter.Liste
@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
RpcContext.RestoreServiceContext originServiceContext = RpcContext.storeServiceContext();
try {
RpcContext.getServiceContext()
.setInvoker(invoker)
.setInvocation(invocation)
.setLocalAddress(NetUtils.getLocalHost(), 0);
RpcContext.getServiceContext()
.setInvoker(invoker)
.setInvocation(invocation)
.setLocalAddress(NetUtils.getLocalHost(), 0);
RpcContext context = RpcContext.getClientAttachment();
context.setAttachment(REMOTE_APPLICATION_KEY, invoker.getUrl().getApplication());
if (invocation instanceof RpcInvocation) {
((RpcInvocation) invocation).setInvoker(invoker);
}
if (CollectionUtils.isNotEmpty(supportedSelectors)) {
for (PenetrateAttachmentSelector supportedSelector : supportedSelectors) {
Map<String, Object> selected = supportedSelector.select(invocation, RpcContext.getClientAttachment(), RpcContext.getServerAttachment());
if (CollectionUtils.isNotEmptyMap(selected)) {
((RpcInvocation) invocation).addObjectAttachments(selected);
}
}
} else {
((RpcInvocation) invocation).addObjectAttachments(RpcContext.getServerAttachment().getObjectAttachments());
}
Map<String, Object> contextAttachments = RpcContext.getClientAttachment().getObjectAttachments();
if (CollectionUtils.isNotEmptyMap(contextAttachments)) {
/**
* invocation.addAttachmentsIfAbsent(context){@link RpcInvocation#addAttachmentsIfAbsent(Map)}should not be used here,
* because the {@link RpcContext#setAttachment(String, String)} is passed in the Filter when the call is triggered
* by the built-in retry mechanism of the Dubbo. The attachment to update RpcContext will no longer work, which is
* a mistake in most cases (for example, through Filter to RpcContext output traceId and spanId and other information).
*/
((RpcInvocation) invocation).addObjectAttachments(contextAttachments);
}
// pass default timeout set by end user (ReferenceConfig)
Object countDown = RpcContext.getServerAttachment().getObjectAttachment(TIME_COUNTDOWN_KEY);
if (countDown != null) {
String methodName = RpcUtils.getMethodName(invocation);
// When the client has enabled the timeout-countdown function,
// the subsequent calls launched by the Server side will be enabled by default,
// and support to turn off the function on a node to get rid of the timeout control.
if (invoker.getUrl().getMethodParameter(methodName, ENABLE_TIMEOUT_COUNTDOWN_KEY, true)) {
context.setObjectAttachment(TIME_COUNTDOWN_KEY, countDown);
TimeoutCountDown timeoutCountDown = (TimeoutCountDown) countDown;
if (timeoutCountDown.isExpired()) {
return AsyncRpcResult.newDefaultAsyncResult(new RpcException(RpcException.TIMEOUT_TERMINATE,
"No time left for making the following call: " + invocation.getServiceName() + "."
+ RpcUtils.getMethodName(invocation) + ", terminate directly."), invocation);
}
}
}
RpcContext.removeClientResponseContext();
return invoker.invoke(invocation);
} finally {
RpcContext.restoreServiceContext(originServiceContext);
RpcContext context = RpcContext.getClientAttachment();
context.setAttachment(REMOTE_APPLICATION_KEY, invoker.getUrl().getApplication());
if (invocation instanceof RpcInvocation) {
((RpcInvocation) invocation).setInvoker(invoker);
}
if (CollectionUtils.isNotEmpty(supportedSelectors)) {
for (PenetrateAttachmentSelector supportedSelector : supportedSelectors) {
Map<String, Object> selected = supportedSelector.select(invocation, RpcContext.getClientAttachment(), RpcContext.getServerAttachment());
if (CollectionUtils.isNotEmptyMap(selected)) {
((RpcInvocation) invocation).addObjectAttachments(selected);
}
}
} else {
((RpcInvocation) invocation).addObjectAttachments(RpcContext.getServerAttachment().getObjectAttachments());
}
Map<String, Object> contextAttachments = RpcContext.getClientAttachment().getObjectAttachments();
if (CollectionUtils.isNotEmptyMap(contextAttachments)) {
/**
* invocation.addAttachmentsIfAbsent(context){@link RpcInvocation#addAttachmentsIfAbsent(Map)}should not be used here,
* because the {@link RpcContext#setAttachment(String, String)} is passed in the Filter when the call is triggered
* by the built-in retry mechanism of the Dubbo. The attachment to update RpcContext will no longer work, which is
* a mistake in most cases (for example, through Filter to RpcContext output traceId and spanId and other information).
*/
((RpcInvocation) invocation).addObjectAttachments(contextAttachments);
}
// pass default timeout set by end user (ReferenceConfig)
Object countDown = RpcContext.getServerAttachment().getObjectAttachment(TIME_COUNTDOWN_KEY);
if (countDown != null) {
String methodName = RpcUtils.getMethodName(invocation);
// When the client has enabled the timeout-countdown function,
// the subsequent calls launched by the Server side will be enabled by default,
// and support to turn off the function on a node to get rid of the timeout control.
if (invoker.getUrl().getMethodParameter(methodName, ENABLE_TIMEOUT_COUNTDOWN_KEY, true)) {
context.setObjectAttachment(TIME_COUNTDOWN_KEY, countDown);
TimeoutCountDown timeoutCountDown = (TimeoutCountDown) countDown;
if (timeoutCountDown.isExpired()) {
return AsyncRpcResult.newDefaultAsyncResult(new RpcException(RpcException.TIMEOUT_TERMINATE,
"No time left for making the following call: " + invocation.getServiceName() + "."
+ RpcUtils.getMethodName(invocation) + ", terminate directly."), invocation);
}
}
}
RpcContext.removeClientResponseContext();
return invoker.invoke(invocation);
}
@Override

2
dubbo-plugin/dubbo-plugin-loadbalance-adaptive/src/main/java/org/apache/dubbo/rpc/cluster/loadbalance/AdaptiveLoadBalance.java
View File

@ -110,7 +110,7 @@ public class AdaptiveLoadBalance extends AbstractLoadBalance {
private <T> Invoker<T> chooseLowLoadInvoker(Invoker<T> invoker1,Invoker<T> invoker2,Invocation invocation){
int weight1 = getWeight(invoker1, invocation);
int weight2 = getWeight(invoker2, invocation);
int timeout1 = getTimeout(invoker2, invocation);
int timeout1 = getTimeout(invoker1, invocation);
int timeout2 = getTimeout(invoker2, invocation);
long load1 = Double.doubleToLongBits(adaptiveMetrics.getLoad(getServiceKey(invoker1,invocation),weight1,timeout1 ));
long load2 = Double.doubleToLongBits(adaptiveMetrics.getLoad(getServiceKey(invoker2,invocation),weight2,timeout2 ));

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

@ -66,6 +66,13 @@ public interface Constants {
*/
long LEAST_HEARTBEAT_DURATION = 1000;
/**
* the least reconnect during is 60000 ms.
*/
long LEAST_RECONNECT_DURATION = 60000;
String LEAST_RECONNECT_DURATION_KEY = "dubbo.application.least-reconnect-duration";
/**
* ticks per wheel.
*/

4
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeChannel.java
View File

@ -50,7 +50,7 @@ final class HeaderExchangeChannel implements ExchangeChannel {
private final Channel channel;
private final long shutdownTimeout;
private final int shutdownTimeout;
private volatile boolean closed = false;
@ -167,7 +167,7 @@ final class HeaderExchangeChannel implements ExchangeChannel {
closed = true;
try {
// graceful close
DefaultFuture.closeChannel(channel, shutdownTimeout);
DefaultFuture.closeChannel(channel, ConfigurationUtils.reCalShutdownTime(shutdownTimeout));
} catch (Exception e) {
logger.warn(TRANSPORT_FAILED_CLOSE, "", "", e.getMessage(), e);
}

10
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeClient.java
View File

@ -37,6 +37,8 @@ import java.util.concurrent.TimeUnit;
import static org.apache.dubbo.remoting.Constants.HEARTBEAT_CHECK_TICK;
import static org.apache.dubbo.remoting.Constants.LEAST_HEARTBEAT_DURATION;
import static org.apache.dubbo.remoting.Constants.LEAST_RECONNECT_DURATION;
import static org.apache.dubbo.remoting.Constants.LEAST_RECONNECT_DURATION_KEY;
import static org.apache.dubbo.remoting.Constants.TICKS_PER_WHEEL;
import static org.apache.dubbo.remoting.utils.UrlUtils.getHeartbeat;
import static org.apache.dubbo.remoting.utils.UrlUtils.getIdleTimeout;
@ -214,7 +216,8 @@ public class HeaderExchangeClient implements ExchangeClient {
private void startReconnectTask(URL url) {
if (shouldReconnect(url)) {
long heartbeatTimeoutTick = calculateLeastDuration(idleTimeout);
reconnectTimerTask = new ReconnectTimerTask(() -> Collections.singleton(this), IDLE_CHECK_TIMER.get(), heartbeatTimeoutTick, idleTimeout);
reconnectTimerTask = new ReconnectTimerTask(() -> Collections.singleton(this), IDLE_CHECK_TIMER.get(),
calculateReconnectDuration(url, heartbeatTimeoutTick), idleTimeout);
}
}
@ -240,6 +243,11 @@ public class HeaderExchangeClient implements ExchangeClient {
}
}
private long calculateReconnectDuration(URL url, long tick) {
long leastReconnectDuration = url.getParameter(LEAST_RECONNECT_DURATION_KEY, LEAST_RECONNECT_DURATION);
return Math.max(leastReconnectDuration, tick);
}
private boolean shouldReconnect(URL url) {
return url.getParameter(Constants.RECONNECT_KEY, true);
}

2
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeHandler.java
View File

@ -148,7 +148,7 @@ public class HeaderExchangeHandler implements ChannelHandlerDelegate {
if (timeoutObj instanceof Integer) {
shutdownTimeout = (Integer) timeoutObj;
}
DefaultFuture.closeChannel(channel, shutdownTimeout);
DefaultFuture.closeChannel(channel, ConfigurationUtils.reCalShutdownTime(shutdownTimeout));
HeaderExchangeChannel.removeChannel(channel);
}
}

4
dubbo-remoting/dubbo-remoting-netty/src/test/java/org/apache/dubbo/remoting/transport/netty/ClientReconnectTest.java
View File

@ -29,11 +29,13 @@ import org.apache.dubbo.remoting.exchange.Exchangers;
import org.apache.dubbo.remoting.exchange.support.ExchangeHandlerAdapter;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.model.FrameworkModel;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.apache.dubbo.common.constants.CommonConstants.EXECUTOR_MANAGEMENT_MODE_DEFAULT;
import static org.apache.dubbo.remoting.Constants.LEAST_RECONNECT_DURATION_KEY;
/**
* Client reconnect test
@ -76,7 +78,7 @@ class ClientReconnectTest {
public Client startClient(int port, int heartbeat) throws RemotingException {
URL url = URL.valueOf("exchange://127.0.0.1:" + port + "/client.reconnect.test?check=false&codec=exchange&client=netty3&" +
Constants.HEARTBEAT_KEY + "=" + heartbeat);
Constants.HEARTBEAT_KEY + "=" + heartbeat + "&" + LEAST_RECONNECT_DURATION_KEY + "=0");
ApplicationModel applicationModel = ApplicationModel.defaultModel();
ApplicationConfig applicationConfig = new ApplicationConfig("provider-app");
applicationConfig.setExecutorManagementMode(EXECUTOR_MANAGEMENT_MODE_DEFAULT);

2
dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyPortUnificationServer.java
View File

@ -178,7 +178,7 @@ public class NettyPortUnificationServer extends AbstractPortUnificationServer {
try {
if (bootstrap != null) {
long timeout = serverShutdownTimeoutMills;
long timeout = ConfigurationUtils.reCalShutdownTime(serverShutdownTimeoutMills);
long quietPeriod = Math.min(2000L, timeout);
Future<?> bossGroupShutdownFuture = bossGroup.shutdownGracefully(quietPeriod,
timeout, MILLISECONDS);

2
dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyServer.java
View File

@ -174,7 +174,7 @@ public class NettyServer extends AbstractServer {
}
try {
if (bootstrap != null) {
long timeout = serverShutdownTimeoutMills;
long timeout = ConfigurationUtils.reCalShutdownTime(serverShutdownTimeoutMills);
long quietPeriod = Math.min(2000L, timeout);
Future<?> bossGroupShutdownFuture = bossGroup.shutdownGracefully(quietPeriod, timeout, MILLISECONDS);
Future<?> workerGroupShutdownFuture = workerGroup.shutdownGracefully(quietPeriod, timeout, MILLISECONDS);

5
dubbo-remoting/dubbo-remoting-netty4/src/test/java/org/apache/dubbo/remoting/transport/netty4/ClientReconnectTest.java
View File

@ -32,11 +32,13 @@ import org.apache.dubbo.remoting.exchange.support.ExchangeHandlerAdapter;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.model.FrameworkModel;
import org.apache.dubbo.rpc.model.ModuleModel;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.apache.dubbo.common.constants.CommonConstants.EXECUTOR_MANAGEMENT_MODE_DEFAULT;
import static org.apache.dubbo.remoting.Constants.LEAST_RECONNECT_DURATION_KEY;
/**
* Client reconnect test
@ -81,7 +83,8 @@ class ClientReconnectTest {
public Client startClient(int port, int heartbeat) throws RemotingException {
URL url = URL.valueOf("exchange://127.0.0.1:" + port + "/client.reconnect.test?client=netty4&check=false&" + Constants.HEARTBEAT_KEY + "=" + heartbeat);
URL url = URL.valueOf("exchange://127.0.0.1:" + port + "/client.reconnect.test?client=netty4&check=false&" +
Constants.HEARTBEAT_KEY + "=" + heartbeat + "&" + LEAST_RECONNECT_DURATION_KEY + "=0");
FrameworkModel frameworkModel = new FrameworkModel();
ApplicationModel applicationModel = frameworkModel.newApplication();
ApplicationConfig applicationConfig = new ApplicationConfig("provider-app");

12
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/ReferenceCountInvokerWrapper.java
View File

@ -17,6 +17,7 @@
package org.apache.dubbo.rpc.protocol;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.config.ConfigurationUtils;
import org.apache.dubbo.common.constants.LoggerCodeConstants;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
@ -25,6 +26,7 @@ import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@ -53,8 +55,16 @@ public class ReferenceCountInvokerWrapper<T> implements Invoker<T> {
@Override
public void destroy() {
try {
lock.writeLock().lock();
int timeout = ConfigurationUtils.getServerShutdownTimeout(invoker.getUrl().getScopeModel());
boolean locked = lock.writeLock().tryLock(
timeout, TimeUnit.MILLISECONDS);
if (!locked) {
logger.warn(LoggerCodeConstants.PROTOCOL_CLOSED_SERVER, "", "",
"Failed to wait for invocation end in " + timeout + "ms.");
}
destroyed.set(true);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
lock.writeLock().unlock();
}

87
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/InvocationUtil.java
View File

@ -24,6 +24,7 @@ import org.apache.dubbo.common.profiler.Profiler;
import org.apache.dubbo.common.profiler.ProfilerEntry;
import org.apache.dubbo.common.profiler.ProfilerSwitch;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcInvocation;
import org.apache.dubbo.rpc.RpcServiceContext;
import org.apache.dubbo.rpc.support.RpcUtils;
@ -36,53 +37,59 @@ public class InvocationUtil {
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(InvokerInvocationHandler.class);
public static Object invoke(Invoker<?> invoker, RpcInvocation rpcInvocation) throws Throwable {
URL url = invoker.getUrl();
String serviceKey = url.getServiceKey();
rpcInvocation.setTargetServiceUniqueName(serviceKey);
RpcContext.RestoreServiceContext originServiceContext = RpcContext.storeServiceContext();
// invoker.getUrl() returns consumer url.
RpcServiceContext.getServiceContext().setConsumerUrl(url);
try {
URL url = invoker.getUrl();
String serviceKey = url.getServiceKey();
rpcInvocation.setTargetServiceUniqueName(serviceKey);
if (ProfilerSwitch.isEnableSimpleProfiler()) {
ProfilerEntry parentProfiler = Profiler.getBizProfiler();
ProfilerEntry bizProfiler;
if (parentProfiler != null) {
bizProfiler = Profiler.enter(parentProfiler,
"Receive request. Client invoke begin. ServiceKey: " + serviceKey + " MethodName:" + rpcInvocation.getMethodName());
} else {
bizProfiler = Profiler.start("Receive request. Client invoke begin. ServiceKey: " + serviceKey + " " + "MethodName:" + rpcInvocation.getMethodName());
}
rpcInvocation.put(Profiler.PROFILER_KEY, bizProfiler);
try {
return invoker.invoke(rpcInvocation).recreate();
} finally {
Profiler.release(bizProfiler);
Long timeout = RpcUtils.convertToNumber(rpcInvocation.getObjectAttachmentWithoutConvert(TIMEOUT_KEY));
// invoker.getUrl() returns consumer url.
RpcServiceContext.getServiceContext().setConsumerUrl(url);
if (timeout == null) {
timeout = (long) url.getMethodPositiveParameter(rpcInvocation.getMethodName(),
TIMEOUT_KEY,
DEFAULT_TIMEOUT);
if (ProfilerSwitch.isEnableSimpleProfiler()) {
ProfilerEntry parentProfiler = Profiler.getBizProfiler();
ProfilerEntry bizProfiler;
if (parentProfiler != null) {
bizProfiler = Profiler.enter(parentProfiler,
"Receive request. Client invoke begin. ServiceKey: " + serviceKey + " MethodName:" + rpcInvocation.getMethodName());
} else {
bizProfiler = Profiler.start("Receive request. Client invoke begin. ServiceKey: " + serviceKey + " " + "MethodName:" + rpcInvocation.getMethodName());
}
long usage = bizProfiler.getEndTime() - bizProfiler.getStartTime();
if ((usage / (1000_000L * ProfilerSwitch.getWarnPercent())) > timeout) {
StringBuilder attachment = new StringBuilder();
rpcInvocation.foreachAttachment((entry) -> {
attachment.append(entry.getKey()).append("=").append(entry.getValue()).append(";\n");
});
rpcInvocation.put(Profiler.PROFILER_KEY, bizProfiler);
try {
return invoker.invoke(rpcInvocation).recreate();
} finally {
Profiler.release(bizProfiler);
Long timeout = RpcUtils.convertToNumber(rpcInvocation.getObjectAttachmentWithoutConvert(TIMEOUT_KEY));
logger.warn(PROXY_TIMEOUT_REQUEST, "", "", String.format(
"[Dubbo-Consumer] execute service %s#%s cost %d.%06d ms, this invocation almost (maybe already) timeout. Timeout: %dms\n" + "invocation context:\n%s" + "thread info: \n%s",
rpcInvocation.getProtocolServiceKey(),
rpcInvocation.getMethodName(),
usage / 1000_000,
usage % 1000_000,
timeout,
attachment,
Profiler.buildDetail(bizProfiler)));
if (timeout == null) {
timeout = (long) url.getMethodPositiveParameter(rpcInvocation.getMethodName(),
TIMEOUT_KEY,
DEFAULT_TIMEOUT);
}
long usage = bizProfiler.getEndTime() - bizProfiler.getStartTime();
if ((usage / (1000_000L * ProfilerSwitch.getWarnPercent())) > timeout) {
StringBuilder attachment = new StringBuilder();
rpcInvocation.foreachAttachment((entry) -> {
attachment.append(entry.getKey()).append("=").append(entry.getValue()).append(";\n");
});
logger.warn(PROXY_TIMEOUT_REQUEST, "", "", String.format(
"[Dubbo-Consumer] execute service %s#%s cost %d.%06d ms, this invocation almost (maybe already) timeout. Timeout: %dms\n" + "invocation context:\n%s" + "thread info: \n%s",
rpcInvocation.getProtocolServiceKey(),
rpcInvocation.getMethodName(),
usage / 1000_000,
usage % 1000_000,
timeout,
attachment,
Profiler.buildDetail(bizProfiler)));
}
}
}
return invoker.invoke(rpcInvocation).recreate();
} finally {
RpcContext.restoreServiceContext(originServiceContext);
}
return invoker.invoke(rpcInvocation).recreate();
}
}

8
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/ByteAccessor.java
View File

@ -20,6 +20,8 @@ package org.apache.dubbo.rpc.protocol.dubbo;
import org.apache.dubbo.common.extension.SPI;
import org.apache.dubbo.remoting.Channel;
import org.apache.dubbo.remoting.exchange.Request;
import org.apache.dubbo.remoting.exchange.Response;
import org.apache.dubbo.rpc.Invocation;
import java.io.InputStream;
@ -38,4 +40,10 @@ public interface ByteAccessor {
* The parameters are the same as {@link DecodeableRpcInvocation}
*/
DecodeableRpcInvocation getRpcInvocation(Channel channel, Request req, InputStream is, byte proto);
/**
* Get an enhanced DecodeableRpcResult subclass to allow custom decode.
* The parameters are the same as {@link DecodeableRpcResult}
*/
DecodeableRpcResult getRpcResult(Channel channel, Response res, InputStream is, Invocation invocation, byte proto);
}

25
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCodec.java
View File

@ -114,14 +114,29 @@ public class DubboCodec extends ExchangeCodec {
}
} else {
DecodeableRpcResult result;
Invocation inv = (Invocation) getRequestData(channel, res, id);
if (channel.getUrl().getParameter(DECODE_IN_IO_THREAD_KEY, DEFAULT_DECODE_IN_IO_THREAD)) {
result = new DecodeableRpcResult(channel, res, is,
(Invocation) getRequestData(channel, res, id), proto);
if (customByteAccessor != null) {
result = customByteAccessor.getRpcResult(channel, res,
new UnsafeByteArrayInputStream(readMessageData(is)),
inv, proto);
} else {
result = new DecodeableRpcResult(channel, res,
new UnsafeByteArrayInputStream(readMessageData(is)),
inv, proto);
}
result.decode();
} else {
result = new DecodeableRpcResult(channel, res,
new UnsafeByteArrayInputStream(readMessageData(is)),
(Invocation) getRequestData(channel, res, id), proto);
if (customByteAccessor != null) {
result = customByteAccessor.getRpcResult(channel, res,
new UnsafeByteArrayInputStream(readMessageData(is)),
inv, proto);
} else {
result = new DecodeableRpcResult(channel, res,
new UnsafeByteArrayInputStream(readMessageData(is)),
inv, proto);
}
}
data = result;
}

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

@ -34,7 +34,6 @@ import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.InvokeMode;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.RpcInvocation;
import org.apache.dubbo.rpc.protocol.AbstractInvoker;
@ -111,8 +110,6 @@ public class DubboInvoker<T> extends AbstractInvoker<T> {
invocation.setAttachment(TIMEOUT_KEY, String.valueOf(timeout));
RpcContext.getServiceContext().setRemoteAddress(currentClient.getRemoteAddress());
Integer payload = getUrl().getParameter(PAYLOAD, Integer.class);
Request request = new Request();
@ -182,7 +179,7 @@ public class DubboInvoker<T> extends AbstractInvoker<T> {
if (invokers != null) {
invokers.remove(this);
}
clientsProvider.close(serverShutdownTimeout);
clientsProvider.close(ConfigurationUtils.reCalShutdownTime(serverShutdownTimeout));
} finally {
destroyLock.unlock();
}

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

@ -554,7 +554,7 @@ public class DubboProtocol extends AbstractProtocol {
logger.info("Closing dubbo server: " + server.getLocalAddress());
}
server.close(getServerShutdownTimeout(protocolServer));
server.close(ConfigurationUtils.reCalShutdownTime(getServerShutdownTimeout(protocolServer)));
} catch (Throwable t) {
logger.warn(PROTOCOL_ERROR_CLOSE_SERVER, "", "", "Close dubbo server [" + server.getLocalAddress() + "] failed: " + t.getMessage(), t);

3
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/SharedClientsProvider.java
View File

@ -16,6 +16,7 @@
*/
package org.apache.dubbo.rpc.protocol.dubbo;
import org.apache.dubbo.common.config.ConfigurationUtils;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CollectionUtils;
@ -116,7 +117,7 @@ public class SharedClientsProvider implements ClientsProvider {
logger.info("Close dubbo connect: " + client.getLocalAddress() + "-->" + client.getRemoteAddress());
}
client.close(client.getShutdownWaitTime());
client.close(ConfigurationUtils.reCalShutdownTime(client.getShutdownWaitTime()));
// TODO
/*