分布式链路追踪 —— 基于Dubbo的traceId追踪传递

文章目录

原文链接
RpcContext 上下文对象
Dubbo 过滤器（Filter）对象
基于Dubbo的traceId追踪传递实现

原文链接

RpcContext 上下文对象

在实现 Dubbo 调用之间的链路跟踪之前，先简单了解 RpcContext 上下文对象和 Filter 过滤器对象，Dubbo 分布式链路追踪是基于这两个对象实现。

概要

RpcContext 是 Dubbo 框架提供的一个类，它的设计目标是提供一个 dubbo 调用的上下文对象，用于在远程过程调用（RPC）期间传递、共享请求和响应的相关信息。它可以用于存储、访问与当前 RPC 调用相关的数据，如调用方的 IP 地址、附加参数、上下文变量等。它提供了一些静态方法和属性，可以方便地获取和设置与当前线程相关的 RPC 上下文。

public class RpcContext {

    private static final InternalThreadLocal<RpcContext> LOCAL = new InternalThreadLocal<RpcContext>() {
        protected RpcContext initialValue() {
            return new RpcContext();
        }
    };
    private static final InternalThreadLocal<RpcContext> SERVER_LOCAL = new InternalThreadLocal<RpcContext>() {
        protected RpcContext initialValue() {
            return new RpcContext();
        }
    };
    private final Map<String, String> attachments = new HashMap();
    private final Map<String, Object> values = new HashMap();
    // ......
}

原理

RpcContext 是基于 ThreadLocal 实现的，做到了线程隔离。RpcContext是与线程绑定的，每个线程都有自己的一个 RpcContext 实例并使用 ThreadLocal 变量来存储，避免并发访问问题。当客户端发起一个 RPC 请求时，Dubbo 框架会创建一个新的线程来处理该请求，并且会将 RpcContext 与该线程进行绑定，这样看，对于每次 RPC 请求，RpcContext 也是唯一的。但对于同一个线程内的多个 RPC 请求，它们共享同一个 RpcContext 实例。
RpcContext 实例会在请求处理期间一直存在，并在请求处理完成后需要清理当前线程上的 RpcContext 实例中的数据，以确保下次使用该线程处理新的请求时，RpcContext 是一个干净的状态。

对于服务消费方，Dubbo 框架在请求发送、响应后没有清除 RpcContext 实例中的数据；

对于服务提供方，Dubbo 框架在收到请求并处理后，会去清除 RpcContext 实例中的数据，这个在 ContextFilter 服务提供方过滤器中可以看到。

使用场景

可以在 dubbo 的拦截器、过滤器或服务提供者/消费者的代码中使用 RpcContext 来获取和设置上下文信息，以满足特定的业务需求，如日志跟踪、传递身份验证信息等。

下面是一些常用的 RpcContext 方法和属性：

RpcContext.getContext(): 获取当前线程的 RpcContext 实例。
RpcContext.isConsumerSide(): 判断当前线程是否处于消费者端。
RpcContext.isProviderSide(): 判断当前线程是否处于提供者端。
RpcContext.getRemoteAddress(): 获取远程调用的地址。
RpcContext.getLocalAddress(): 获取本地调用的地址。
RpcContext.setAttachment(String key, String value): 设置附加参数。
RpcContext.getAttachment(String key): 获取指定键的附件信息。
RpcContext.getAttachments(): 获取所有的附件信息。

Dubbo 过滤器（Filter）对象

Dubbo Filter 介绍

dubbo 的 Filter 是 dubbo 框架提供的一个功能扩展点，用于对服务提供者和消费者之间的请求和响应进行拦截过滤处理，比如认证和授权、日志跟踪、传递一些公共信息等。

如 dubbo 原生 Filter 实现类，如：ConsumerContextFilter 和 ContextFilter

ConsumerContextFilter 是一个服务消费方的过滤器，用于在服务消费者发起 RPC 调用之前或之后，对上下文信息进行处理和传递，用于收集和发送调用方的上下文信息到服务提供者端。

ConsumerContextFilter 源码：通过这个过滤器可以看到在 RPC 调用之前会获取 RpcContext 对象并设置相关参数，Dubbo 框架会借助 RpcContext 对象将相关数据透传到服务提供方。
```
@Activate(
    group = {"consumer"},
    order = -10000
)
public class ConsumerContextFilter implements Filter {
    public ConsumerContextFilter() {
    }

    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        RpcContext.getContext().setInvoker(invoker).setInvocation(invocation).setLocalAddress(NetUtils.getLocalHost(), 0).setRemoteAddress(invoker.getUrl().getHost(), invoker.getUrl().getPort()).setRemoteApplicationName(invoker.getUrl().getParameter("remote.application")).setAttachment("remote.application", invoker.getUrl().getParameter("application"));
        if (invocation instanceof RpcInvocation) {
            ((RpcInvocation)invocation).setInvoker(invoker);
        }

        return invoker.invoke(invocation);
    }
}
```
@Activate 注解是 Dubbo 框架提供的一个扩展点激活注解，用于指定在特定条件下激活扩展点。

在上述过滤器中，@Activate(group = Constants.CONSUMER, order = -10000) 是对一个扩展点的激活配置。具体解释如下：
- group = {"consumer"}: 在指定的分组激活，如 consumer，表示该扩展点在消费者端被激活。
- order = -10000: 指定激活的顺序为 -10000。在 Dubbo 框架中，扩展点的激活顺序可以通过 order 值来进行控制，值越小表示优先级越高。

ContextFilter 是一个服务提供方的过滤器，用于在服务提供者收到 RPC 调用请求之前或之后，对上下文信息进行处理和传递，如清除 RpcContext 实例中的数据，以确保下次使用该线程处理新的请求时，RpcContext 是一个干净的状态。

ContextFilter 源码：在处理请求后去清除 RpcContext 中相关数据。

@Activate(
    group = {"provider"},
    order = -10000
)
public class ContextFilter implements Filter, Filter.Listener {
    private static final String TAG_KEY = "dubbo.tag";
    public ContextFilter() {
    }
    
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        // ......
        Result var6;
        try {
            // 禁止清除RpcContext的功能。这意味着在调用invoker.invoke(invocation)之后，RpcContext中的上下文信息不能被清除。
            RpcContext.getContext().clearAfterEachInvoke(false);
            var6 = invoker.invoke(invocation);
        } finally {
            // 开启允许清除RpcContext的功能。
            RpcContext.getContext().clearAfterEachInvoke(true);
            // 显式清除RpcContext中的上下文信息。
            RpcContext.removeContext();
            RpcContext.removeServerContext();
        }

        return var6;
    }
    // ......
}

自定义 Dubbo Filter

实现 org.apache.dubbo.rpc.Filter接口：

import org.apache.dubbo.rpc.*;

public class CustomFilter implements Filter {
    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        // 自定义过滤逻辑
        return invoker.invoke(invocation);
    }
}

创建 Dubbo 的 SPI 扩展文件（META-INF/dubbo/org.apache.dubbo.rpc.Filter）中，将自定义过滤器的实现类指定为对应的扩展点：
```
customFilter=com.example.CustomFilter
```
使过滤器生效：

方式一：在自定义过滤器上使用注解：
```
@Activate(group = {"consumer"})
```
方式二：在配置文件（这里使用application.properties）中配置：
```
dubbo.consumer.filter=customFilter
```
通过以上步骤，在服务消费方指定了一个自定义过滤器，该过滤器将在服务消费者发起远程调用前后执行自定义的逻辑。

基于Dubbo的traceId追踪传递实现

要实现在 Dubbo 接口之间传递 TraceID，可以使用 Dubbo 的拦截器（Filter）机制来实现。下面是一个示例代码，演示了如何在 Dubbo 接口调用中传递 TraceID 进行追踪，其中具体 Filter 实现过程前面已讲述，这里只展示实现类代码。需要 demo 示例代码，请关注【Qin的学习营地】，回复【基于Dubbo的traceId追踪传递】。

这里使用 spring boot 整合 dubbo，详细搭建过程请参考：Dubbo 快速入门使用教程

这里通过打印日志来可视化结果，使用了 Slf4J 的 MDC，通过设置 MDC.put(key, value)，并在日志配置文件中配置 key，日志打印时会将配置 key 的地方转换为 value 打印出来。

创建 Dubbo 的服务提供方拦截器类，从 RpcContext 中获取 traceid 参数，并设置到 MDC 中，请求处理完后清除 MDC 中的 traceid 参数：

@Slf4j
@Activate(group = {"provider"})
public class TraceIdProviderFilter implements Filter {
    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        String traceId = RpcContext.getContext().getAttachment("traceId");
        if (traceId != null) {
            MDC.put("traceId", traceId);
        }
        try {
            return invoker.invoke(invocation);
        } finally {
            MDC.remove("traceId");
        }
    }
}

创建 Dubbo 的服务消费方拦截器类，向 RpcContext 中写入 traceid 参数：

@Slf4j
@Activate(group = {"consumer"})
public class TraceIdConsumerFilter implements Filter {
    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        String traceId = MDC.get("traceId");
        if (traceId == null) {
            traceId = UUID.randomUUID().toString().replace("-", "");
        }
        RpcContext.getContext().setAttachment("traceId", traceId);
        MDC.put("traceId", traceId);
        log.info("consumer ——> provider");
        return invoker.invoke(invocation);
    }
}

服务消费方调用逻辑：

@Slf4j
@Component
public class ProducerService {

    @Reference(retries = -1, version="1.0.0", timeout = 15000)
    private HelloService helloService;

    public String consumerSayHello(String name){
        String traceId = UUID.randomUUID().toString().replace("-", "");
        MDC.put("traceId", traceId);
        String hello = helloService.sayHello(name);
        log.info("consumer receive response : "+ hello);
        return hello;
    }
}