Hystrix Feign 父子线程数据传递、Feign Header的添加

左手的ㄟ右手 2022-12-08 05:37 217阅读 0赞

**问题场景**

*  微服务A通过feign调用微服务B
 *  使用了Hystrix并开启了线程隔离模式（默认模式），所以A调用B的请求会单独起一个子线程的方式去调用
 *  现在需要将微服务A中ThreadLocal里的数据，放入feign请求B时的http header中（这里的http请求会在子线程中）

**要解决如上问题，需要做两件事情**

*  找到可以给feign调用添加header的切入点，在这里获取主线程ThreadLocal中的数据并添加到header中
 *  找到Hystrix开启新的子线程的切入点，在线程执行run方法的先后分别做数据传递和数据清理工作

**实现**

细节本人就不做过多赘述，下面先给UML类图，描述出类之间的关系，然后给出具体的实现类源码，拷贝到项目中即可食用。

上菜谱：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NhdG9vcA_size_16_color_FFFFFF_t_70_pic_center]  
**解释一下上面类的清单**（有底色的两个类/接口是jar包中的）：

*  RequestInterceptor 来自jar包，请求拦截器接口
 *  FeignClientRequestInterceptor 扩展请求拦截器接口，添加header
 *  HystrixConfiguration 自实现配置类
 *  HystrixConcurrencyStrategy 来自jar包，用于使用默认实现为系统的并发相关方面定义不同的行为或实现。
 *  MusesHystrixConcurrencyStrategy 扩展 HystrixConcurrencyStrategy 并重写 wrapCallable 方法
 *  HystrixCallableWrapper 自定义接口用于包装 HystrixCallable 的 call 方法，以达到可以无限扩展
 *  PDCAwareCallableWrapper 实现接口 HystrixCallableWrapper，添加PDC数据
 *  PDCAwareCallable 内部类，具体对 call 的包装
 *  RequestAttributeAwareCallableWrapper 实现接口 HystrixCallableWrapper，用于拷贝 RequestAttribute 上限文数据
 *  RequestAttributeAwareCallable 内部类，具体对 call 的包装

> 注：  
> 1.文中的PDC就是我项目中用户操作ThreadLocal的工具类，带有 get 和 put 方法，就这样。  
> 2.HystrixCallableWrapper 抽象定义方法，以后直接实现该接口即可完成无限扩展。

**下面是文件源码**

/** * 与 Hystrix 相关的公共配置 * * @author shanhy * @date 2020/9/17 13:48 */
    @Slf4j
    @Configuration
    public class HystrixConfiguration { 
    
        @Autowired(required = false)
        private List<HystrixCallableWrapper> wrappers;
    
        @Bean
        public HystrixCallableWrapper requestAttributeAwareCallableWrapper() { 
            return new RequestAttributeAwareCallableWrapper();
        }
    
        @Bean
        public HystrixCallableWrapper pDCAwareCallableWrapper() { 
            return new PDCAwareCallableWrapper();
        }
    
        @Bean
        public HystrixConcurrencyStrategy requestContextHystrixConcurrencyStrategy() { 
            return new MusesHystrixConcurrencyStrategy(wrappers);
        }
    
        @PostConstruct
        public void init() { 
            if (!Collections.isEmpty(wrappers)) { 
                try { 
                    HystrixConcurrencyStrategy strategy = HystrixPlugins.getInstance().getConcurrencyStrategy();
                    if (strategy instanceof MusesHystrixConcurrencyStrategy) { 
                        return;
                    }
                    HystrixConcurrencyStrategy hystrixConcurrencyStrategy = new MusesHystrixConcurrencyStrategy(wrappers);
    
                    // 获取原来的相关数据配置
                    HystrixCommandExecutionHook commandExecutionHook = HystrixPlugins
                            .getInstance().getCommandExecutionHook();
                    HystrixEventNotifier eventNotifier = HystrixPlugins.getInstance()
                            .getEventNotifier();
                    HystrixMetricsPublisher metricsPublisher = HystrixPlugins.getInstance()
                            .getMetricsPublisher();
                    HystrixPropertiesStrategy propertiesStrategy = HystrixPlugins.getInstance()
                            .getPropertiesStrategy();
    
                    // 打印日志
                    if (log.isDebugEnabled()) { 
                        log.debug("Current Hystrix plugins configuration is [concurrencyStrategy [{}], eventNotifier [{}], metricPublisher [{}], propertiesStrategy [{}]]",
                                hystrixConcurrencyStrategy, eventNotifier, metricsPublisher, propertiesStrategy);
                        log.debug("Registering Muses Hystrix Concurrency Strategy.");
                    }
    
                    // 重置再重新填充
                    // 直接设置会触发异常 Caused by: java.lang.IllegalStateException: Another strategy was already registered.
                    HystrixPlugins.reset();
                    HystrixPlugins.getInstance().registerConcurrencyStrategy(hystrixConcurrencyStrategy);
                    HystrixPlugins.getInstance()
                            .registerCommandExecutionHook(commandExecutionHook);
                    HystrixPlugins.getInstance().registerEventNotifier(eventNotifier);
                    HystrixPlugins.getInstance().registerMetricsPublisher(metricsPublisher);
                    HystrixPlugins.getInstance().registerPropertiesStrategy(propertiesStrategy);
                } catch (Exception e) { 
                    log.error("Failed to register Muses Hystrix Concurrency Strategy", e);
                }
            }
        }
    }

/** * 用于包装hystrix中Callable实例的接口 * * @author shanhy * @date 2020/9/17 13:42 */
    public interface HystrixCallableWrapper { 
    
        /** * 包装Callable实例 * * @param callable 待包装实例 * @param <T> 返回类型 * @return 包装后的实例 */
        <T> Callable<T> wrap(Callable<T> callable);
    
    }

/** * 实现HystrixCallableWrapper接口，实现PDC数据在父子线程之间的传递 * * @author shanhy * @date 2020/9/17 13:44 */
    public final class PDCAwareCallableWrapper implements HystrixCallableWrapper { 
    
        @Override
        public <T> Callable<T> wrap(Callable<T> callable) { 
            return new PDCAwareCallable<>(callable, PDC.getCopyOfContextMap());
        }
    
        static class PDCAwareCallable<T> implements Callable<T> { 
    
            private final Callable<T> delegate;
            private final Map<String, Object> contextMap;
    
            PDCAwareCallable(Callable<T> callable, Map<String, Object> contextMap) { 
                this.delegate = callable;
                this.contextMap = contextMap;
            }
    
            @Override
            public T call() throws Exception { 
                try { 
                    if(Objects.nonNull(contextMap))
                        PDC.setContextMap(contextMap);
                    return delegate.call();
                } finally { 
                    PDC.clear();
                }
            }
        }
    }

/** * 实现HystrixCallableWrapper接口，定义一个包装RequestContextHolder上下文处理的实现类 * 当有多个HystrixCallableWrapper实现类的时候，方法getOrder决定执行顺序 * * @author shanhy * @date 2020/9/17 13:44 */
    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();
                }
            }
        }
    }

/** * 类似 SecurityContextConcurrencyStrategy，插件式扩展实现：被执行目标方法执行前后的处理包装 * * @author shanhy * @date 2020/9/17 13:42 */
    public class MusesHystrixConcurrencyStrategy extends HystrixConcurrencyStrategy { 
    
        private final Collection<HystrixCallableWrapper> wrappers;
    
        public MusesHystrixConcurrencyStrategy(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;
            }
        }
    }

/** * feign拦截器 处理userID orgId * * @author shanhy * @CreateDate 2020/8/18 19:36 */
    @Slf4j
    @Component
    @ConditionalOnClass(RequestInterceptor.class)
    @SuppressWarnings("unused")
    public class FeignClientRequestInterceptor implements RequestInterceptor { 
    
        @Override
        public void apply(RequestTemplate requestTemplate) { 
            requestTemplate.header(MusesDataContext.CONTEXT_CURRENT_ID_USER, PDC.get(MusesDataContext.CONTEXT_CURRENT_ID_USER));
            requestTemplate.header(MusesDataContext.CONTEXT_CURRENT_ID_ORG, PDC.get(MusesDataContext.CONTEXT_CURRENT_ID_ORG));
        }
    }

--------------------

（END）

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NhdG9vcA_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/868c59b23c224cae896c46a2b88b100d.png