Oinone远程调用链路源码分析

前提

源码分析版本是 5.1.x版本

概要

在服务启动时,获取注解REMOTE的函数,通过dubbo的泛化调用发布。在调用函数时,通过dubbo泛化调用获取结果。

注册服务者

  1. 在spring 启动方法installOrLoad中初始化
  2. 寻找定义REMOTE的方法
  3. 组装dubbo的服务配置
  4. 组装服务对象实现引用,内容如下,用于注册
    • 调用前置处理
      • 放信息到SessionApi
      • 函数调用链追踪,放到本地TransmittableThreadLocal
      • 从redis中获取到的数据进行反序列化并存在到本地的线程里
      • Trace信息,放一份在sessionApi中 和ThreadLocal
    • 调用函数执行
    • 返回数据转成特定格式
  5. 通过线程组调用dubbo的ServiceConfig.export 服务发布

时序图

Oinone远程调用链路源码分析
注册

源码分析

根据条件判断,确定向dubbo进行服务发布
RemoteServiceLoader

public void publishService(List<FunctionDefinition> functionList,Map<String,Runnable> isPublished) {
        // 因为泛化接口只能控制到namespace,控制粒度不能到fun级别,这里进行去重处理
        Map<String, Function> genericNamespaceMap = new HashMap<>();
        for (FunctionDefinition functionDefinition : functionList) {
            Function function = new Function(functionDefinition)

            try {
               //定义REMOTE, 才给予远程调用
                if (FunctionOpenEnum.REMOTE.in(function.getOpen()) && !ClassUtils.isInterface(function.getClazz())) {
                    genericNamespaceMap.putIfAbsent(RegistryUtils.getRegistryInterface(function), function);
                }
            } catch (PamirsException e) {
            }
        }
        // 发布远程服务
        for (String namespace : genericNamespaceMap.keySet()) {
            Function function = genericNamespaceMap.get(namespace);
            if(isPublished.get(RegistryUtils.getRegistryInterface(function)) == null){
                // 发布,注册远程函数服务,底层使用dubbo的泛化调用
                Runnable registryTask = () -> remoteRegistry.registryService(function);
                isPublished.put(RegistryUtils.getRegistryInterface(function),registryTask);
            }else{

            }
        }
    }

构造ServiceConfig方法,设置成泛化调用,进行发布export()
DefaultRemoteRegistryComponent

     public void registryGenericService(String interfaceName, List<MethodConfig> methods,
                                       String group, String version, Integer timeout, Integer retries) {
        ....
        try {
            ServiceConfig<GenericService> service = new ServiceConfig<>();
            // 服务接口名
            service.setInterface(interfaceName);
            // 服务对象实现引用
            service.setRef(genericService(interfaceName));
            if (null != methods) {
                service.setMethods(methods);
            }
            // 声明为泛化接口
            service.setGeneric(Boolean.TRUE.toString());
            // 基础元数据
            constructService(group, version, timeout, retries, service);
            service.export();
        } catch (Exception e) {
           .....
        }
    }

// 服务对象实现引用
private GenericService genericService(String interfaceName) {
        return (method, parameterTypes, args) -> {
            PamirsSession.clear();
            Function function = Objects.requireNonNull(PamirsSession.getContext()).getFunction(RegistryUtils.getFunctionNamespace(method), RegistryUtils.getFunctionFun(method));
            if (log.isDebugEnabled()) {
                log.debug("interfaceName: " + interfaceName + ", isDataManage: " + function.isDataManager());
            }
            try {
                //前置处理:服务提供者,对请求参数进行对象化拆解,并对请求携带的上下文进行处理
                // 放信息到SessionApi
                // 函数调用链追踪,放到本地TransmittableThreadLocal
                // CommonMetaDataCacheApi.computeMetaData() 从redis中获取到的数据进行反序列化并存在到本地的TL中
                // DataAuditApi.computeDataAuditSession() Trace信息,放一份在sessionApi中 和ThreadLocal中
                Object[] args1 = Spider.getDefaultExtension(RemoteRequestArgApi.class).providerHandle(function.getNamespace(), function.getFun(), args, function.getArguments());

                Object result = FunEngine.get().exclude(ScriptType.REMOTE).run(function, args1);

                //后置处理:服务提供者,对结果进行对象化封装、携带请求上下文进行处理
                return Spider.getDefaultExtension(RemoteResponseApi.class).providerHandle(function, method, result);
            } catch (Throwable e) {
                return Spider.getDefaultExtension(RemoteResponseApi.class).providerExceptionHandle(function, method, e);
            } finally {
                PamirsSession.clear();
            }
        };
    }

注册消费者

  1. 函数处理调用
  2. 注册服务消费者
    • 从ReferenceConfigCache获取泛化
  3. 调用dubbo泛化调用接口
  4. 获取返回信息
    • 获取用户id,放入PamirsSession
    • 如果开启debug模式
      • 存入DEBUG_THREAD_LOCAL本地线程
    • 返回格式
      • IWrapper
      • Pagination
      • Result

时序图

Oinone远程调用链路源码分析

源码分析

泛化调用dobbo接口,并解析返回对象
RemoteComputer

public Object compute(Function function, Object... args) {

        .....
        List<Arg> functionArguments = function.getArguments();
        String methodName = RegistryUtils.getGenericServiceMethodName(function);
        String[] argTypes = FunctionUtils.fetchArgTypes(functionArguments);
        //前置处理:服务消费者,对请求参数进行对象化封装、携带请求上下文进行处理
        Object[] arguments = getRemoteRequestApi().consumerHandle(function.getNamespace(),function.getFun(), args, functionArguments);
        // 泛化调用
        Object result = invoke(function, methodName, argTypes, arguments);
        .....
        //后置处理:服务消费者,对返回结果进行对象化拆解,并对结果携带的上下文进行处理
        // 数据转成IWrapper/Pagination/Result
        return getRemoteResponseApi().consumerHandle(function, result);
}

// 配置请求信息,通过$invoke 实际调用
private Object invoke(Function function, String methodName, String[] argTypes, Object[] arguments) {
        Object result;
        String configCdOwnSign = getSessionFillOwnSignApi().getConfigCdOwnSign();
        if (StringUtils.isBlank(configCdOwnSign)) {
.....
        } else {
            try {
            // 获取服务,由于是泛化调用,所以获取的一定是GenericService类型
                GenericService remoteClient = CommonApiFactory.getApi(RemoteRegistry.class).registryOriginConsumer(function);
         // 第一个参数是需要调用的方法名
         // 第二个参数是需要调用的方法的参数类型数组,为String数组,里面存入参数的全类名。
         // 第三个参数是需要调用的方法的参数数组,为Object数组,里面存入需要的参数
                result = remoteClient.$invoke(methodName, argTypes, arguments);
            } catch (RpcException e) {
                ....            }
        }
        return result;
    }

从缓存中获取泛化
DefaultRemoteRegistryComponent

    public GenericService registryGenericConsumer(String interfaceName, List<MethodConfig> methods,
                                                  String group, String version, Integer timeout, Integer retries) {
        ....
        // 创建服务引用配置
        ReferenceConfig<GenericService> reference = new ReferenceConfig<>();
        reference.setInterface(interfaceName);
        // 设置为泛化调用
        reference.setGeneric(Boolean.TRUE.toString());
        if (null != methods) {
            reference.setMethods(methods);
        }
        constructReference(group, version, timeout, retries, reference);
        return ReferenceConfigCache.getCache().get(reference);
    }

名词解释

泛化调用是指在调用方没有服务方提供的API(SDK)的情况下,对服务方进行调用,并且可以正常拿到调用结果
泛化调用(客户端泛化)
实现泛化实现(服务端泛化)

Oinone社区 作者:oinone原创文章,如若转载,请注明出处:https://doc.oinone.top/backend/17027.html

访问Oinone官网:https://www.oinone.top获取数式Oinone低代码应用平台体验

(0)
oinone的头像oinone
上一篇 2024年9月3日 pm12:53
下一篇 2024年9月5日 pm8:13

相关推荐

  • 自定义RSQL占位符(placeholder)及在权限中使用

    1 自定义RSQL占位符常用场景 统一的数据权限配置 查询表达式的上下文变量扩展 2 自定义RSQL的模板 /** * 演示Placeholder占位符基本定义 * * @author Adamancy Zhang at 13:53 on 2024-03-24 */ @Component public class DemoPlaceHolder extends AbstractPlaceHolderParser { private static final String PLACEHOLDER_KEY = "${thisPlaceholder}"; /** * 占位符 * * @return placeholder */ @Override public String namespace() { return PLACEHOLDER_KEY; } /** * 占位符替换值 * * @return the placeholder replace to the value */ @Override protected String value() { return PamirsSession.getUserId().toString(); } /** * 优先级 * * @return execution order of placeholders, ascending order. */ @Override public Integer priority() { return 0; } /** * 是否激活 * * @return the placeholder is activated */ @Override public Boolean active() { return true; } } 注意事项 在一些旧版本中,priority和active可能不起作用,为保证升级时不受影响,请保证该属性配置正确。 PLACEHOLDER_KEY变量表示自定义占位符使用的关键字,需按照所需业务场景的具体功能并根据上下文语义正确定义。 为保证占位符可以被正确替换并执行,所有占位符都不应该出现重复,尤其是不能与系统内置的重复。 3 占位符使用时的优先级问题 多个占位符在进行替换时,会根据优先级按升序顺序执行,如需要指定替换顺序,可使用Spring的Order注解对其进行排序。 import org.springframework.core.annotation.Order; @Order(0) 4 Oinone平台内置的占位符 占位符 数据类型 含义 备注 ${currentUser} String 当前用户ID 未登录时无法使用 ${currentRoles} Set<String> 当前用户的角色ID集合 未登录时无法使用 5 如何覆盖平台内置的占位符? 通过指定占位符的优先级,并定义相同的namespace可优先替换。 6 如何定义会话级别的上下文变量? 在上述模板中,我们使用的是Oinone平台内置的上下文变量进行演示,通常情况下,我们需要根据实际业务场景增加上下文变量,以此来实现所需功能。 下面,我们将根据当前用户获取当前员工ID定义该上下文变量进行演示。 /** * 员工Session * * @author Adamancy Zhang at 14:33 on 2024-03-24 */ @Component public class EmployeeSession implements HookBefore { private static final String SESSION_KEY = "CUSTOM_EMPLOYEE_ID"; @Autowired private DemoEmployeeService demoEmployeeService; public static String getEmployeeId() { return PamirsSession.getTransmittableExtend().get(SESSION_KEY);…

    2024年3月24日
    1.4K00
  • JSON转换工具类

    JSON转换工具类 JSON转对象 pro.shushi.pamirs.meta.util.JsonUtils JSON转模型 pro.shushi.pamirs.framework.orm.json.PamirsDataUtils

    2023年11月1日
    1.4K00
  • Oinone协同开发源码分析

    前提 源码分析版本是 5.1.x版本 什么是协同开发模式 协同开发模式解决的是不同开发,在开发同一个模型时,不会相互影响,也不会影响到测试环境详见:Oinone协同开发使用手册 协同开发原理 在协同模式下,本地开发的元数据,配置pamirs.data.distribution.session.ownSign参数后,元数据前缀加ownSign值,然后只存在redis缓存,不落库。其它环境无法直接访问到该数据。测试环境,或其它环境访问,需要在url上加ownSign等于设置的,则读redis数据时,除了加载通用数据,也会合并ownSign前缀的redis数据,显示出来 注意事项 协同开发仅支持界面设计器,其他设计器均不支持 不支持权限配置 不支持工作流触发 版本支持 完整支持5.1.0及以上 功能详解 启动时操作 做元数据保护检查 配置ownSign,则key拼接为 ownSign + ‘:’ + key 清除掉ownSign的redis缓存数据;非ownSign不用清理 计算差量数据 有差量数据,放入ownSign标识数据,并清理本地标识 dubbo注册服务,group拼接group + ownSign 后进行注册 读取时操作 读本地 组装key: ownSign + ‘:’ + key 本地缓存有数据,更新缓存本地数据,返回 本地没有数据,读redis,并插入本地缓存 读远程 dubbo注册消费者,group拼接group + ownSign 后进行泛化调用 元数据保护检查 开启数据保护模式,在启动参数里加-PmetaProtected=pamirs 会在启动时,往redis里写入数据 private static final String META_PROTECTED_KEY = “pamirs:check:meta-protected”; private void writeMetaProtected(String metaProtected) { stringRedisTemplate.opsForValue().set(META_PROTECTED_KEY, metaProtected); } 如果同时又设置 pamirs.data.distribution.session.ownSign则会报错 在使用元数据保护模式下,不允许设置 [pamirs.distribution.session.ownSign] 处理逻辑如下 看redis是否启用保护标识的值 获取pamirs.distribution.session.ownSign配置 没有启动参数 且redis没有值,则retrun 如果有启动参数且配置了ownSign,报错 在使用元数据保护模式下,不允许设置 [pamirs.distribution.session.ownSign] 如果有启动参数且 redis没有值或启动参数设置 -P metaForceProtected,则写入redis 如果有启动参数, 且启动参数跟redis值不同,则报错[公共环境开启了元数据保护模式,本地开发环境需配置[pamirs.distribution.session.ownSign]] 如果没有启动参数且redis有值,但没有配置ownSign 报错[公共环境开启了元数据保护模式,本地开发环境需配置[pamirs.distribution.session.ownSign]] 核心代码如下MetadataProtectedChecker public void process(AppLifecycleCommand command, Set<String> runModules, List<ModuleDefinition> installModules, List<ModuleDefinition> upgradeModules, List<ModuleDefinition> reloadModules) { String currentMetaProtected = stringRedisTemplate.opsForValue().get(META_PROTECTED_KEY); String metaProtected = getMetaProtected(); boolean hasCurrentMetaProtected = StringUtils.isNotBlank(currentMetaProtected); boolean hasMetaProtected = StringUtils.isNotBlank(metaProtected); if (!hasCurrentMetaProtected && !hasMetaProtected) { return; } if (hasMetaProtected) { if (Spider.getDefaultExtension(SessionFillOwnSignApi.class).handleOwnSign()) { // 如果有启动参数且配置了ownSign throw new UnsupportedOperationException(“在使用元数据保护模式下,不允许设置 [pamirs.distribution.session.ownSign]”); } if (!hasCurrentMetaProtected || isForceProtected()) { writeMetaProtected(metaProtected); } else if (!metaProtected.equals(currentMetaProtected)) { // 如果有启动参数, 且启动参数跟redis值不同 throw unsupportedLocalOperation(); } } else { if (Spider.getDefaultExtension(SessionFillOwnSignApi.class).handleOwnSign()) { return; } // 没有启动参数且redis有值,但没有配置ownSign 报错 throw unsupportedLocalOperation(); } } 取ownSign方式 看header是否有ownSign这个标识 header没有,则从配置里取,并放到header里 ownSign的获取核心代码 CdDistributionSessionFillOwnSignApi @Override public String getCdOwnSign() { String cdOwnSign = null; // 看header是否有ownSign这个标识…

    2024年9月12日
    1.4K00
  • 如何重写获取首页的方法

    介绍 用户登录成功后或者访问网页不带任何路由参数的时候前端会请求全局的首页的视图动作viewAction配置,然后跳转到该视图动作viewAction 方案 我们可以通过在该方法的后置hook自定义获取首页的逻辑,下面以按角色跳转不同首页的需求示例 package pro.shushi.pamirs.demo.core.hook; import org.apache.commons.collections4.CollectionUtils; import org.springframework.stereotype.Component; import pro.shushi.pamirs.auth.api.model.AuthRole; import pro.shushi.pamirs.boot.base.enmu.BaseExpEnumerate; import pro.shushi.pamirs.boot.base.model.ViewAction; import pro.shushi.pamirs.boot.web.loader.PageLoadAction; import pro.shushi.pamirs.demo.api.model.DemoItemCategory; import pro.shushi.pamirs.demo.api.model.DemoItemLabel; import pro.shushi.pamirs.meta.annotation.Hook; import pro.shushi.pamirs.meta.api.CommonApiFactory; import pro.shushi.pamirs.meta.api.core.faas.HookAfter; import pro.shushi.pamirs.meta.api.dto.fun.Function; import pro.shushi.pamirs.meta.api.session.PamirsSession; import pro.shushi.pamirs.meta.common.exception.PamirsException; import pro.shushi.pamirs.user.api.model.PamirsUser; import java.util.List; import java.util.stream.Collectors; @Component public class DemoHomepageHook implements HookAfter { private static final String TEST_ROLE_CODE_01 = "ROLE_1211"; private static final String TEST_ROLE_CODE_02 = "ROLE_1211_1"; @Override @Hook(module = {"base"}, model = {ViewAction.MODEL_MODEL}, fun = {"homepage"}) public Object run(Function function, Object ret) { if (ret == null) { return null; } ViewAction viewAction = getViewActionByCurrentRole(); if (viewAction != null) { ViewAction retNew = CommonApiFactory.getApi(PageLoadAction.class).load(viewAction); ViewAction viewActionRet = (ViewAction) ((Object[]) ret)[0]; viewActionRet.set_d(retNew.get_d()); } return ret; } protected ViewAction getViewActionByCurrentRole() { try { PamirsUser user = new PamirsUser(); user.setId(PamirsSession.getUserId()); user.fieldQuery(PamirsUser::getRoles); List<AuthRole> roles = user.getRoles(); if (CollectionUtils.isNotEmpty(roles)) { List<String> roleCodes = roles.stream().map(AuthRole::getCode).collect(Collectors.toList()); if (roleCodes.contains(TEST_ROLE_CODE_01)) { return new ViewAction().setModel(DemoItemCategory.MODEL_MODEL).setName("DemoMenus_ItemPMenu_DemoItemAndCateMenu_DemoItemCategoryMenu").queryOne(); } else if (roleCodes.contains(TEST_ROLE_CODE_02)) { return new ViewAction().setModel(DemoItemLabel.MODEL_MODEL).setName("DemoMenus_ItemPMenu_DemoItemAndCateMenu_DemoItemLabelMenu").queryOne(); } } } catch (PamirsException exception) { if (PamirsSession.getUserId() == null) { throw PamirsException.construct(BaseExpEnumerate.BASE_USER_NOT_LOGIN_ERROR, exception.getCause()).errThrow(); } else { throw exception; } } return null; } }

    2024年7月6日
    2.2K00
  • 项目中工作流引入和流程触发

    目录 1. 使用工作流需要依赖的包和设置2. 触发方式2.1 自动触发方式2.2 触发方式 1.使用工作流需要依赖的包和设置 1.1 工作流需要依赖的模块 需在pom.xml中增加workflow、sql-record和trigger相关模块的依赖 workflow:工作流运行核心模块 sql-record:监听流程发布以后对应模型的增删改监听 trigger:异步任务调度模块 <dependency> <groupId>pro.shushi.pamirs.workflow</groupId> <artifactId>pamirs-workflow-api</artifactId> </dependency> <dependency> <groupId>pro.shushi.pamirs.workflow</groupId> <artifactId>pamirs-workflow-core</artifactId> </dependency> <dependency> <groupId>pro.shushi.pamirs.core</groupId> <artifactId>pamirs-sql-record-core</artifactId> </dependency> <dependency> <groupId>pro.shushi.pamirs.core</groupId> <artifactId>pamirs-trigger-core</artifactId> </dependency> <dependency> <groupId>pro.shushi.pamirs.core</groupId> <artifactId>pamirs-trigger-bridge-tbschedule</artifactId> </dependency> 在application.yml中增加对应模块的依赖以及sql-record路径以及其他相关设置 pamirs: … record: sql: #改成自己路径 store: /opt/pamirs/logs … boot: init: true sync: true modules: … – sql_record – trigger – workflow … sharding: define: data-sources: ds: pamirs models: "[trigger.PamirsSchedule]": tables: 0..13 event: enabled: true schedule: enabled: true # ownSign区分不同应用 ownSign: demo rocket-mq: # enabled 为 false情况不用配置 namesrv-addr: 192.168.6.2:19876 trigger: auto-trigger: true 2.触发方式 2.1自动触发方式 在流程设计器中设置触发方式,如果设置了代码触发方式则不会自动触发 2.2代码调用方式触发 2.2.1.再流程设计器中触发设置中,设置为是否人工触发设置为是 2.2.2.查询数据库获取该流程的编码 2.2.3.在代码中调用 /** * 触发⼯作流实例 */ private Boolean startWorkflow(WorkflowD workflowD, IdModel modelData) { WorkflowDefinition workflowDefinition = new WorkflowDefinition().queryOneByWrapper( Pops.<WorkflowDefinition>lambdaQuery() .from(WorkflowDefinition.MODEL_MODEL) .eq(WorkflowDefinition::getWorkflowCode, workflowD.getCode()) .eq(WorkflowDefinition::getActive, 1) ); if (null == workflowDefinition) { // 流程没有运⾏实例 return Boolean.FALSE; } String model = Models.api().getModel(modelData); //⼯作流上下⽂ WorkflowDataContext wdc = new WorkflowDataContext(); wdc.setDataType(WorkflowVariationTypeEnum.ADD); wdc.setModel(model); wdc.setWorkflowDefinitionDefinition(workflowDefinition.parseContent()); wdc.setWorkflowDefinition(workflowDefinition); wdc.setWorkflowDefinitionId(workflowDefinition.getId()); IdModel copyData = KryoUtils.get().copy(modelData); // ⼿动触发创建的动作流,将操作⼈设置为当前⽤户,作为流程的发起⼈ copyData.setCreateUid(PamirsSession.getUserId()); copyData.setWriteUid(PamirsSession.getUserId()); String jsonData = JsonUtils.toJSONString(copyData.get_d()); //触发⼯作流 新增时触发-onCreateManual 更新时触发-onUpdateManual Fun.run(WorkflowModelTriggerFunction.FUN_NAMESPACE, "onCreateManual", wdc, msgId, jsonData); return Boolean.TRUE; }

    2023年11月7日
    1.2K00

Leave a Reply

登录后才能评论