代码技巧——dubbo泛化调用

约定不等于承诺〃 2023-09-28 22:01 105阅读 0赞

现在有一种业务场景：

（1）业务方B调用任务系统A，提交一个任务，由A保证任务尽可能的被执行成功（系统A自带幂等、重试），在A完成任务的执行后将执行情况同步回调通知B应用；

（2）因此，B应用在提交任务时需要把自己的回调地址作为参数传递给A，这个参数被定义为dubbo接口的全路径+方法名+分组+版本号；

（3）系统A不可能提前知晓这个dubbo接口并注册对应的consumer，因此只能通过dubbo的「**泛化调用**」方式调用业务方B的dubbo接口，即Dubbo的调用方，在不引入服务接口类的情况下，远程调用其他Dubbo服务；

> "泛化接口调用方式主要用于客户端没有API接口及模型类元的情况，参数及返回值中的所有POJO均用Map表示，通常用于框架集成，比如：实现一个通用的服务测试框架，可通过GenericService调用所有服务实现。"——[使用泛化调用 | Apache Dubbo][_ Apache Dubbo]

下面给出代码示例；

**（1）核心方法**

import com.alibaba.fastjson.JSON;
    import com.google.common.collect.Maps;
    import com.internet.bocfg.audit.client.model.AuditFlowCallbackRequest;
    import com.internet.bocfg.audit.client.model.AuditFlowCallbackResponse;
    import com.internet.bocfg.audit.constant.AuditConstants;
    import com.internet.bocfg.audit.exception.AuditBusinessException;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.lang3.StringUtils;
    import org.apache.dubbo.config.ApplicationConfig;
    import org.apache.dubbo.config.ReferenceConfig;
    import org.apache.dubbo.config.RegistryConfig;
    import org.apache.dubbo.config.utils.ReferenceConfigCache;
    import org.apache.dubbo.rpc.service.GenericService;
    import org.springframework.stereotype.Service;
    
    import javax.annotation.Resource;
    import java.util.Map;
    
    /**
     * @author Akira
     * @description dubbo泛化调用服务
     * @date 2022/8/28
     */
    @Slf4j
    @Service
    public class DubboGenericService {
    
        /**
         * 缓存已经注册过的GenericService
         */
        private static final Map<String, GenericService> stubCache = Maps.newConcurrentMap();
    
        private final ApplicationConfig applicationConfig = new ApplicationConfig("bocfg-audit");
    
        @Resource(name = "myRegistryConfig")
        private RegistryConfig registryConfig;
    
        /**
         * (泛化)远程调用
         *
         * @param serviceName 接口路径
         * @param methodName  方法名
         * @param group       分组
         * @param version     版本
         * @param request     自定义的dubbo接口参数
         */
        private void remoteInvoke(String serviceName, String methodName, String group, String version,
                                  AuditFlowCallbackRequest request) {
            try {
                // 先尝试从缓存中拿GenericService 设置缓存的原因：ReferenceConfig实例很重，封装了与注册中心的连接以及与provider的连接，需要缓存，否则重复生成ReferenceConfig可能造成性能问题并且会有内存和连接泄漏
                String stubKey = stubKey(serviceName, methodName, version);
                GenericService genericService = stubCache.get(stubKey);
                // 未取到则准备初始化当前GenericService
                if (genericService == null) {
                    ReferenceConfig<GenericService> reference = new ReferenceConfig<>();
                    // 连接相关的配置，包括：zk地址、应用名、协议等
                    reference.setApplication(applicationConfig);
                    reference.setRegistry(registryConfig);
                    // 调用相关的配置，包括：声明为泛化接口、不检查状态、负载均衡方式、超时时间等
                    reference.setGeneric(true);
                    reference.setLoadbalance("roundrobin");
                    reference.setCheck(false);
                    reference.setTimeout(AuditConstants.AUDIT_CALLBACK_DEFALUT_TIMEOUT);
                    // 调用目标provider相关的配置，包括：接口、group、version、
                    reference.setInterface(serviceName);
                    reference.setGroup(StringUtils.isBlank(group) ? AuditConstants.AUDIT_CALLBACK_DEFALUT_GROUP : group);
                    reference.setVersion(StringUtils.isBlank(version) ? AuditConstants.AUDIT_CALLBACK_DEFALUT_VERSION : version);
                    // 这里优先使用dubbo内置的简单缓存工具类进行缓存，若没有则放入自己定义的缓存stubCache中
                    ReferenceConfigCache cache = ReferenceConfigCache.getCache();
                    genericService = cache.get(reference);
                    if (genericService != null) {
                        stubCache.putIfAbsent(stubKey, genericService);
                    }
                    genericService = stubCache.get(stubKey);
                }
                // 至此，拿到了dubbo接口的genericService
                if (genericService == null) {
                    throw new IllegalStateException("No provider available: " + stubKey);
                }
                // 泛化调用的参数：方法名、方法参数类型全路径、方法参数
                String[] parameterTypes = new String[]{AuditConstants.AUDIT_CALLBACK_REQUEST_CLASSNAME};
                Object[] args = new Object[]{request};
                Object result = genericService.$invoke(methodName, parameterTypes, args);
                if (result != null) {
                    // 泛化调用返参为Object类型，这里做一个参数转换
                    AuditFlowCallbackResponse callbackResponse = JSON.parseObject(JSON.toJSONString(result), AuditFlowCallbackResponse.class);
                    if (!callbackResponse.isSuccess()) {
                        log.warn("callbackResponse fail. serviceName={} methodName={} version={} request={} return={}", serviceName, methodName, version, JSON.toJSONString(request), JSON.toJSONString(callbackResponse));
                        throw new AuditBusinessException("auditCallback fail:" + callbackResponse.getMsg());
                    }
                }
            } catch (Throwable e) {
                log.error("dubbo remoteInvoke fail. serviceName={} methodName={} version={} auditFlowCallbackRequest={}", serviceName, methodName, version, JSON.toJSONString(request));
                throw e;
            }
        }
    
        /**
         * 拼接dubbo的GenericService缓存stubCache(map)中的存根key
         */
        private String stubKey(String serviceName, String method, String version) {
            return serviceName + "$" + method + "$" +
                    (StringUtils.isBlank(version) ? AuditConstants.AUDIT_CALLBACK_DEFALUT_VERSION : version);
        }
    
    }

**（2）RegistryConfig配置**

import com.internet.vivocfg.client.ConfigManager;
    import org.apache.dubbo.config.RegistryConfig;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class PropertyConfig {
    
        @Bean("myRegistryConfig")
        public RegistryConfig imExchangeConfig() {
            RegistryConfig registryConfig = null;
            registryConfig = new RegistryConfig(ConfigManager.get("dubbo.zk.registry.address"));
            registryConfig.setProtocol("zookeeper");
            registryConfig.setClient("curator");
            return registryConfig;
        }
    }

参考：

[使用泛化调用 | Apache Dubbo][_ Apache Dubbo]

[dubbo实战之“泛化调用”探索 - 博客园][dubbo_ -]

[_ Apache Dubbo]: https://dubbo.apache.org/zh/docs/advanced/generic-reference/
[dubbo_ -]: https://www.cnblogs.com/kendoziyu/p/dubbo-GenericService-invoke.html