赞
踩
dubbo 对callback的说明在 https://cn.dubbo.apache.org/zh-cn/overview/mannual/java-sdk/advanced-features-and-usage/service/callback-parameter/
关键内容如下:
特性说明
参数回调方式与调用本地 callback 或 listener 相同,只需要在 Spring 的配置文件中声明哪个参数是 callback 类型即可。Dubbo 将基于长连接生成反向代理,这样就可以从服务器端调用客户端逻辑。可以参考 dubbo 项目中的示例代码。
使用场景
回调函数通知客户端执行结果,或发送通知,在方法执行时间比较长时,类似异步调用,审批工作流中回调客户端审批结果。
最核心的配置其实是下面这一段
<bean id="callbackService" class="com.callback.impl.CallbackServiceImpl" />
<dubbo:service interface="com.callback.CallbackService" ref="callbackService" connections="1" callbacks="1000">
<dubbo:method name="addListener">
<dubbo:argument index="1" callback="true" />
<!--也可以通过指定类型的方式-->
<!--<dubbo:argument type="com.demo.CallbackListener" callback="true" />-->
</dubbo:method>
</dubbo:service>
实际example 代码是示意annotation 的,如下
@DubboService(token = "true", connections = 1, callbacks = 1000,
methods = @Method(name = "addListener", arguments = @Argument(index = 1, callback = true)))
public class CallbackServiceImpl implements CallbackService {
private final Map<String, CallbackListener> listeners = new ConcurrentHashMap<String, CallbackListener>();
注意上的@Method及@Argument ,通过Argument 的index=1 定义了第一个参数是callback 函数,目前Argument只支持callback一种类型,@Method 还有一个Parameter 用于定义其它自定义参数。
跟踪分析,可以看到consumer获取到的service信息如下
DefaultServiceInstance{serviceName=‘callback-provider’, host=‘192.168.1.14’, port=20880, enabled=true, healthy=true, metadata={dubbo.endpoints=[{“port”:20880,“protocol”:“dubbo”}], dubbo.metadata-service.url-params={“prefer.serialization”:“fastjson2,hessian2”,“token”:“491567bd-21d1-4465-a500-15c852ce9899”,“version”:“1.0.0”,“dubbo”:“2.0.2”,“release”:“3.2.0”,“side”:“provider”,“port”:“20880”,“protocol”:“dubbo”}, dubbo.metadata.revision=5289ac26fb59bd436247504d9762b074, dubbo.metadata.storage-type=local, timestamp=1686452741890}}, service{name=‘org.apache.dubbo.samples.callback.api.CallbackService’,group=‘null’,version=‘null’,protocol=‘dubbo’,port=‘20880’,params={executor-management-mode=isolation, side=provider, file-cache=true, addListener.return=true, release=3.2.0, methods=addListener, deprecated=false, dubbo=2.0.2, callbacks=1000, interface=org.apache.dubbo.samples.callback.api.CallbackService, service-name-mapping=true, generic=false, token=4eb91b86-6f19-4277-ada3-4a2e2618c3db, addListener.sent=true, application=callback-provider, prefer.serialization=fastjson2,hessian2, addListener.1.callback=true, background=false, dynamic=true, connections=1, anyhost=true},}
然后通过代码public class CallbackServiceCodec的
private static byte isCallBack(URL url, String protocolServiceKey, String methodName, int argIndex) {
// parameter callback rule: method-name.parameter-index(starting from 0).callback
byte isCallback = CALLBACK_NONE;
if (url != null && url.hasServiceMethodParameter(protocolServiceKey, methodName)) {
String callback = url.getServiceParameter(protocolServiceKey, methodName + "." + argIndex + ".callback");
if (callback != null) {
if ("true".equalsIgnoreCase(callback)) {
isCallback = CALLBACK_CREATE;
} else if ("false".equalsIgnoreCase(callback)) {
isCallback = CALLBACK_DESTROY;
}
}
}
return isCallback;
}
核心是String callback = url.getServiceParameter(protocolServiceKey, methodName + “.” + argIndex + “.callback”);
开放一个服务,在log可以看到如下信息
2023-06-11 11:08:53.112 INFO 15528 — [ main] o.a.d.r.p.dubbo.CallbackServiceCodec : [DUBBO] Export a callback service :dubbo://192.168.1.14:55315/org.apache.dubbo.samples.callback.api.CallbackListener.522631570?addListener.1.callback=true&addListener.return=true&addListener.sent=true&anyhost=true&application=callback-consumer&background=false&callbacks=1000&connections=1&deprecated=false&dubbo=2.0.2&dubbo.endpoints=[{“port”:20880,“protocol”:“dubbo”}]&dubbo.metadata-service.url-params={“prefer.serialization”:“fastjson2,hessian2”,“token”:“491567bd-21d1-4465-a500-15c852ce9899”,“version”:“1.0.0”,“dubbo”:“2.0.2”,“release”:“3.2.0”,“side”:“provider”,“port”:“20880”,“protocol”:“dubbo”}&dubbo.metadata.revision=5289ac26fb59bd436247504d9762b074&dubbo.metadata.storage-type=local&dynamic=true&executor-management-mode=isolation&file-cache=true&generic=false&interface=org.apache.dubbo.samples.callback.api.CallbackListener&is_callback_service=true&isserver=false&methods=changed&pid=15528&prefer.serialization=fastjson2,hessian2&qos.enable=false®ister.ip=192.168.1.14&release=3.2.0&service-name-mapping=true&side=consumer&sticky=false×tamp=1686452741890&token=4eb91b86-6f19-4277-ada3-4a2e2618c3db&unloadClusterRelated=false, on NettyChannel [channel=[id: 0xd3dc4a8a, L:/192.168.1.14:55315 - R:/192.168.1.14:20880]], url is: DefaultServiceInstance{serviceName=‘callback-provider’, host=‘192.168.1.14’, port=20880, enabled=true, healthy=true, metadata={dubbo.endpoints=[{“port”:20880,“protocol”:“dubbo”}], dubbo.metadata-service.url-params={“prefer.serialization”:“fastjson2,hessian2”,“token”:“491567bd-21d1-4465-a500-15c852ce9899”,“version”:“1.0.0”,“dubbo”:“2.0.2”,“release”:“3.2.0”,“side”:“provider”,“port”:“20880”,“protocol”:“dubbo”}, dubbo.metadata.revision=5289ac26fb59bd436247504d9762b074, dubbo.metadata.storage-type=local, timestamp=1686452741890}}, service{name=‘org.apache.dubbo.samples.callback.api.CallbackService’,group=‘null’,version=‘null’,protocol=‘dubbo’,port=‘20880’,params={executor-management-mode=isolation, side=provider, file-cache=true, addListener.return=true, release=3.2.0, methods=addListener, deprecated=false, dubbo=2.0.2, callbacks=1000, interface=org.apache.dubbo.samples.callback.api.CallbackService, service-name-mapping=true, generic=false, token=4eb91b86-6f19-4277-ada3-4a2e2618c3db, addListener.sent=true, application=callback-provider, prefer.serialization=fastjson2,hessian2, addListener.1.callback=true, background=false, dynamic=true, connections=1, anyhost=true},}, dubbo version: 3.2.0, current host: 192.168.1.14
所谓notify就是利用Method 的onXXX实现调用跟踪
public @interface Method { String name(); int timeout() default -1; int retries() default -1; String loadbalance() default ""; boolean async() default false; boolean sent() default true; int actives() default -1; int executes() default -1; boolean deprecated() default false; boolean sticky() default false; boolean isReturn() default true; String oninvoke() default ""; String onreturn() default ""; String onthrow() default ""; String cache() default ""; String validation() default ""; String merger() default ""; Argument[] arguments() default {}; /** * Customized parameter key-value pair, for example: {key1, value1, key2, value2} or {"key1=value1", "key2=value2"} */ String[] parameters() default {}; }
官方example code如下
// @Component
public class Task implements CommandLineRunner {
// @DubboReference
@DubboReference(methods = @Method(name = "sayHello", onreturn = "notify.onReturn", onthrow = "notify.onThrow"))
private DemoService demoService;
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。