AREX 如何支持 Dubbo 自定义私有协议的录制回放

目前 AREX 支持录制使用 Dubbo2x、Dubbo3x 协议的接口请求，本文介绍了一种通用的方案，让大家可以自行二次开发以对各种自定义的 Dubbo 协议进行适配。

背景

AREX 是一款开源的基于真实请求与数据的自动化回归测试平台，利用 Java Agent 技术与比对技术，通过流量录制回放能力实现快速有效的回归测试。

Dubbo 是一款高性能的分布式服务框架，它基于 RPC 的服务调用和服务治理，具有透明化的远程调用、负载均衡、服务注册与发现、高度可扩展性、服务治理等特点。

目前 AREX 支持录制使用 Dubbo2x、Dubbo3x 协议的接口请求，并且在基于 Dubbo3x 版本开发的调度服务（Schedule Service）中支持回放使用这两种原生版本 Dubbo 协议的用例。但是对于用户自定义扩展的 Dubbo协议、序列化方式，甚至是基于 Dubbo 做的很多个性化改造（如dubbox），目前都不支持。

例如社区某用户与我们反馈，其公司使用的 Dubbo 协议是基于 Dubbox 做的改造，对于协议的包结构和序列化反序列化方式都做了改造，几乎可以理解为是一种全新的协议，这种特定类型的 Dubbo 协议无法使用 AREX 进行录制和回放操作。为了解决该问题，我们在主分支上 fork 出来了一个定制化版本，实现了对该协议的录制回放，但这种做法仅适用于该特定协议，缺乏普适性。

考虑到今后可能会有很多社区用户出现类似的问题，我们需要一种通用的方案，让用户可以自行二次开发以对各种自定义的 Dubbo 协议进行适配。本文将具体实现思路和细节进行了记录，希望对其他用户在今后的使用中提供参考。

方案

为方便理解，首先来介绍一下 AREX 各服务组件的工作流程。

架构图

在使用 AREX 流量录制功能时，AREX Java Agent 会记录生产环境中 Java 应用的数据流量和请求信息，并将这些信息发送给 AREX 数据存取服务（Storage Service），由数据存取服务导入 Mongodb 数据库中进行存储。当需要进行回放测试时，AREX 调度服务（Schedule Service）将会根据用户的配置和需求，通过数据存取服务从数据库中提取被测应用的录制数据（请求），然后向目标验证服务发送接口请求。同时，Java Agent 会将录制的外部依赖（外部请求/DB）的响应返回给被测应用，目标服务处理完成请求逻辑后返回响应报文。随后调度服务会将录制的响应报文与回放的响应报文进行比对，验证系统逻辑正确性，并将比对结果推送给分析服务（Report Service），由其生成回放报告，供测试人员检查。

问题的难点在于，由于调度服务是通过泛化调用来实现回放的，因此当某服务的 Dubbo 版本、协议、序列化方式与服务提供方（Provider）不一致时，就会出现各种问题导致调用失败。

AREX 希望能够尽可能地减少对用户服务的干扰，因此不希望关心用户服务的 Dubbo 配置。为了解决 Dubbo 版本、协议、序列化方式不一致的问题，可以考虑将回放请求的发送交给用户服务自己来实现，让用户服务自己去进行适配。

带着这个思路，我们考虑使用 Java SPI （Service Provider Interface，服务提供者接口）来解决问题。Java SPI 是 Java 中一种标准的服务发现机制。它可以帮助开发者在不修改代码的情况下扩展和替换程序中的组件。

用户只需要实现调度服务定义的回放接口，然后使用与录制时同样的框架来进行回放操作。在回放时，调度服务会通过 Java SPI 机制动态载入用户实现的接口。这样可以让用户服务自己去适配 Dubbo 版本、协议、序列化方式。对于原生的 Dubbo 协议，AREX 实现了它的扩展，并以缺省的方式加载了进来。

代码分析

SPI 定义

Invoker

public interface ReplayExtensionInvoker {
    /**
     * check caseType by InvokerConstants#XXXCaseType.
     */
    boolean isSupported(String caseType);

    ReplayInvokeResult invoke(ReplayInvocation invocation);

    default int order() {
        return 1;
    }
}

这里参考了 Dubbo 源码里 Invoker 的写法，定义了一个名为 ReplayExtensionInvoker 的接口，该接口定义了两个方法：isSupported 和 invoke，分别用于检查所支持的 caseType 类型和执行回放操作。invoke 方法的参数 Invocation 是一个打包好的请求，其中包含了请求的各种信息，如方法名、参数值等等。invoke 方法的返回值是一个 InvokeResult 对象，其中包含了回放的结果信息。接口中还定义了一个默认方法 order，用于指定实现类的执行顺序。

isSupported()

caseType 表示回放用例的协议类型，可以使用 InvokerConstant 类中预定义的 XXXCaseType 常量来表示。用户需要通过实现 ReplayExtensionInvoker 接口的 isSupported 方法，将自己实现的回放扩展与指定的 caseType 进行绑定（匹配）。这样，当 AREX 进行回放操作时，会自动根据回放用例中的协议类型，匹配对应的回放扩展，从而实现对应协议类型的回放功能。

order()

同一 caseType 可能会实现多个扩展，这里通过指定顺序来优先加载，降序排列。

Invocation

public interface ReplayInvocation {
    /**
     * eg: dubbo:127.0.0.1:20880
     * protocol + host + port
     */
    String getUrl();

    Map<String, Object> getAttributes();

    ReplayExtensionInvoker getInvoker();

    void put(String key, Object value);

    /**
     * Get specified class item from attributes.
     * key: refer to InvokerConstants.
     */
    <T> T get(String key, Class<T> clazz);

    /**
     * Get object item from attributes.
     * key: refer to InvokerConstants.
     */
    Object get(String key);
}

Invocation 被定义为一个接口，用户需要实现这个接口中的一些方法，以便在回放过程中获取请求参数。

getUrl()

返回请求的协议、主机和端口号，参考了其他协议，如 gRPC、Triple、Rest 等，Url 都是不必可少的。希望能够支持今后对于其他协议的扩展，将 Url 单独作为一个参数，由协议名+ IP 地址+端口号组合成。例如：dubbo:127.0.0.1:20880。

getAttributes()

调度服务将请求的其他参数统一打包在一个键值对类型为的 Map 对象里，其中 String 类型的 key 需要和 InvokerConstants 中约定的名称保持一致。在实现回放扩展时，用户可以通过调用 getAttributes() 方法获取这个 Map 对象，然后根据键的名称从 InvokerConstant 中获取对应的参数值，以便进行回放过程中的处理。

Result

public class ReplayInvokeResult {
    /**
     * invoke result.
     */
    private Object result;

    private Map<String, String> responseProperties;

    /**
     * if invoke failed.
     */
    privdate String errorMsg;

    /**
     * if invoke failed.
     */
    private Exception exception;
}

ReplayInvokeResult 类表示请求调用的结果，它包含以下字段：

result：请求调用的结果，存储在一个 Object 类型的变量中。
responseProperties：一些需要 provider 传递的隐式参数，存储在一个 Map。
类型的变量中。
errorMsg：如果请求调用失败，则存储错误信息的字符串。
exception：如果请求调用失败，则存储异常信息的对象。

扩展的加载

在扩展的加载过程中，首先通过获取当前类的 ClassLoader 来获得一个 URLClassLoader 对象，然后通过反射的方式调用其 addURL() 方法来加载扩展实现类。

loadJar() 方法接受一个 jarPath 参数，表示扩展实现类所在的 jar 包路径，然后通过判断当前 JDK 版本和获取 ClassLoader 等方式，得到一个 URLClassLoader 对象和 addURL() 方法的引用。接着，判断当前 ClassLoader 是否为 URLClassLoader 类型，如果是，则通过反射调用 addURL() 方法来加载 jar 包，并把 jar 包中的类添加到 ClassLoader 中，以便在程序运行时可以动态地调用扩展实现类的方法。如果加载失败，则输出错误信息。

public void loadJar(String jarPath) {
       try {
           int javaVersion = getJavaVersion();
           ClassLoader classLoader = this.getClassLoader();
           File jarFile = new File(jarPath);
           if (!jarFile.exists()) {
               LOGGER.error("JarFile doesn't exist! path:{}", jarPath);
           }
           Method addURL = Class.forName("java.net.URLClassLoader").getDeclaredMethod("addURL", URL.class);
           addURL.setAccessible(true);
           if (classLoader instanceof URLClassLoader) {
               addURL.invoke(classLoader, jarFile.toURI().toURL());
           }    
       } catch (Exception e) {
           LOGGER.error("loadJar failed, jarPath:{}, message:{}", jarPath, e.getMessage());
       }
   }

如何使用

步骤一：开发 SPI

1. 添加 pom 依赖

开发 SPI 扩展，首先需要在项目的 pom.xml 文件中添加 arex-schedule-extension 依赖。

      <dependency>
           <groupId>com.arextest</groupId>
           <artifactId>arex-schedule-extension</artifactId>
       </dependency>

2. 实现 SPI

实现 SPI 扩展，需要实现 com.arextest.schedule.extension.invoker.ReplayExtensionInvoker。

需要注意的是，在 provider 中，通过 Dubbo 隐式参数 attachments 中的 Arex-Record-Id 来判断当前流量是否为回放流量，并将生成的 Arex-Record-Id 也通过 attachments 传递回来。

因此，在实现 ReplayExtensionInvoker 接口时，需要从 ReplayInvocation#attributes 属性中获取 Arex-Record-Id，并将其添加到 attachments 中；同时，将 attachments 中的 Arex-Record-Id 添加到 ReplayInvokeResult#responseProperties 属性中，以便在回放流量中正确地使用这些参数。

以下是一个默认 dubboInvoker 的例子可以参考：

package com.arextest.dubboInvoker;

import com.arextest.schedule.extension.invoker.InvokerConstants;
import com.arextest.schedule.extension.invoker.ReplayExtensionInvoker;
import com.arextest.schedule.extension.invoker.ReplayInvocation;
import com.arextest.schedule.extension.model.ReplayInvokeResult;
import org.apache.dubbo.config.ApplicationConfig;
import org.apache.dubbo.config.ReferenceConfig;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.service.GenericService;

import java.util.List;
import java.util.Map;

public class DefaultDubboInvoker implements ReplayExtensionInvoker {

    @Override
    public boolean isSupported(String caseType) {
        return InvokerConstants.DUBBO_CASE_TYPE.equalsIgnoreCase(caseType);
    }

    @Override
    public ReplayInvokeResult invoke(ReplayInvocation replayInvocation) {
        ReplayInvokeResult replayInvokeResult = new ReplayInvokeResult();
        try {

            RpcContext.getServiceContext().setAttachments(replayInvocation.get(InvokerConstants.HEADERS, Map.class));
            ReferenceConfig<GenericService> reference = new ReferenceConfig<>();
            reference.setApplication(new ApplicationConfig("defaultDubboInvoker"));
            reference.setUrl(replayInvocation.getUrl());
            reference.setInterface(replayInvocation.get(InvokerConstants.INTERFACE_NAME, String.class));
            reference.setGeneric(true);
            GenericService genericService = reference.get();
            if (genericService == null) {
                return replayInvokeResult;
            }

            Object result = genericService.$invoke(replayInvocation.get(InvokerConstants.METHOD_NAME, String.class),
                    (String[]) replayInvocation.get(InvokerConstants.PARAMETER_TYPES, List.class).toArray(new String[0]),
                    (replayInvocation.get(InvokerConstants.PARAMETERS, List.class)).toArray());

            replayInvokeResult.setResult(result);
            replayInvokeResult.setResponseProperties(RpcContext.getServerContext().getAttachments());
        } catch (Exception e) {
            replayInvokeResult.setException(e);
            replayInvokeResult.setErrorMsg(e.getMessage());
        }
        return replayInvokeResult;
    }

}

3. 创建 SPI 扩展文件

在项目的 resources/META-INF/services 目录下创建一个文本文件，文件名为 com.arextest.schedule.extension.invoker.ReplayExtensionInvoker。文件内容为扩展实现类的全限定名，即实现了 ReplayExtensionInvoker 接口的类的 Reference。

4. 打 jar 包

将实现的 SPI 扩展打包成一个 jar 包，注意需要包含所有相关的依赖。

步骤二：导入 AREX 项目

1. jar 包映射 Docker 镜像

首先需要将 jar 包映射到 Docker 镜像中，这可以通过修改 docker-compose.yml 文件来实现，将本地目录映射到 Docker 镜像，并将 jar 包添加到本机中对应的目录下。

jar 包映射

2. 修改调度服务启动参数

接着，要修改调度服务（Schedule Service）的启动参数，同样是通过修改 docker-compose.yml 文件来实现。将以下启动参数添加到 Schedule 的启动参数中。

-Dreplay.sender.extension.jarPath=/usr/local/tomcat/custom-dubbo/xxx.jar

其中 xxx.jar 是上一步打包的 jar 包的名称， /usr/local/tomcat/custom-dubbo/ 是在 Docker 镜像中的 jar 包存放路径。这个启动参数的作用是告诉 AREX 在哪里可以找到 SPI 扩展的 jar 包。

展望

未来还有三个方面可以进行改进：

jar 包的加载模式

目前，用户需要手动管理配置 jar 包路径，并且需要手动修改 docker-compose.yml 文件，这样操作上略微有些繁琐。未来希望能够做到在 AREX 平台上进行相关参数配置，上传 jar 包等操作。

参数约定

目前参数的管理是在代码中以 hardcode 的方式实现的，新的接入方需要提前约定好，再更新 SPI 这个包。未来希望做到在 AREX 平台上进行参数配置，这样新的接入方可以通过简单的配置实现自己的需求。

其他协议的支持

目前 invoker 框架可以直接支持 Dubbo 协议的各种个性化扩展。如果有用户需要接入其他的协议，例如 gRPC、Triple、Rest 等，也可以提出 [issue] (https://github.com/arextest/arex/issues)，在社区中进行讨论，以便扩展 invoker 框架的适用范围。

背景​

方案​

代码分析​

SPI 定义​

Invoker​

Invocation​

Result​

扩展的加载​

如何使用​

步骤一：开发 SPI​

1. 添加 pom 依赖​

2. 实现 SPI​

3. 创建 SPI 扩展文件​

4. 打 jar 包​

步骤二：导入 AREX 项目​

1. jar 包映射 Docker 镜像​

2. 修改调度服务启动参数​

展望​

jar 包的加载模式​

参数约定​

其他协议的支持​

背景

方案