Skip to content

Commit 03520a3

Browse files
CodingSingerCodingSinger
committed
support apache dubbo asynchronous call #1114 
1 parent 74a40aa commit 03520a3

File tree

8 files changed

+321
-138
lines changed

8 files changed

+321
-138
lines changed

‎sentinel-adapter/sentinel-apache-dubbo-adapter/src/main/java/com/alibaba/csp/sentinel/adapter/dubbo/BaseSentinelDubboFilter.java

+41
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
package com.alibaba.csp.sentinel.adapter.dubbo;
2+
3+
4+
import com.alibaba.csp.sentinel.AsyncEntry;
5+
import com.alibaba.csp.sentinel.Entry;
6+
import com.alibaba.csp.sentinel.Tracer;
7+
import com.alibaba.csp.sentinel.context.ContextUtil;
8+
import org.apache.dubbo.rpc.Filter;
9+
import org.apache.dubbo.rpc.Invocation;
10+
import org.apache.dubbo.rpc.Invoker;
11+
import org.apache.dubbo.rpc.Result;
12+
import org.apache.dubbo.rpc.RpcContext;
13+
14+
public abstract class BaseSentinelDubboFilter implements Filter {
15+
16+
@Override
17+
public Result onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) {
18+
//it's unnecessary to trace the business exception
19+
trace(null, invocation);
20+
return appResponse;
21+
}
22+
23+
24+
static void trace(Throwable throwable, Invocation invocation) {
25+
Entry interfaceEntry = (Entry) RpcContext.getContext().get(DubboUtils.DUBBO_INTERFACE_ENTRY_KEY);
26+
Entry methodEntry = (Entry) RpcContext.getContext().get(DubboUtils.DUBBO_METHOD_ENTRY_KEY);
27+
if (methodEntry != null) {
28+
Tracer.traceEntry(throwable, methodEntry);
29+
methodEntry.exit();
30+
RpcContext.getContext().remove(DubboUtils.DUBBO_METHOD_ENTRY_KEY);
31+
}
32+
if (interfaceEntry != null) {
33+
Tracer.traceEntry(throwable, interfaceEntry);
34+
interfaceEntry.exit();
35+
RpcContext.getContext().remove(DubboUtils.DUBBO_INTERFACE_ENTRY_KEY);
36+
}
37+
if (!(interfaceEntry instanceof AsyncEntry)) {
38+
ContextUtil.exit();
39+
}
40+
}
41+
}

‎sentinel-adapter/sentinel-apache-dubbo-adapter/src/main/java/com/alibaba/csp/sentinel/adapter/dubbo/DubboUtils.java

+7-18
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,6 @@
1515
*/
1616
package com.alibaba.csp.sentinel.adapter.dubbo;
1717

18-
import com.alibaba.csp.sentinel.util.StringUtil;
1918
import org.apache.dubbo.rpc.Invocation;
2019
import org.apache.dubbo.rpc.Invoker;
2120

@@ -25,6 +24,8 @@
2524
public final class DubboUtils {
2625

2726
public static final String SENTINEL_DUBBO_APPLICATION_KEY = "dubboApplication";
27+
public static final String DUBBO_METHOD_ENTRY_KEY = "dubboMethodEntry";
28+
public static final String DUBBO_INTERFACE_ENTRY_KEY = "dubboInterfaceEntry";
2829

2930
public static String getApplication(Invocation invocation, String defaultValue) {
3031
if (invocation == null || invocation.getAttachments() == null) {
@@ -35,10 +36,10 @@ public static String getApplication(Invocation invocation, String defaultValue)
3536

3637
public static String getResourceName(Invoker<?> invoker, Invocation invocation) {
3738
StringBuilder buf = new StringBuilder(64);
38-
buf.append(invoker.getInterface().getName())
39-
.append(":")
40-
.append(invocation.getMethodName())
41-
.append("(");
39+
buf.append(invoker.getUrl().getEncodedServiceKey())
40+
.append(":")
41+
.append(invocation.getMethodName())
42+
.append("(");
4243
boolean isFirst = true;
4344
for (Class<?> clazz : invocation.getParameterTypes()) {
4445
if (!isFirst) {
@@ -51,17 +52,5 @@ public static String getResourceName(Invoker<?> invoker, Invocation invocation)
5152
return buf.toString();
5253
}
5354

54-
public static String getResourceName(Invoker<?> invoker, Invocation invocation, String prefix) {
55-
if (StringUtil.isNotBlank(prefix)) {
56-
return new StringBuilder(64)
57-
.append(prefix)
58-
.append(getResourceName(invoker, invocation))
59-
.toString();
60-
} else {
61-
return getResourceName(invoker, invocation);
62-
}
63-
}
64-
65-
private DubboUtils() {
66-
}
55+
private DubboUtils() {}
6756
}

‎sentinel-adapter/sentinel-apache-dubbo-adapter/src/main/java/com/alibaba/csp/sentinel/adapter/dubbo/SentinelDubboConsumerFilter.java

+29-22
Original file line numberDiff line numberDiff line change
@@ -18,17 +18,17 @@
1818
import com.alibaba.csp.sentinel.Entry;
1919
import com.alibaba.csp.sentinel.EntryType;
2020
import com.alibaba.csp.sentinel.SphU;
21-
import com.alibaba.csp.sentinel.Tracer;
22-
import com.alibaba.csp.sentinel.adapter.dubbo.config.DubboConfig;
2321
import com.alibaba.csp.sentinel.adapter.dubbo.fallback.DubboFallbackRegistry;
2422
import com.alibaba.csp.sentinel.log.RecordLog;
2523
import com.alibaba.csp.sentinel.slots.block.BlockException;
2624
import org.apache.dubbo.common.extension.Activate;
27-
import org.apache.dubbo.rpc.Filter;
25+
import org.apache.dubbo.rpc.AsyncRpcResult;
2826
import org.apache.dubbo.rpc.Invocation;
2927
import org.apache.dubbo.rpc.Invoker;
3028
import org.apache.dubbo.rpc.Result;
29+
import org.apache.dubbo.rpc.RpcContext;
3130
import org.apache.dubbo.rpc.RpcException;
31+
import org.apache.dubbo.rpc.support.RpcUtils;
3232

3333
/**
3434
* <p>Dubbo service consumer filter for Sentinel. Auto activated by default.</p>
@@ -42,7 +42,7 @@
4242
* @author Eric Zhao
4343
*/
4444
@Activate(group = "consumer")
45-
public class SentinelDubboConsumerFilter implements Filter {
45+
public class SentinelDubboConsumerFilter extends BaseSentinelDubboFilter {
4646

4747
public SentinelDubboConsumerFilter() {
4848
RecordLog.info("Sentinel Apache Dubbo consumer filter initialized");
@@ -52,32 +52,39 @@ public SentinelDubboConsumerFilter() {
5252
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
5353
Entry interfaceEntry = null;
5454
Entry methodEntry = null;
55+
RpcContext rpcContext = RpcContext.getContext();
5556
try {
56-
String resourceName = DubboUtils.getResourceName(invoker, invocation, DubboConfig.getDubboConsumerPrefix());
57-
interfaceEntry = SphU.entry(invoker.getInterface().getName(), EntryType.OUT);
58-
methodEntry = SphU.entry(resourceName, EntryType.OUT);
59-
57+
boolean isAsync = RpcUtils.isAsync(invoker.getUrl(), invocation);
58+
String resourceName = DubboUtils.getResourceName(invoker, invocation);
59+
if (!isAsync) {
60+
interfaceEntry = SphU.entry(invoker.getUrl().getEncodedServiceKey(), EntryType.OUT);
61+
methodEntry = SphU.entry(resourceName, EntryType.OUT);
62+
} else {
63+
// should generate the AsyncEntry when the invoke model in future or async
64+
interfaceEntry = SphU.asyncEntry(invoker.getUrl().getEncodedServiceKey(), EntryType.OUT);
65+
methodEntry = SphU.asyncEntry(resourceName, EntryType.OUT);
66+
}
67+
rpcContext.set(DubboUtils.DUBBO_INTERFACE_ENTRY_KEY, interfaceEntry);
68+
rpcContext.set(DubboUtils.DUBBO_METHOD_ENTRY_KEY, methodEntry);
6069
Result result = invoker.invoke(invocation);
61-
if (result.hasException()) {
62-
Throwable e = result.getException();
63-
// Record common exception.
64-
Tracer.traceEntry(e, interfaceEntry);
65-
Tracer.traceEntry(e, methodEntry);
70+
if (result instanceof AsyncRpcResult) {
71+
// catch timeout or nonbiz-exception when in async model
72+
AsyncRpcResult asyncRpcResult = (AsyncRpcResult) result;
73+
asyncRpcResult.getValueFuture().whenComplete((rs,ex) -> {
74+
if (ex != null){
75+
trace((Throwable) ex, invocation);
76+
}
77+
});
6678
}
6779
return result;
6880
} catch (BlockException e) {
6981
return DubboFallbackRegistry.getConsumerFallback().handle(invoker, invocation, e);
7082
} catch (RpcException e) {
71-
Tracer.traceEntry(e, interfaceEntry);
72-
Tracer.traceEntry(e, methodEntry);
83+
// catch timeout or nonbiz-exception when in sync model
84+
trace(e, invocation);
7385
throw e;
74-
} finally {
75-
if (methodEntry != null) {
76-
methodEntry.exit();
77-
}
78-
if (interfaceEntry != null) {
79-
interfaceEntry.exit();
80-
}
8186
}
8287
}
88+
89+
8390
}

‎sentinel-adapter/sentinel-apache-dubbo-adapter/src/main/java/com/alibaba/csp/sentinel/adapter/dubbo/SentinelDubboProviderFilter.java

+12-26
Original file line numberDiff line numberDiff line change
@@ -18,17 +18,15 @@
1818
import com.alibaba.csp.sentinel.Entry;
1919
import com.alibaba.csp.sentinel.EntryType;
2020
import com.alibaba.csp.sentinel.SphU;
21-
import com.alibaba.csp.sentinel.Tracer;
22-
import com.alibaba.csp.sentinel.adapter.dubbo.config.DubboConfig;
2321
import com.alibaba.csp.sentinel.adapter.dubbo.fallback.DubboFallbackRegistry;
2422
import com.alibaba.csp.sentinel.context.ContextUtil;
2523
import com.alibaba.csp.sentinel.log.RecordLog;
2624
import com.alibaba.csp.sentinel.slots.block.BlockException;
2725
import org.apache.dubbo.common.extension.Activate;
28-
import org.apache.dubbo.rpc.Filter;
2926
import org.apache.dubbo.rpc.Invocation;
3027
import org.apache.dubbo.rpc.Invoker;
3128
import org.apache.dubbo.rpc.Result;
29+
import org.apache.dubbo.rpc.RpcContext;
3230
import org.apache.dubbo.rpc.RpcException;
3331

3432
/**
@@ -44,7 +42,7 @@
4442
* @author Eric Zhao
4543
*/
4644
@Activate(group = "provider")
47-
public class SentinelDubboProviderFilter implements Filter {
45+
public class SentinelDubboProviderFilter extends BaseSentinelDubboFilter {
4846

4947
public SentinelDubboProviderFilter() {
5048
RecordLog.info("Sentinel Apache Dubbo provider filter initialized");
@@ -54,40 +52,28 @@ public SentinelDubboProviderFilter() {
5452
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
5553
// Get origin caller.
5654
String application = DubboUtils.getApplication(invocation, "");
57-
55+
RpcContext rpcContext = RpcContext.getContext();
5856
Entry interfaceEntry = null;
5957
Entry methodEntry = null;
6058
try {
61-
String resourceName = DubboUtils.getResourceName(invoker, invocation, DubboConfig.getDubboProviderPrefix());
62-
String interfaceName = invoker.getInterface().getName();
59+
String resourceName = DubboUtils.getResourceName(invoker, invocation);
60+
String interfaceName = invoker.getUrl().getEncodedServiceKey();
6361
// Only need to create entrance context at provider side, as context will take effect
6462
// at entrance of invocation chain only (for inbound traffic).
6563
ContextUtil.enter(resourceName, application);
6664
interfaceEntry = SphU.entry(interfaceName, EntryType.IN);
6765
methodEntry = SphU.entry(resourceName, EntryType.IN, 1, invocation.getArguments());
68-
69-
Result result = invoker.invoke(invocation);
70-
if (result.hasException()) {
71-
Throwable e = result.getException();
72-
// Record common exception.
73-
Tracer.traceEntry(e, interfaceEntry);
74-
Tracer.traceEntry(e, methodEntry);
75-
}
76-
return result;
66+
rpcContext.set(DubboUtils.DUBBO_INTERFACE_ENTRY_KEY, interfaceEntry);
67+
rpcContext.set(DubboUtils.DUBBO_METHOD_ENTRY_KEY, methodEntry);
68+
return invoker.invoke(invocation);
7769
} catch (BlockException e) {
7870
return DubboFallbackRegistry.getProviderFallback().handle(invoker, invocation, e);
7971
} catch (RpcException e) {
80-
Tracer.traceEntry(e, interfaceEntry);
81-
Tracer.traceEntry(e, methodEntry);
72+
trace(e, invocation);
8273
throw e;
83-
} finally {
84-
if (methodEntry != null) {
85-
methodEntry.exit(1, invocation.getArguments());
86-
}
87-
if (interfaceEntry != null) {
88-
interfaceEntry.exit();
89-
}
90-
ContextUtil.exit();
9174
}
9275
}
76+
77+
9378
}
79+

‎sentinel-adapter/sentinel-apache-dubbo-adapter/src/test/java/com/alibaba/csp/sentinel/BaseTest.java

+15-3
Original file line numberDiff line numberDiff line change
@@ -15,10 +15,14 @@
1515
*/
1616
package com.alibaba.csp.sentinel;
1717

18+
import com.alibaba.csp.sentinel.context.ContextUtil;
1819
import com.alibaba.csp.sentinel.slots.clusterbuilder.ClusterBuilderSlot;
1920

2021
import org.apache.dubbo.rpc.RpcContext;
2122

23+
import java.lang.reflect.InvocationTargetException;
24+
import java.lang.reflect.Method;
25+
2226
/**
2327
* Base test class, provide common methods for subClass
2428
* The package is same as CtSph, to call CtSph.resetChainMap() method for test
@@ -33,8 +37,16 @@ public class BaseTest {
3337
* Clean up resources for context, clusterNodeMap, processorSlotChainMap
3438
*/
3539
protected static void cleanUpAll() {
36-
RpcContext.removeContext();
37-
ClusterBuilderSlot.getClusterNodeMap().clear();
38-
CtSph.resetChainMap();
40+
try {
41+
RpcContext.removeContext();
42+
ClusterBuilderSlot.getClusterNodeMap().clear();
43+
CtSph.resetChainMap();
44+
Method method = ContextUtil.class.getDeclaredMethod("resetContextMap");
45+
method.setAccessible(true);
46+
method.invoke(null, null);
47+
ContextUtil.exit();
48+
} catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
49+
e.printStackTrace();
50+
}
3951
}
4052
}

0 commit comments

Comments
 (0)