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

相关推荐

  • 【Oracle】后端部署使用Oracle数据库

    Oracle数据库配置 驱动配置 jdbc仓库 https://mvnrepository.com/artifact/com.oracle.database.jdbc/ojdbc8 Maven配置(11g版本可用) <ojdbc.version>23.2.0.0</ojdbc.version> <dependency> <groupId>com.oracle.database.jdbc</groupId> <artifactId>ojdbc8</artifactId> <version>${ojdbc.version}</version> </dependency> JDBC连接配置 pamirs: datasource: base: type: com.alibaba.druid.pool.DruidDataSource driverClassName: oracle.jdbc.OracleDriver url: jdbc:oracle:thin:@//127.0.0.1:1521/orcl username: YOUR_SCHEMA_NAME password: xxxxxx Oracle默认为每个用户创建了一个与当前用户名同名的模式,每个用户应该只使用该模式(DBA用户除外),因此平台使用Oracle时应该通过username处指定与该模式同名的用户名来指定模式。(Oracle多数据源时每一个数据库创建一个用户) 创建用户时用户名应全大写。 连接url配置 官方文档 https://odbc.postgresql.org/docs/config-opt.html url格式 jdbc:oracle:thin:@//ip:端口号/服务名或SID 每一个Oracle进程默认为一个Oracle数据库实例,使用服务名或sid登录该Oralce数据库实例。一个Oracle sid 对应一个数据库实例,而一个服务名可以标识多个数据库实例。远程连接时推荐使用服务名进行连接。可以在安装Oracle的机器上打开SQLPlus,用SYSTEM用户登录上去后使用SELECT SYS_CONTEXT('USERENV', 'INSTANCE_NAME') AS SID FROM DUAL;查询登录使用的sid;也可以使用SELECT VALUE AS SERVICE_NAME FROM V$PARAMETER WHERE NAME = 'service_names';查询登录使用的服务名。 其他连接参数如需配置,可自行查阅相关资料进行调优。 方言配置 pamirs方言配置 pamirs: dialect: ds: base: type: Oracle version: 11.2 major-version: 11g pamirs: type: Oracle version: 11.2 major-version: 11g plus: configuration: jdbc-type-for-null: "NULL" using-model-as-property: true using-statement-handler-dialect: true mapper: batch: useAffectRows global: table-pattern: '${table_30}' column-pattern: '${column_30}' 数据库版本 type version majorVersion 11g – 11.2.0.1.0 Oracle 11.2 11g 12c – 12.2.0.1.0 Oracle 12.2 12c PS:由于方言开发环境为Oracle Database 11g Enterprise Edition Release 11.2.0.1.0版本,其他类似版本(11.2.x)原则上不会出现太大差异,如出现其他版本无法正常支持的,可在文档下方留言。 schedule方言配置 pamirs: event: enabled: true schedule: enabled: true dialect: type: Oracle version: 11.2 major-version: 11g 其他配置 逻辑删除的值配置 pamirs: mapper: global: table-info: logic-delete-value: (CAST(SYSTIMESTAMP AS DATE) – TO_DATE('1970-01-01 08:00:00', 'YYYY-MM-DD HH24:MI:SS')) * 8640000000000 Oracle数据库用户初始化及授权 — 以下命令均使用dba账户执行 — 创建用户 ONE_TEST (用户名需全大写) 密码 123456 CREATE USER ONE_TEST IDENTIFIED BY 123456; — 解锁用户 ALTER USER ONE_TEST ACCOUNT UNLOCK; — 将用户的默认表空间设置为USERS,临时表空间设置为TEMP ALTER USER ONE_TEST DEFAULT TABLESPACE USERS; ALTER USER ONE_TEST TEMPORARY TABLESPACE TEMP; — 可以用以下命令查询某用户的表空间: SELECT…

    2025年7月10日
    38800
  • 工作流引入流程概览与流程监控

    流程概览依赖说明 使用 流程概览 功能前,需要在项目中引入 pamirs-workflow-datavi-core、 pamirs-data-visualization-core依赖,并启动datavi模块: <dependency> <groupId>pro.shushi.pamirs.workflow</groupId> <artifactId>pamirs-workflow-datavi-core</artifactId> </dependency> <dependency> <groupId>pro.shushi.pamirs.data.visualization</groupId> <artifactId>pamirs-data-visualization-core</artifactId> </dependency> 警告: 在 oinone 平台启用「流程概览」能力时,应用启动模块一旦引入 pamirs-workflow-api/core,必须同时引入 pamirs-workflow-datavi-api/core。在多启动模块架构下,严禁出现仅部分启动模块引入 pamirs-workflow-core 而未引入 pamirs-workflow-datavi-core 的情况,否则将导致流程概览相关元数据计算异常,出现删表等情况。 流程概览配置项 流程概览页面内置缓存机制,可通过配置项调整缓存刷新周期及图表展示的数据条数: pamirs: workflow: dashboard: cache-time: 10 # 流程概览缓存刷新时间(单位:分钟),默认 10 分钟 page-size: 10 # 流程运行分析中 4 个图表的展示数量,默认查询前 10 条数据 统计指标说明 引入 pamirs-workflow-datavi-core 依赖后,系统会按照以下规则进行数据同步: 当日数据同步:每小时同步一次当日数据; 昨日数据同步:次日凌晨同步前一日数据。 由于在引入依赖后才会开始执行数据同步,统计指标页提供了「同步」按钮,可用于对历史数据进行补采。即使不执行历史同步,也不会影响核心业务流程,仅会影响统计数据和图表的展示效果。 统计指标数据主要用于 支撑 流程概览 和 流程监控 中的统计图表展示; 为数据分析与可视化提供基础数据。 上述统计数据对工作流的审批、流转等核心业务无任何影响。如有需要,也可以基于流程监控的数据,配合数据可视化设计器,自定义构建符合业务需求的展示页面。

    2025年11月17日
    17100
  • 平台配置日志输出和推送到APM与LogStash

    场景描述 目前设计器镜像启动后日志文件为out.log,是启动脚本中定向输出了(>>)out.log文件。实际项目可能: 日志输出到特定目录的特定文件名中 指定以日志保留策略(单个文件大小和文件保留个数) 日志输出到APM工具中(如skywalking) 日志推送到LogStash 日志自定义输出 不定向输出,采用自己配置的方式,与标准的SpringBoot工程配置日志一样。两种方式(都是Spring提供的方式): 方式一 bootstrap.yml 里面可以按profiles指定logback的配置文件,具体文件名和文件输入在logback里面进行配置,跟通用的logback配置一致. 例如: logging: config: classpath:logback-pre.xml 方式二 resources的根目录,直接配置 logback-spring.xml, 启动会自动加载。 日志自定义场景 配置日志推送到LogStash <!–配置日志推送到LogStash–> <contextListener class="pro.shushi.pamirs.demo.core.config.DemoLogbackFiledConfig"/> <appender name="LogStash" class="net.logstash.logback.appender.LogstashTcpSocketAppender"> <destination>127.0.0.1:4560</destination> <!– encoder必须配置,有多种可选 –> <encoder charset="UTF-8" class="net.logstash.logback.encoder.LogstashEncoder"> <!– SkyWalking插件, log加tid–> <provider class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.logstash.TraceIdJsonProvider" /> <!–在生成的json中会加这些字段–> <customFields> {"app.name":"pamirs-demo", "app.type":"Microservice", "platform":"pamirs", "env":"dev"} </customFields> <timeZone>Asia/Shanghai</timeZone> <writeVersionAsInteger>true</writeVersionAsInteger> <providers> <pattern> <pattern> <!–动态的变量–> { "ip": "%{ip}", "server.name": "%{server.name}", "logger_name": "%logger" } </pattern> </pattern> </providers> </encoder> </appender> skywalking的日志rpc上传 <!– skywalking的日志rpc上传 –> <appender name="SkyWalkingLogs" class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.log.GRPCLogClientAppender"> <encoder class="ch.qos.logback.core.encoder.LayoutWrappingEncoder"> <layout class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.mdc.TraceIdMDCPatternLogbackLayout"> <Pattern>${CONSOLE_LOG_PATTERN}</Pattern> </layout> </encoder> </appender> 完整的代码示例 Logback自定义字段 package pro.shushi.pamirs.demo.core.config; import ch.qos.logback.classic.Level; import ch.qos.logback.classic.Logger; import ch.qos.logback.classic.LoggerContext; import ch.qos.logback.classic.spi.LoggerContextListener; import ch.qos.logback.core.Context; import ch.qos.logback.core.spi.ContextAwareBase; import ch.qos.logback.core.spi.LifeCycle; import java.net.InetAddress; import java.net.UnknownHostException; /** * Logback自定义字段 * * @author wx@shushi.pro * @date 2024/4/17 */ public class DemoLogbackFiledConfig extends ContextAwareBase implements LoggerContextListener, LifeCycle { private boolean started = false; @Override public boolean isResetResistant() { return false; } @Override public void onStart(LoggerContext loggerContext) { } @Override public void onReset(LoggerContext loggerContext) { } @Override public void onStop(LoggerContext loggerContext) { } @Override public void onLevelChange(Logger logger, Level level) { } @Override public void start() { if (started) { return; } Context context = getContext();…

    2024年5月18日
    1.5K00
  • Schedule相关

    1、Schedule初始化 TODO 2、Schedule执行器的入口 通常本地创建了Schedule,没有被正常执行,可以通过这个入口去排查问题 pro.shushi.pamirs.middleware.schedule.core.tasks.AbstractScheduleTaskDealSingle#selectTasks 3、Schedule执行环境隔离 项目中开发如果本地进行任务调试,通过通过指定ownSign进行环境隔离,如果不配置可能会导致这个任务被别的机器执行,本机的代码无法调试,如果开发的时候出现任务未执行可能是这个原因导致的 event: enabled: true schedule: enabled: true ownSign: dev_wx rocket-mq: namesrv-addr: 127.0.0.1:9876

    后端 2023年11月16日
    1.2K00
  • 【OpenGauss】后端部署使用OpenGauss高斯数据库

    Gauss数据库配置 适配版本 4.7.8.3之后的版本 配置步骤 Maven配置 去华为官网下周驱动包:gsjdbc4.jar;https://support.huaweicloud.com/mgtg-dws/dws_01_0032.html <dependency> <groupId>org.postgresql</groupId> <artifactId>gsjdbc</artifactId> <version>4</version> <scope>system</scope> <!– 下面两种方式都可以–> <systemPath>${pom.basedir}/libs/gsjdbc4.jar</systemPath> <!–<systemPath>/Users/wangxian/java-tools/guassdb/gsjdbc4.jar</systemPath>–> </dependency> 导入 scope 为 system 的包,spring 编译插件需要增加 includeSystemScope: true 配置。 <plugin> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-maven-plugin</artifactId> <configuration> <includeSystemScope>true</includeSystemScope> </configuration> <executions> <execution> <goals> <goal>repackage</goal> </goals> </execution> </executions> </plugin> JDBC连接配置 pamirs: datasource: pamirs: type: com.alibaba.druid.pool.DruidDataSource driverClassName: org.postgresql.Driver url: jdbc:postgresql://127.0.0.1:5432/pamirs?currentSchema=demo username: XXXXXX password: XXXXXX initialSize: 5 maxActive: 200 minIdle: 5 maxWait: 60000 timeBetweenEvictionRunsMillis: 60000 testWhileIdle: true testOnBorrow: false testOnReturn: false poolPreparedStatements: true asyncInit: true base: type: com.alibaba.druid.pool.DruidDataSource driverClassName: org.postgresql.Driver url: jdbc:postgresql://127.0.0.1:5432/pamirs?currentSchema=demo_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} 在pamirs连接配置时,${database}和${schema}必须完整配置,不可缺省。 其他连接参数如需配置,可自行查阅相关资料进行调优。 方言配置 pamirs方言配置 pamirs: dialect: ds: base: type: GaussDB version: 5 majorVersion: 5.0.1 pamirs: type: GaussDB version: 5 majorVersion: 5.0.1 数据库版本 type version majorVersion 5.x GaussDB 5 5.0.1 PS:由于方言开发环境为5.0.1版本,其他类似版本(5.x)原则上不会出现太大差异,如出现其他版本无法正常支持的,可在文档下方留言。 schedule方言配置 pamirs: event: enabled: true schedule: enabled: true dialect: type: GaussDB version: 5 major-version: 5.0.1 type version majorVersion GaussDB 5 5.0.1 PS:由于schedule的方言在多个版本中并无明显差异,目前仅提供一种方言配置。 其他配置 逻辑删除的值配置 pamirs: mapper: global: table-info: logic-delete-value: (EXTRACT(epoch FROM CURRENT_TIMESTAMP) * 1000000 + EXTRACT(MICROSECONDS FROM CURRENT_TIMESTAMP))::bigint Gauss数据库用户初始化及授权 — init root…

    2024年3月27日
    2.2K00

Leave a Reply

登录后才能评论