Hystrix实现ThreadLocal上下文的传递_requestcontextholder.currentrequestattributes().se

本章概要

1、资料搜索；

2、源码分析；

3、扩展HystrixConcurrencyStrategy解决前言中的意外；

4、提高HystrixConcurrencyStrategy包装扩展性；

资料搜索

既然遇到了问题，就到springcloud的官方文档先检索下，找到如下对应的描述

红色框部分主要意思是，我们可以声明一个定制化的HystrixConcurrencyStrategy实例，并通过HystrixPlugins注册。先找到HystrixConcurrencyStrategy类，其有下面一段类注释

For example, every {@link Callable} executed by {@link HystrixCommand} will call {@link #wrapCallable(Callable)} to give a chance for custom implementations to decorate the {@link Callable} with additional behavior.

被 @HystrixCommand注解的方法，其执行源 Callable可以通过 wrapCallable方法进行定制化装饰，加入附加的行为，继续来看看 wrapCallable方法的定义

其同样提供了非常详细的注释，该方法提供了在方法被执行前进行装饰的机会，可以用来复制线程状态等附加行为，这个貌似就是我们需要的，很合意。

同样在Hystrix官方文档提供了更加详细的说明（ https://github.com/Netflix/Hystrix/wiki/Plugins#concurrency-strategy）， Concurrency Strategy 作为了Plugin的一种类别，描述如下

可以看到红色框中的重点描述，其已经说了非常明确，可以从父线程复制线程状态至子线程。自定义的Plugin如何被 HystrixCommand应用呢，继续查看官方的描述

其提供了 HystrixPlugins帮助我们注册自定义的Plugin，除了我们本章节重点关注的 Concurrency Strategy类别plugin，还有如下类别以及对应的抽象实现

源码分析

在springcloud中还有如下一段话

既然提高了定制化的实现，不如来看看官方已经提供了哪些默认实现

首先来看看 HystrixConcurrencyStrategyDefault，

很精简的一段代码，并没有任何方法重写，其作为了一个标准提供默认实现。继续来看看 SecurityContextConcurrencyStrategy实现，直接找到wrapCallable方法

其对Callabe进行了二次包装，继续跟进来看看 DelegatingSecurityContextCallable的定义

其主要实现均在call方法中，红色框中标出了重点，在调用call方法前，我们可以将当前上下文信息放入 SecurityContextHolder中，在执行完成后清空 SecurityContextHolder对应的设置。再来看看 SecurityContextConcurrencyStrategy是如何被应用的，在 HystrixSecurityAutoConfiguration中有如下代码段

在启动注册配置过程中机会通过 HystrixPlugins注册当前扩展的 HystrixConcurrencyStrategy实现。

小节：自定义扩展类实现Callable接口，并传入当前Callable变量delegate，在delegate执行call方法前后进行线程上线文的操作即可实现线程状态在父线程与子线程间的传播。

扩展HystrixConcurrencyStrategy解决前言中的意外

通过源码部分的解读，基本了解springcloud是如何实现扩展的，又是如何被应用的，照葫芦画瓢下。

public class RequestContextHystrixConcurrencyStrategy extends HystrixConcurrencyStrategy {
 
    @Override
    public <T> Callable<T> wrapCallable(Callable<T> callable) {
        return new RequestAttributeAwareCallable<>(callable, RequestContextHolder.getRequestAttributes());
    }
 
    static class RequestAttributeAwareCallable<T> implements Callable<T> {
 
        private final Callable<T> delegate;
        private final RequestAttributes requestAttributes;
 
        public RequestAttributeAwareCallable(Callable<T> callable, RequestAttributes requestAttributes) {
            this.delegate = callable;
            this.requestAttributes = requestAttributes;
        }
 
        @Override
        public T call() throws Exception {
            try {
                RequestContextHolder.setRequestAttributes(requestAttributes);
                return delegate.call();
            } finally {
                RequestContextHolder.resetRequestAttributes();
            }
        }
    }
}

其中定义 RequestAttributeAwareCallable装饰类，通过构造函数传入当前待执行 Callable代理和当前待传播的 RequestAttributes值，并在 delegate的 call方法执行前对 RequestContextHolder的 RequestAttributes赋值，在 finally块中重置。

2、同样在任意配置类中添加如下代码段即可，通过 HystrixPlugins注册 RequestContextHystrixConcurrencyStrategy：

@PostConstruct
public void init() {
    HystrixPlugins.getInstance().registerConcurrencyStrategy(new RequestContextHystrixConcurrencyStrategy());
}

3、启动服务验证，子线程取值成功：

小节：以上参考 SecurityContextConcurrencyStrategy的实现，完成了Hystrix中 RequestContextHolder上下文信息传播。

提高HystrixConcurrencyStrategy包装扩展性

上一个小节介绍了如果在Hystrix线程隔离场景下实现 ThreadLocal定义的上下文传播，根据示例，在实际应用过程中如果我们有多个类似 RequestContextHystrixConcurrencyStrategy策略，需要将每个自定义 HystrixConcurrencyStrategy示例注册至 HystrixPlugins中，这在扩展性方面显然是缺失的，借鉴spring的实践，我们可以定义对Callable的包装接口 HystrixCallableWrapper，根据实际的业务只需要对 HystrixCallableWrapper进行实现，并注册对应的实现bean即可。具体实现如下：

1、定义用于包装hystrix中Callable实例的接口：

public interface HystrixCallableWrapper {
 
    /**
     * 包装Callable实例
     *
     * @param callable 待包装实例
     * @param <T>      返回类型
     * @return 包装后的实例
     */
    <T> Callable<T> wrap(Callable<T> callable);
 
}

2、通过之前的源码阅读与实践，基本已经发现实现线程上线文传播的核心在于对Callable进行包装，通过多次对Callable包装即实现了一个链式包装过程，如下扩展 HystrixConcurrencyStrategy接口实现 RequestContextHystrixConcurrencyStrategy，其中定义 CallableWrapperChain类对所有注入的 HystrixCallableWrapper包装实现进行装配:

public class RequestContextHystrixConcurrencyStrategy extends HystrixConcurrencyStrategy {
    private final Collection<HystrixCallableWrapper> wrappers;
 
    public RequestContextHystrixConcurrencyStrategy(Collection<HystrixCallableWrapper> wrappers) {
        this.wrappers = wrappers;
    }
 
    @Override
    public <T> Callable<T> wrapCallable(Callable<T> callable) {
        return new CallableWrapperChain(callable, wrappers.iterator()).wrapCallable();
    }
 
    private static class CallableWrapperChain<T> {
 
        private final Callable<T> callable;
 
        private final Iterator<HystrixCallableWrapper> wrappers;
 
        CallableWrapperChain(Callable<T> callable, Iterator<HystrixCallableWrapper> wrappers) {
            this.callable = callable;
            this.wrappers = wrappers;
        }
 
        Callable<T> wrapCallable() {
            Callable<T> delegate = callable;
            while (wrappers.hasNext()) {
                delegate = wrappers.next().wrap(delegate);
            }
            return delegate;
        }
    }
}

3、实现 HystrixCallableWrapper接口，定义一个包装 RequestContextHolder上下文处理的实现类：

public final class RequestAttributeAwareCallableWrapper implements HystrixCallableWrapper {
    @Override
    public <T> Callable<T> wrap(Callable<T> callable) {
        return new RequestAttributeAwareCallable(callable, RequestContextHolder.getRequestAttributes());
    }
 
    static class RequestAttributeAwareCallable<T> implements Callable<T> {
 
        private final Callable<T> delegate;
        private final RequestAttributes requestAttributes;
 
        RequestAttributeAwareCallable(Callable<T> callable, RequestAttributes requestAttributes) {
            this.delegate = callable;
            this.requestAttributes = requestAttributes;
        }
 
        @Override
        public T call() throws Exception {
            try {
                RequestContextHolder.setRequestAttributes(requestAttributes);
                return delegate.call();
            } finally {
                RequestContextHolder.resetRequestAttributes();
            }
        }
    }
}

4、实现 HystrixCallableWrapper接口，定义一个包装Mdc日志处理上下文的实现类：

public class MdcAwareCallableWrapper implements HystrixCallableWrapper {
    @Override
    public <T> Callable<T> wrap(Callable<T> callable) {
        return new MdcAwareCallable<>(callable, MDC.getCopyOfContextMap());
    }
 
    private class MdcAwareCallable<T> implements Callable<T> {
 
        private final Callable<T> delegate;
 
        private final Map<String, String> contextMap;
 
        public MdcAwareCallable(Callable<T> callable, Map<String, String> contextMap) {
            this.delegate = callable;
            this.contextMap = contextMap != null ? contextMap : new HashMap();
        }
 
        @Override
        public T call() throws Exception {
            try {
                MDC.setContextMap(contextMap);
                return delegate.call();
            } finally {
                MDC.clear();
            }
        }
    }
}

5、最后通过在Configuration配置类中注册如下 HystrixCallableWrapper 实现类的bean实例，并通过 HystrixPlugins注册扩展包装实现:

@Bean
public HystrixCallableWrapper requestAttributeAwareCallableWrapper() {
    return new RequestAttributeAwareCallableWrapper();
}
 
@Bean
public HystrixCallableWrapper mdcAwareCallableWrapper(){
    return new MdcAwareCallableWrapper();
}
 
@Autowired(required = false)
private List<HystrixCallableWrapper> wrappers = new ArrayList<>();
 
@PostConstruct
public void init() {
    HystrixPlugins.getInstance().registerConcurrencyStrategy(new RequestContextHystrixConcurrencyStrategy(wrappers));
}

总结

本章从官方网站与源码出发，逐步实现了hystrix中如何进行线程上下文的传播。同时为了更好的扩展性，提供了基于自定义接口并注入实现的方式。