使用Dubbo的RpcContext居然那么多坑-阿里云开发者社区

一、前言

前一段时间在公司写了一个链路追踪的服务，其中SpringMVC做为门面对外提供服务，微服务之间采用Dubbo接口调用。对于Dubbo接口之间传递链路信息，采用RpcContext将需要的参数透传过去。然而在使用RpcContext时遇到了几个问题导致RpcContext未按我设想的方式传递。

二、RpcContext介绍

RpcContext 本质上是一个使用 ThreadLocal 实现的临时状态记录器，当接收到 RPC 请求，或发起 RPC 请求时，RpcContext 的状态都会变化。
比如：A调B，B再调C，则B机器上，在B调C之前，RpcContext记录的是A调B的信息，在B调C之后，RpcContext记录的是B调C的信息。

RpcContext使用ThreadLocal的部分源码如下：

public class RpcContext {

    /**
     * 存放主体内容
     */
    private static final ThreadLocal<RpcContext> LOCAL = new ThreadLocal<RpcContext>() {
        protected RpcContext initialValue() {
            return new RpcContext();
        }
    };

    /**
     * 获取RpcContext信息
     */
    public static RpcContext getContext() {
        return (RpcContext)LOCAL.get();
    }

    /**
     * 清空RpcContext信息
     */
    public static void removeContext() {
            LOCAL.remove();
        }
}

注意：不同Dubbo版本的RpcContext略有区别，本质上都是使用的ThreadLocal。

二、RpcContext的使用

//服务提供方使用,获取参数
RpcContext.getContext().getAttachments()
//服务器消费方使用,设置参数
RpcContext.getContext().setAttachment()

1> 消费端（DubboConsumer）：

// 远程调用之前，通过attachment传KV给提供方  
RpcContext.getContext().setAttachment("userKey", "userValue");  
// 远程调用  
xxxService.xxx();  
// 此时 RpcContext 的状态已变化   
RpcContext.getContext();

2> 服务端（DubboProvider）：

public class XxxServiceImpl implements XxxService {  
  
    public void xxx() {  
        // 通过RpcContext获取用户传参，这里会返回userValue  
        String value = RpcContext.getContext().getAttachment("userKey");  
        // 本端是否为提供端，这里会返回true  
        boolean isProviderSide = RpcContext.getContext().isProviderSide();  
        // 获取调用方IP地址  
        String clientIP = RpcContext.getContext().getRemoteHost();  
        // 获取当前服务配置信息，所有配置信息都将转换为URL的参数  
        String application = RpcContext.getContext().getUrl().getParameter("application");   
    }   
}

三、结合Filter使用RpcContext

一般修改RpcContext信息都是在Dubbo的拦截器中，这样有两个好处：

统一入口设置参数，方便维护。
解决一次完整请求调用涉及多次嵌套RPC调用时获取不到上下文中设置的参数值问题。

例如：

1> 在DubboConsumerFilter中设置RpcContext信息：

@Slf4j
@Activate(group = {CommonConstants.CONSUMER})
public class DubboConsumerFilter implements Filter {

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {

        try {
            // 服务端从dubbo上下文中取出rpcContext信息
            String jsonStr = null;
            if (!CollectionUtils.isEmpty(RpcContext.getContext().getObjectAttachments())) {
                jsonStr = RpcContext.getContext().getAttachment(HttpHeaderKeys.APP_NAME.getKey());
            }
            // 获取不到rpcContext信息，则手动塞入
            if (StringUtils.isEmpty(jsonStr)) {
                RpcContext.getContext().setAttachment("userKey", "saint");
            }
        } catch (Exception e){
            log.error("Exception in process DubboConsumerFilter" ,e);
            // do nothing
        }

        return invoker.invoke(invocation);

    }
}

2> 在DubboProviderFilter中获取RpcContext信息：

@Slf4j
@Activate(group = {CommonConstants.PROVIDER})
public class DubboProviderFilter implements Filter {

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {

        // 服务端从dubbo上下文中取出traceContext信息
        String jsonStr = null;
        if (!CollectionUtils.isEmpty(RpcContext.getContext().getObjectAttachments())) {
            jsonStr = RpcContext.getContext().getAttachment(HttpHeaderKeys.APP_NAME.getKey());
            log.info("DubboProviderFilter get dubbo RpcContext is : {}", jsonStr);
        }
        return invoker.invoke(invocation);

    }
}

四、使用RpcContext的坑

1、一个dubbo接口调用多个dubbo接口，RpcContext会改变

一个dubbo接口同步调用多个dubbo接口（比如Dubbo接口B和Dubbo接口C）时，在调用Dubbo接口C时，RPCContext已经发生改变了，需要重新获取调用链路信息。

原因分析见后面的RpcContext原理。
在这里插入图片描述

参考解决方案

可以采用ThreadLocal保存，然后在ProviderFilter中清除ThreadLocal，防止数据错乱，但是会造成一定的内存泄漏（数据量较小是可以接收的）。

0> 用于存储调用链路信息的ThreadLocal：

import lombok.Data;
import lombok.experimental.Accessors;

import java.io.Serializable;

/**
 * 链路信息上下文
 *
 * @author Saint
 */
public class RpcTraceContext implements Serializable {

    private final static ThreadLocal<TraceContext> traceContextHolder = new ThreadLocal<>();


    public static ThreadLocal<TraceContext> get() {
        return traceContextHolder;
    }

    /**
     * 设置traceContext
     *
     * @param traceContext traceContext
     */
    public static void setTraceContext(TraceContext traceContext) {
        traceContextHolder.set(traceContext);
    }

    /**
     * 获取traceContext
     *
     * @return traceContext
     */
    public static TraceContext getTraceContext() {
        return traceContextHolder.get();
    }

    /**
     * 清空trace上下文
     */
    public static void clear() {
        traceContextHolder.remove();
    }

    @Data
    @Accessors(chain = true)
    public static class TraceContext implements Serializable {
        private Long userId;
        private String traceId;
        private String controllerAction;
        private String visitIp;
        private String appName;
    }
}

1> DubboProviderFilter：

@Slf4j
@Activate(group = {CommonConstants.CONSUMER})
public class DubboConsumerFilter implements Filter {

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        
        try {
            //服务端从dubbo上下文中取出traceContext信息
            String jsonStr = null;
            if (!CollectionUtils.isEmpty(RpcContext.getContext().getAttachments())) {
                jsonStr = RpcContext.getContext().getAttachment("TRACE_CONTEXT");
            }
            if (StringUtils.isNotEmpty(jsonStr)) {
                // 这里是为了解决duboo接口调用多个dubbo接口，第一个dubbo接口之后的dubbo接口获取到的RpcContext为空的问题。
                RpcTraceContext.TraceContext traceContext = JSON.parseObject(jsonStr, RpcTraceContext.TraceContext.class);
                RpcTraceContext.setTraceContext(traceContext);
            }
        } catch (Exception e){
            log.error("Exception in process DubboConsumerFilter" ,e);
        }

        return invoker.invoke(invocation);
    }
}

2> DubboProviderFilter：

@Slf4j
@Activate(group = {CommonConstants.PROVIDER})
public class DubboProviderFilter implements Filter {

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {

        RpcTraceContext.clear();

        // 服务端从dubbo上下文中取出traceContext信息
        String jsonStr = null;
        if (!CollectionUtils.isEmpty(RpcContext.getContext().getAttachments())) {
            jsonStr = RpcContext.getContext().getAttachment("TRACE_CONTEXT");
        }
        return invoker.invoke(invocation);

    }
}

2、异步调用的两个坑

1）异步调用依赖传递性

问题表现：

如果consumer-A异步调用provider-B，而provider-B本身又调用了provider-C。当provider-B调用provider-C时，会变成异步。

问题原因：

是否异步调用取决于RpcContext中async的值，其次才是服务本身的配置。
当A调用B时，会把async=true传给B的RpcContext；B调用C时，虽然服务本身async=false，但RpcContext中async=true，自然也就成了异步调用。

2）异步回调返回null

问题表现：

consumer-A调用provider-B，而provider-B本身又调用了provider-C。consumer-A调用provider-B返回null。

问题原因：

异步调用直接返回空的RpcResult，需要后序通过RpcContext.getContext().getFuture() .get()获取返回值。
async透传到provider-B端之后，也是异步调用provider-C，但是直接返回空的RpcResult给consumer-A。

3）解决方案

不让async参数应用到provider端。需要修改ContextFilter源码，重写RpcContext时删除async参数；
在这里插入图片描述

五、RpcContext原理

首先RpcContext内部有一个ThreadLocal变量（高版本用的InternalThreadLocal本质上也是ThreadLocal），它是作为ThreadLocalMap的key，表明每个线程有一个RpcContext。
在这里插入图片描述

其次Dubbo内嵌了两Filter，分别为：ContextFilter、ConsumerContextFilter，分别用来拦截Dubbo服务提供者和消费者。

1、ConsumerContextFilter

消费端在执行Rpc调用之前，经过Filter处理, 会将一些信息（比如：服务调用信息）写入RpcContext。
在这里插入图片描述

2、ContextFilter

服务端在执行调用之前，也会经过Filter处理，将信息写入RpcContext；最后清空RpcContext。
在这里插入图片描述

使用Dubbo的RpcContext居然那么多坑