文章目录
- 问题场景
- 原因分析
-
- RequestContextHolder
- Hystrix隔离策略
- 解决方案
-
- 方案1 修改隔离策略为SEMAPHORE
- 方案2 自定义隔离策略
问题场景
在使用Hystrix的情况下，使用RequestInterceptor获取Oauth2 令牌并传递到下游服务时，发现从RequestContextHolder中获取到的HttpServletRequest为NULL。
@Component
public class AAOauth2TokenRequestInterceptor implements RequestInterceptor {
@Override
public void apply(RequestTemplate requestTemplate) {
// 1. 获取 HttpServletRequest
HttpServletRequest request = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();
}
}
原因分析
RequestContextHolder
RequestContextHolder 请求上下文持有者，可以看到在其ThreadLocal中，保存了请求对象，我们可以在当前线程的任意位置，通过这个类来获取到当前请求的RequestAttributes。
Hystrix隔离策略
Hystrix提供了两个隔离策略：THREAD、SEMAPHORE。其默认的策略为THREAD（线程池）。
在执行请求的实际，可以看到其代理对象是通过HystrixInvocationHandler实现的，首先会创建一个HystrixCommand对象。
在其构造方法中，可以看到，会根据客户端名称，创建一个线程池，其最大线程数10。
由以上可知，RequestContextHolder 中的RequestAttributes是保存在ThreadLocal中，而ThreadLocal，在子线程中（包括new Thread和new线程池）是无法获取到的，由此造成了获取到的HttpServletRequest为null。
解决方案
方案1 修改隔离策略为SEMAPHORE
线程池和信号量最大的区别在于:
- 线程池通过每次都开启一个单独线程运行，每个服务单独用线程池，可以是异步，当线程池到达maxSize后，再请求会触发fallback接口进行熔断。调用线程和hystrixCommand线程不是同一个线程。
- 通过信号量的计数器，当信号量达到maxConcurrentRequests后。再请求会触发fallback。调用线程和hystrixCommand线程是同一个线程。
所以在SEMAPHORE策略下，因为是同一个线程，所以可以获取到ThreadLocal中值。
通过以下配置修改隔离策略为SEMAPHORE（信号量）:
hystrix:
command:
default:
execution:
isolation:
strategy: SEMAPHORE
重启项目，发现成功获取到了HttpServletRequest。
但是信号量存在一个很大的问题，信号量的调用是同步的，每次调用都得阻塞调用方的线程，直到结果返回。这样就导致了无法对访问做超时（只能依靠调用协议超时，无法主动释放）。官网也不推荐使用这种模式！
方案2 自定义隔离策略
参考文档
实现HystrixConcurrencyStrategy类，重写其wrapCallable方法，该方可以在执行前包装/修饰{@code Callable<T>}。
@Component
public class RequestAttributeHystrixConcurrencyStrategy extends HystrixConcurrencyStrategy {
private static final Log log = LogFactory
.getLog(RequestAttributeHystrixConcurrencyStrategy.class);
private HystrixConcurrencyStrategy delegate;
public RequestAttributeHystrixConcurrencyStrategy() {
try {
this.delegate = HystrixPlugins.getInstance().getConcurrencyStrategy();
if (this.delegate instanceof RequestAttributeHystrixConcurrencyStrategy) {
// Welcome to singleton hell...
return;
}
HystrixCommandExecutionHook commandExecutionHook = HystrixPlugins
.getInstance().getCommandExecutionHook();
HystrixEventNotifier eventNotifier = HystrixPlugins.getInstance()
.getEventNotifier();
HystrixMetricsPublisher metricsPublisher = HystrixPlugins.getInstance()
.getMetricsPublisher();
HystrixPropertiesStrategy propertiesStrategy = HystrixPlugins.getInstance()
.getPropertiesStrategy();
this.logCurrentStateOfHystrixPlugins(eventNotifier, metricsPublisher,
propertiesStrategy);
HystrixPlugins.reset();
HystrixPlugins.getInstance().registerConcurrencyStrategy(this);
HystrixPlugins.getInstance()
.registerCommandExecutionHook(commandExecutionHook);
HystrixPlugins.getInstance().registerEventNotifier(eventNotifier);
HystrixPlugins.getInstance().registerMetricsPublisher(metricsPublisher);
HystrixPlugins.getInstance().registerPropertiesStrategy(propertiesStrategy);
}
catch (Exception e) {
log.error("Failed to register Sleuth Hystrix Concurrency Strategy", e);
}
}
private void logCurrentStateOfHystrixPlugins(HystrixEventNotifier eventNotifier,
HystrixMetricsPublisher metricsPublisher,
HystrixPropertiesStrategy propertiesStrategy) {
if (log.isDebugEnabled()) {
log.debug("Current Hystrix plugins configuration is ["
+ "concurrencyStrategy [" + this.delegate + "]," + "eventNotifier ["
+ eventNotifier + "]," + "metricPublisher [" + metricsPublisher + "],"
+ "propertiesStrategy [" + propertiesStrategy + "]," + "]");
log.debug("Registering Sleuth Hystrix Concurrency Strategy.");
}
}
@Override
public <T> Callable<T> wrapCallable(Callable<T> callable) {
RequestAttributes requestAttributes = RequestContextHolder.getRequestAttributes();
return new WrappedCallable<>(callable, requestAttributes);
}
@Override
public ThreadPoolExecutor getThreadPool(HystrixThreadPoolKey threadPoolKey,
HystrixProperty<Integer> corePoolSize,
HystrixProperty<Integer> maximumPoolSize,
HystrixProperty<Integer> keepAliveTime, TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
return this.delegate.getThreadPool(threadPoolKey, corePoolSize, maximumPoolSize,
keepAliveTime, unit, workQueue);
}
@Override
public ThreadPoolExecutor getThreadPool(HystrixThreadPoolKey threadPoolKey,
HystrixThreadPoolProperties threadPoolProperties) {
return this.delegate.getThreadPool(threadPoolKey, threadPoolProperties);
}
@Override
public BlockingQueue<Runnable> getBlockingQueue(int maxQueueSize) {
return this.delegate.getBlockingQueue(maxQueueSize);
}
@Override
public <T> HystrixRequestVariable<T> getRequestVariable(
HystrixRequestVariableLifecycle<T> rv) {
return this.delegate.getRequestVariable(rv);
}
static class WrappedCallable<T> implements Callable<T> {
private final Callable<T> target;
private final RequestAttributes requestAttributes;
public WrappedCallable(Callable<T> target, RequestAttributes requestAttributes) {
this.target = target;
this.requestAttributes = requestAttributes;
}
@Override
public T call() throws Exception {
try {
RequestContextHolder.setRequestAttributes(requestAttributes);
return target.call();
}
finally {
RequestContextHolder.resetRequestAttributes();
}
}
}
}
测试，发现能正常获取到HttpServletRequest。
版权声明：本文不是「本站」原创文章，版权归原作者所有 | 原文地址: