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

相关推荐

  • 函数之异步执行

    总体介绍 异步任务是非常常见的一种开发模式,它在分布式的开发模式中有很多应用场景如: 高并发场景中,我们一般采用把长流程切短,用异步方式去掉可以异步的非关键功能,缩小主流程响应时间,提升用户体验 异构系统的集成调用,通过异步任务完成解耦与自动重试 分布式系统最终一致性的可选方案 本文将介绍oinone是如何结合Spring+TbSchedule来完成异步任务 构建第一个异步任务 新建PetShopService和PetShopServiceImpl 1、 新建PetShopService定义updatePetShops方法 package pro.shushi.pamirs.demo.api.service; import pro.shushi.pamirs.demo.api.model.PetShop; import pro.shushi.pamirs.meta.annotation.Fun; import pro.shushi.pamirs.meta.annotation.Function; import java.util.List; @Fun(PetShopService.FUN_NAMESPACE) public interface PetShopService { String FUN_NAMESPACE = "demo.PetShop.PetShopService"; @Function void updatePetShops(List<PetShop> petShops); } 2、PetShopServiceImpl实现PetShopService接口并在updatePetShops增加@XAsync注解 package pro.shushi.pamirs.demo.core.service; import org.springframework.stereotype.Component; import pro.shushi.pamirs.demo.api.model.PetShop; import pro.shushi.pamirs.demo.api.service.PetShopService; import pro.shushi.pamirs.meta.annotation.Fun; import pro.shushi.pamirs.meta.annotation.Function; import pro.shushi.pamirs.trigger.annotation.XAsync; import java.util.List; @Fun(PetShopService.FUN_NAMESPACE) @Component public class PetShopServiceImpl implements PetShopService { @Override @Function @XAsync(displayName = "异步批量更新宠物商店",limitRetryNumber = 3,nextRetryTimeValue = 60) public void updatePetShops(List<PetShop> petShops) { new PetShop().updateBatch(petShops); } } a. displayName = "异步批量更新宠物商店",定义异步任务展示名称b. limitRetryNumber = 3,定义任务失败重试次数,,默认:-1不断重试c. nextRetryTimeValue = 60,定义任务失败重试的时间数,默认:3d. nextRetryTimeUnit,定义任务失败重试的时间单位,默认:TimeUnitEnum.SECONDe. delayTime,定义任务延迟执行的时间数,默认:0f. delayTimeUnit,定义任务延迟执行的时间单位,默认:TimeUnitEnum.SECOND 修改PetShopBatchUpdateAction调用异步任务 引入PetShopService 修改conform方法,调用petShopService.updatePetShops方法 package pro.shushi.pamirs.demo.core.action; @Model.model(PetShopBatchUpdate.MODEL_MODEL) @Component public class PetShopBatchUpdateAction { @Autowired private PetShopService petShopService; @Action(displayName = "确定",bindingType = ViewTypeEnum.FORM,contextType = ActionContextTypeEnum.SINGLE) public PetShopBatchUpdate conform(PetShopBatchUpdate data){ List<PetShop> shops = ArgUtils.convert(PetShopProxy.MODEL_MODEL, PetShop.MODEL_MODEL,proxyList); // 调用异步任务 petShopService.updatePetShops(shops); }); return data; } } 不同应用如何隔离执行单元 在schedule跟模块部署一起的时候,多模块独立boot的情况下,需要做必要的配置。如果schedule独立部署则没有必要,因为全部走远程,不存在类找不到的问题 通过配置pamirs.zookeeper.rootPath,确保两组机器都能覆盖所有任务分片,这样不会漏数据 通过pamirs.event.schedule.ownSign来隔离。确保两组机器只取各自产生的数据,这样不会重复执行数据 pamirs: zookeeper: zkConnectString: 127.0.0.1:2181 zkSessionTimeout: 60000 rootPath: /demo event: enabled: true schedule: enabled: true ownSign: demo rocket-mq: namesrv-addr: 127.0.0.1:9876

    2024年5月25日
    77700
  • OSS(CDN)配置和文件系统的一些操作

    目前Oinone支持的OSS类型 类型 服务 OSS 阿里云OSS UPYUN 又拍云 MINIO MinIO HUAWEI_OBS 华为云OBS LOCAL 本地NGINX文件存储 TENCENT_COS 腾讯云COS OSS通用yaml配置 cdn: oss: name: # 名称 type: # 类型 bucket: uploadUrl: # 上传URL downloadUrl: # 下载URL accessKeyId: accessKeySecret: mainDir: # 主目录 validTime: 3600000 timeout: 600000 active: true referer: localFolderUrl: others: [key]: name: # 名称 type: # 类型 bucket: uploadUrl: # 上传URL downloadUrl: # 下载URL accessKeyId: accessKeySecret: mainDir: # 主目录 validTime: 3600000 timeout: 600000 active: true referer: localFolderUrl: PS:others中使用自定义key来指定OSS服务进行文件上传/下载功能。上传/下载必须匹配,否则无法正常使用。 OSS配置示例 阿里云OSS cdn: oss: name: 阿里云 type: OSS bucket: pamirs(根据实际情况修改) uploadUrl: oss-cn-hangzhou.aliyuncs.com downloadUrl: oss-cn-hangzhou.aliyuncs.com accessKeyId: 你的accessKeyId accessKeySecret: 你的accessKeySecret # 根据实际情况修改 mainDir: upload/ validTime: 3600000 timeout: 600000 active: true imageResizeParameter: referer: 华为云OBS cdn: oss: name: 华为云 type: HUAWEI_OBS bucket: pamirs(根据实际情况修改) uploadUrl: obs.cn-east-2.myhuaweicloud.com downloadUrl: obs.cn-east-2.myhuaweicloud.com accessKeyId: 你的accessKeyId accessKeySecret: 你的accessKeySecret # 根据实际情况修改 mainDir: upload/ validTime: 3600000 timeout: 600000 active: true allowedOrigin: http://192.168.95.31:8888,https://xxxx.xxxxx.com referer: 华为云OBS需要在启动工程增加以下依赖 <okhttp3.version>4.9.3</okhttp3.version> <dependency> <groupId>com.squareup.okhttp3</groupId> <artifactId>okhttp</artifactId> <version>${okhttp3.version}</version> </dependency> 注意事项华为云OBS的防盗链配置,仅允许携带特定referer的才可以,而excel导入后端处理的逻辑匿名读的时候是不带referer的,所以会被拒绝 MINIO 文件系统,mino的配置: cdn: oss: name: minio type: MINIO bucket: pamirs(根据实际情况修改) uploadUrl: http://192.168.243.6:32190(根据实际情况修改) downloadUrl: http://192.168.243.6:9000(根据实际情况修改) accessKeyId: 你的accessKeyId accessKeySecret: 你的accessKeySecret # 根据实际情况修改 mainDir: upload/ validTime: 3600000 timeout: 600000 active: true referer: localFolderUrl: MINIO无公网访问地址下OSS的配置方式: https://doc.oinone.top/yun-wei-shi-jian/7112.html 又拍云 cdn: oss: name: 又拍云 type: UPYUN…

    后端 2023年11月1日
    81400
  • 导出导入翻译

    http://168.138.179.151/pamirs/file 导出翻译项: mutation { excelExportTaskMutation { createExportTask( data: { workbookDefinition: { model: "file.ExcelWorkbookDefinition" name: "excelLocationTemplate" } } ) { name } } } { "path": "/file", "lang": "en-US" } 导入翻译项: mutation { excelImportTaskMutation { createImportTask( data: { workbookDefinition: { model: "file.ExcelWorkbookDefinition" name: "excelLocationTemplate" } file: { url: "https://minio.oinone.top/pamirs/upload/zbh/test/2024/06/03/导出国际化配置模板_1717390304285_1717391684633.xlsx" } } ) { name } } } PS:导入自行修改url进行导入

    2024年6月28日
    71100
  • 【PostgreSQL】后端部署使用PostgreSQL数据库

    PostgreSQL数据库配置 驱动配置 Maven配置(14.3版本可用) <postgresql.version>42.6.0</postgresql.version> <dependency> <groupId>org.postgresql</groupId> <artifactId>postgresql</artifactId> <version>${postgresql.version}</version> </dependency> 离线驱动下载 postgresql-42.2.18.jarpostgresql-42.6.0.jarpostgresql-42.7.3.jar JDBC连接配置 pamirs: datasource: base: type: com.alibaba.druid.pool.DruidDataSource driverClassName: org.postgresql.Driver url: jdbc:postgresql://127.0.0.1:5432/pamirs?currentSchema=base username: xxxxxx password: xxxxxx initialSize: 5 maxActive: 200 minIdle: 5 maxWait: 60000 timeBetweenEvictionRunsMillis: 60000 testWhileIdle: true testOnBorrow: false testOnReturn: false poolPreparedStatements: true asyncInit: true 连接url配置 暂无官方资料 url格式 jdbc:postgresql://${host}:${port}/${database}?currentSchema=${schema} 在jdbc连接配置时,${database}和${schema}必须完整配置,不可缺省。 其他连接参数如需配置,可自行查阅相关资料进行调优。 方言配置 pamirs方言配置 pamirs: dialect: ds: base: type: PostgreSQL version: 14 major-version: 14.3 pamirs: type: PostgreSQL version: 14 major-version: 14.3 数据库版本 type version majorVersion 14.x PostgreSQL 14 14.3 PS:由于方言开发环境为14.3版本,其他类似版本(14.x)原则上不会出现太大差异,如出现其他版本无法正常支持的,可在文档下方留言。 schedule方言配置 pamirs: event: enabled: true schedule: enabled: true dialect: type: PostgreSQL version: 14 major-version: 14.3 type version majorVersion PostgreSQL 14 14.3 PS:由于schedule的方言在多个版本中并无明显差异,目前仅提供一种方言配置。 其他配置 逻辑删除的值配置 pamirs: mapper: global: table-info: logic-delete-value: (EXTRACT(epoch FROM CURRENT_TIMESTAMP) * 1000000 + EXTRACT(MICROSECONDS FROM CURRENT_TIMESTAMP))::bigint PostgreSQL数据库用户初始化及授权 — init root user (user name can be modified by oneself) CREATE USER root WITH PASSWORD 'password'; — if using automatic database and schema creation, this is very important. ALTER USER root CREATEDB; SELECT * FROM pg_roles; — if using postgres database, this authorization is required. GRANT CREATE ON DATABASE postgres TO root;

    2023年11月1日
    59300
  • 首次登录修改密码和自定义密码规则等

    场景描述 在某些场景下,可能需要实现 用户首次登录强制修改密码的功能,或者存在修改平台默认密码等校验规则等需求;本文将讲解不改变平台代码的情况下,如何实现这些功能需求。 首次登录修改密码 方案概述 自定义User增加是否是第一次登录的属性,登录后执行一个扩展点。 判断是否是一次登录,如果是则返回对应的状态码,前端根据状态码重定向到修改密码的页面。修改完成则充值第一次登录的标识。 PS:首次登录的标识平台前端已默认实现 扩展PamirsUser(例如:DemoUser) /** * @author wangxian */ @Model.model(DemoUser.MODEL_MODEL) @Model(displayName = "用户", labelFields = {"nickname"}) @Model.Advanced(index = {"companyId"}) public class DemoUser extends PamirsUser { public static final String MODEL_MODEL = "demo.DemoUser"; @Field.Integer @Field.Advanced(columnDefinition = "bigint DEFAULT '0'") @Field(displayName = "公司ID", invisible = true) private Long companyId; /** * 默认true->1 */ @Field.Boolean @Field.Advanced(columnDefinition = "tinyint(1) DEFAULT '1'") @Field(displayName = "是否首次登录") private Boolean firstLogin; } 编写扩展点(例如:DemoUserLoginExtPoint) @Order(0) @Component @Ext(PamirsUserTransient.class) @Slf4j public class DemoUserLoginExtPoint implements PamirsUserTransientExtPoint { @Override @ExtPoint.Implement public PamirsUserTransient loginAfter(PamirsUserTransient user) { return checkFirstLogin(user); } private PamirsUserTransient checkFirstLogin(PamirsUserTransient user) { //首次登录需要修改密码 Long userId = PamirsSession.getUserId(); if (userId == null) { return user; } DemoUser companyUser = new DemoUser().queryById(userId); // 判断用户是否是第一次登录,如果是第一次登录,需要返回错误码,页面重新向登录 Boolean isFirst = companyUser.getFirstLogin(); if (isFirst) { //如果是第一次登录,返回一个标识给前端。 // 首次登录的标识平台已默认实现 user.setBroken(Boolean.TRUE); user.setErrorCode(UserExpEnumerate.USER_FIRST_LOGIN_ERROR.code()); return user; } return user; } @Override public PamirsUserTransient loginCustomAfter(PamirsUserTransient user) { return checkFirstLogin(user); } @Override @ExtPoint.Implement public PamirsUserTransient firstResetPasswordAfter(PamirsUserTransient user) { return updateFirstLogin(user); } private PamirsUserTransient updateFirstLogin(PamirsUserTransient user) { Long userId = PamirsSession.getUserId(); if (userId == null) { return user; } //修改密码后 将首次登录标识改为false Integer update = new DemoUser().updateByWrapper(new DemoUser().setFirstLogin(Boolean.FALSE), Pops.<DemoUser>lambdaUpdate() .from(DemoUser.MODEL_MODEL) .eq(IdModel::getId,…

    2024年5月25日
    1.8K00

Leave a Reply

登录后才能评论