Spring Cloud中Hystrix 线程隔离导致ThreadLocal数据丢失（续）

public interface HystrixThreadPool {
    /**
     * Implementation of {@link ThreadPoolExecutor}.
     *
     * @return ThreadPoolExecutor
     */
    public ExecutorService getExecutor();
}

@Override
 public ThreadPoolExecutor getExecutor() {
     touchConfig();
     return threadPool;
 }

 private void touchConfig() {
      final int dynamicCoreSize = properties.coreSize().get();
      final int configuredMaximumSize = properties.maximumSize().get();
      int dynamicMaximumSize = properties.actualMaximumSize();
      final boolean allowSizesToDiverge = properties.getAllowMaximumSizeToDivergeFromCoreSize().get();
      boolean maxTooLow = false;
      if (allowSizesToDiverge && configuredMaximumSize < dynamicCoreSize) {
          //if user sets maximum < core (or defaults get us there), we need to maintain invariant of core <= maximum
          dynamicMaximumSize = dynamicCoreSize;
          maxTooLow = true;
      }
      //......
}

public static ExecutorService getTtlExecutorService(ExecutorService executorService) {
        if (executorService == null || executorService instanceof ExecutorServiceTtlWrapper) {
            return executorService;
        }
        return new ExecutorServiceTtlWrapper(executorService);
}

class ExecutorServiceTtlWrapper extends ExecutorTtlWrapper implements ExecutorService {
    private final ExecutorService executorService;
    ExecutorServiceTtlWrapper(ExecutorService executorService) {
        super(executorService);
        this.executorService = executorService;
    }
    @Override
    public <T> Future<T> submit(Callable<T> task) {
        return executorService.submit(TtlCallable.get(task));
    }
    @Override
    public <T> Future<T> submit(Runnable task, T result) {
        return executorService.submit(TtlRunnable.get(task), result);
    }
    @Override
    public Future<?> submit(Runnable task) {
        return executorService.submit(TtlRunnable.get(task));
    }
    // ...........
}

 private static class ThreadPoolWorker extends Worker {
        private final HystrixThreadPool threadPool;
        private final CompositeSubscription subscription = new CompositeSubscription();
        private final Func0<Boolean> shouldInterruptThread;
        public ThreadPoolWorker(HystrixThreadPool threadPool, Func0<Boolean> shouldInterruptThread) {
            this.threadPool = threadPool;
            this.shouldInterruptThread = shouldInterruptThread;
        }
        @Override
        public void unsubscribe() {
            subscription.unsubscribe();
        }
        @Override
        public boolean isUnsubscribed() {
            return subscription.isUnsubscribed();
        }
        @Override
        public Subscription schedule(final Action0 action) {
            if (subscription.isUnsubscribed()) {
                // don't schedule, we are unsubscribed
                return Subscriptions.unsubscribed();
            }
            // This is internal RxJava API but it is too useful.
            ScheduledAction sa = new ScheduledAction(action);
            subscription.add(sa);
            sa.addParent(subscription);
            ThreadPoolExecutor executor = (ThreadPoolExecutor) threadPool.getExecutor();
            FutureTask<?> f = (FutureTask<?>) executor.submit(sa);
            sa.add(new FutureCompleterWithConfigurableInterrupt(f, shouldInterruptThread, executor));
            return sa;
        }
        @Override
        public Subscription schedule(Action0 action, long delayTime, TimeUnit unit) {
            throw new IllegalStateException("Hystrix does not support delayed scheduling");
        }
}

 public Subscription schedule(final Action0 action) {
     if (subscription.isUnsubscribed()) {
            // don't schedule, we are unsubscribed
            return Subscriptions.unsubscribed();
     }
     // This is internal RxJava API but it is too useful.
     ScheduledAction sa = new ScheduledAction(action);
     subscription.add(sa);
     sa.addParent(subscription);
     ThreadPoolExecutor executor = (ThreadPoolExecutor) threadPool.getExecutor();
     FutureTask<?> f = (FutureTask<?>) executor.submit(TtlRunnable.get(sa));
     sa.add(new FutureCompleterWithConfigurableInterrupt(f, shouldInterruptThread, executor));
     return sa;
}

public class RibbonFilterContextHolder {
    private static final ThreadLocal<RibbonFilterContext> contextHolder = new InheritableThreadLocal<RibbonFilterContext>() {
        @Override
        protected RibbonFilterContext initialValue() {
            return new DefaultRibbonFilterContext();
        }
    };
    public static RibbonFilterContext getCurrentContext() {
        return contextHolder.get();
    }
    public static void clearCurrentContext() {
        contextHolder.remove();
    }
}

private static AtomicInteger ac = new AtomicInteger();
   @Override
   public Object run() {
       RequestContext ctx = RequestContext.getCurrentContext();
       RibbonFilterContextHolder.getCurrentContext().add("servers",ac.addAndGet(1)+"");
       return null;
   }

public class GrayPushRule extends AbstractLoadBalancerRule {
    private AtomicInteger nextServerCyclicCounter;
    private static final boolean AVAILABLE_ONLY_SERVERS = true;
    private static final boolean ALL_SERVERS = false;
    private static Logger log = LoggerFactory.getLogger(RoundRobinRule.class);
    public GrayPushRule() {
        this.nextServerCyclicCounter = new AtomicInteger(0);
    }
    public GrayPushRule(ILoadBalancer lb) {
        this();
        this.setLoadBalancer(lb);
    }
    public Server choose(ILoadBalancer lb, Object key) {
        String servers = RibbonFilterContextHolder.getCurrentContext().get("servers");
        System.out.println(Thread.currentThread().getName()+":"+servers);  
        return null;
    }
    public Server choose(Object key) {
        return this.choose(this.getLoadBalancer(), key);
    }
    public void initWithNiwsConfig(IClientConfig clientConfig) {
    }
}

private static final TransmittableThreadLocal<RibbonFilterContext> contextHolder = new TransmittableThreadLocal<RibbonFilterContext>() {
    @Override
    protected RibbonFilterContext initialValue() {
        return new DefaultRibbonFilterContext();
    }
};

FutureTask<?> f = (FutureTask<?>) executor.submit(TtlRunnable.get(sa));

HystrixPlugins.getInstance().registerConcurrencyStrategy(new ThreadLocalHystrixConcurrencyStrategy());

public class ThreadLocalHystrixConcurrencyStrategy extends HystrixConcurrencyStrategy {
    private final static Logger logger = LoggerFactory.getLogger(ThreadLocalHystrixConcurrencyStrategy.class);
    @Override
    public ThreadPoolExecutor getThreadPool(HystrixThreadPoolKey threadPoolKey, HystrixProperty<Integer> corePoolSize,
            HystrixProperty<Integer> maximumPoolSize, HystrixProperty<Integer> keepAliveTime, TimeUnit unit,
            BlockingQueue<Runnable> workQueue) {
        return this.doGetThreadPool(threadPoolKey, corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }
    @Override
    public ThreadPoolExecutor getThreadPool(HystrixThreadPoolKey threadPoolKey,
            HystrixThreadPoolProperties threadPoolProperties) {
        return this.doGetThreadPool(threadPoolKey, threadPoolProperties);
    }
}

return new ThreadLocalThreadPoolExecutor(dynamicCoreSize, dynamicMaximumSize, keepAliveTime.get(), unit, workQueue,
                    threadFactory);

public class ThreadLocalThreadPoolExecutor extends ThreadPoolExecutor {
    private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();
    public ThreadLocalThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
            BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }
    public ThreadLocalThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
            BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, defaultHandler);
    }
    @Override
    public void execute(Runnable command) {
        super.execute(TtlRunnable.get(command));
    }
}