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日
    1.3K00
  • Dubbo配置详解

    概述 Dubbo是一款高性能、轻量级的开源Java RPC框架,它提供了三大核心能力:面向接口的远程方法调用,智能容错和负载均衡,以及服务自动注册和发现。 Oinone平台默认使用dubbo-v2.7.22版本,本文以该版本为例进行描述。 基本概念 Dubbo在注册provider/consumer时使用Netty作为RPC调用的核心服务,其具备客户端/服务端(C/S)的基本特性。即:provider作为服务端,consumer作为客户端。 客户端通过服务中心发现有服务可被调用时,将通过服务中心提供的服务端调用信息,连接服务端并发起请求,从而实现远程调用。 服务注册(绑定Host/Port) JAVA程序启动时,需要将provider的信息注册到服务中心,并在当前环境为Netty服务开启Host/Port监听,以实现服务注册功能。 在下文中,我们通过绑定Host/Port表示Netty服务的访问地址,通过注册Host/Port表示客户端的访问地址。 使用yaml配置绑定Host/Port PS:该配置可在多种环境中通用,改变部署方式无需修改此配置。 dubbo: protocol: name: dubbo # host: 0.0.0.0 port: -1 假设当前环境的可用IP为192.168.1.100 以上配置将使得Netty服务默认绑定在0.0.0.0:20880地址,服务注册地址为192.168.1.100:20880 客户端将通过192.168.1.100:20880调用服务端服务 若发生20880端口占用,则自动向后查找可用端口。如20881、20882等等 若当前可用端口为20881,则以上配置将使得Netty服务默认绑定在0.0.0.0:20881地址,服务注册地址为192.168.1.100:20881 使用环境变量配置注册Host/Port 当服务端被放置在容器环境中时,由于容器环境的特殊性,其内部的网络配置相对于宿主机而言是独立的。因此为保证客户端可以正常调用服务端,还需在容器中配置环境变量,以确保客户端可以通过指定的注册Host/Port进行访问。 以下示例为体现无法使用20880端口的情况,将宿主机可访问端口从20880改为20881。 DUBBO_IP_TO_REGISTRY=192.168.1.100 DUBBO_PORT_TO_REGISTRY=20881 假设当前宿主机环境的可用IP为192.168.1.100 以上配置将使得Netty服务默认绑定在0.0.0.0:20881地址,服务注册地址为192.168.1.100:20881 客户端将通过192.168.1.100:20881调用服务端服务 使用docker/docker-compose启动 需添加端口映射,将20881端口映射至宿主机20881端口。(此处容器内的端口发生变化,若需要了解具体原因,可参考题外话章节) docker-run IP=192.168.1.100 docker run -d –name designer-allinone-full \ -e DUBBO_IP_TO_REGISTRY=$IP \ -e DUBBO_PORT_TO_REGISTRY=20881 \ -p 20881:20881 \ docker-compose services: backend: container_name: designer-backend image: harbor.oinone.top/oinone/designer-backend-v5.0 restart: always environment: DUBBO_IP_TO_REGISTRY: 192.168.1.100 DUBBO_PORT_TO_REGISTRY: 20881 ports: – 20881:20881 # dubbo端口 使用kubernetes启动 工作负载(Deployment) kind: Deployment apiVersion: apps/v1 spec: replicas: 1 template: spec: containers: – name: designer-backend image: harbor.oinone.top/oinone/designer-backend-v5.0 ports: – name: dubbo containerPort: 20881 protocol: TCP env: – name: DUBBO_IP_TO_REGISTRY value: "192.168.1.100" – name: DUBBO_PORT_TO_REGISTRY value: "20881" 服务(Services) kind: Service apiVersion: v1 spec: type: NodePort ports: – name: dubbo protocol: TCP port: 20881 targetPort: dubbo nodePort: 20881 PS:此处的targetPort为对应Deployment#spec. template.spec.containers.ports.name配置的端口名称。若未配置,可使用20881直接指定对应容器的端口号。 使用kubernetes其他暴露服务方式 在Kubernetes中部署服务,有多种配置方式均可用暴露服务。上述配置仅用于通过Service/NodePort将20881端口暴露至宿主机,其他服务可用通过任意Kubernetes节点IP进行调用。 若其他服务也在Kubernetes中进行部署,则可以通过Service/Service方式进行调用。将DUBBO_IP_TO_REGISTRY配置为${serviceName}.${namespace}即可。 若其他服务无法直接访问Kubernetes的master服务,则可以通过Ingress/Service方式进行调用。将DUBBO_IP_TO_REGISTRY配置为Ingress可解析域名即可。 Dubbo调用链路图解 PS: Consumer的绑定Host/Port是其作为Provider使用的,下面所有图解仅演示单向的调用链路。 名词解释 Provider: 服务提供者(JVM) Physical Machine Provider: 服务提供者所在物理机 Provider Container: 服务提供者所在容器 Kubernetes Service: Kubernetes Service资源类型 Consumer: 服务消费者(JVM) Registration Center: 注册中心;可以是zookeeper、nacos等。 bind: 服务绑定Host/Port到指定ip:port。 registry: 服务注册;注册Host/Port到注册中心的信息。 discovery: 服务发现;注册Host/Port到消费者的信息。 invoke: 服务调用;消费者通过注册中心提供的提供者信息向提供者发起服务调用。 forward: 网络转发;通常在容器环境需要进行必要的网络转发,以使得服务调用可以到达服务提供者。 物理机/物理机调用链路 “` mermaidsequenceDiagram participant p as Provider<br>(bind 0.0.0.0:20880)participant m as Physical Machine Provider<br>(bind 192.168.1.100:20881)participant…

    2024年8月10日
    1.7K00
  • 环境运行时Jar版本控制

    环境运行时Jar版本控制 前景 为了避免基于低代码定义产生的元数据错乱。因此产生了运行时Jar版本检查功能。 现象 如果当前运行时依赖的Ja版本低于已安装版本,启动时会有如下类似信息提示: 解决 按照提示升级依赖Jar版本 通过启动参数 -PgoBack=true 强制覆盖安装当前运行时版本 java -jar 方式 java -jar xxx.jar -PgoBack=true [其他参数] mvn spring-boot run 方式 mvn clean compile spring-boot:run -Dspring-boot.run.arguments=”-PgoBack=true [其他参数]”

    2025年3月10日
    85700
  • 工作流-流程代办等页面自定义

    1. 审批/填写节点的视图页面 在界面设计器中创建对应模型的表单视图,可根据业务场景需要自定义所需流程待办的审批页面 2. 在审批/填写节点中选择刚创建的视图 在工作流待办数据权限可在节点数据权限中可对字段设置查看、编辑、隐藏

    2024年5月14日
    1.6K00
  • 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.3K00

Leave a Reply

登录后才能评论