全链路压测架构设计

/**
 * 链路跟踪Request设置值
 */
public class MvcWormholeWebInterceptor implements WebRequestInterceptor {


    @Override
    public void preHandle(WebRequest webRequest) {
        //失效上下文，解决Tomcat线程复用问题
        WormholeContextHolder.invalidContext();
        String wormholeValue = webRequest.getHeader(WormholeContextHolder.WORMHOLE_REQUEST_MARK);
        if (StringUtils.isNotEmpty(wormholeValue)) {
            WormholeContextHolder.setContext(new WormholeContext(wormholeValue));
        }
    }

    @Override
    public void postHandle(WebRequest webRequest, ModelMap modelMap) throws Exception {

    }

    @Override
    public void afterCompletion(WebRequest webRequest, Exception e) throws Exception {

    }
}

public class WormholeFeignRequestInterceptor implements RequestInterceptor {

    @Override
    public void apply(RequestTemplate requestTemplate) {
        WormholeContext wormholeContext = WormholeContextHolder.getContext();
        if (null != wormholeContext) {
            requestTemplate.header(WormholeContextHolder.WORMHOLE_REQUEST_MARK, wormholeContext.toString());
        }
    }
}

public final class DelegatingWormholeContextCallable<V> implements Callable<V> {
    private final Callable<V> delegate;
    // 用户信息上下文（根据项目实际情况定义ThreadLocal上下文）
    private WormholeContext orginWormholeContext;

    public DelegatingWormholeContextCallable(Callable<V> delegate,
                                             WormholeContext wormholeContext) {
        this.delegate = delegate;
        this.orginWormholeContext = wormholeContext;
    }

    public V call() throws Exception {
        //防止线程复用销毁ThreadLocal的数据
        WormholeContextHolder.invalidContext();
        // 将当前的用户上下文设置进Hystrix线程的TreadLocal中
        WormholeContextHolder.setContext(orginWormholeContext);
        try {
            return delegate.call();
        } finally {
            // 执行完毕，记得清理ThreadLocal资源
            WormholeContextHolder.invalidContext();
        }
    }

    public static <V> Callable<V> create(Callable<V> delegate,
                                         WormholeContext wormholeContext) {
        return new DelegatingWormholeContextCallable<V>(delegate, wormholeContext);
    }
}

public class ThreadLocalAwareStrategy extends HystrixConcurrencyStrategy {

    // 最简单的方式就是引入现有的并发策略，进行功能扩展
    private final HystrixConcurrencyStrategy existingConcurrencyStrategy;

    public ThreadLocalAwareStrategy(
            HystrixConcurrencyStrategy existingConcurrencyStrategy) {
        this.existingConcurrencyStrategy = existingConcurrencyStrategy;
    }

    @Override
    public BlockingQueue<Runnable> getBlockingQueue(int maxQueueSize) {
        return existingConcurrencyStrategy != null
                ? existingConcurrencyStrategy.getBlockingQueue(maxQueueSize)
                : super.getBlockingQueue(maxQueueSize);
    }

    @Override
    public <T> HystrixRequestVariable<T> getRequestVariable(
            HystrixRequestVariableLifecycle<T> rv) {
        return existingConcurrencyStrategy != null
                ? existingConcurrencyStrategy.getRequestVariable(rv)
                : super.getRequestVariable(rv);
    }

    @Override
    public ThreadPoolExecutor getThreadPool(HystrixThreadPoolKey threadPoolKey,
                                            HystrixProperty<Integer> corePoolSize,
                                            HystrixProperty<Integer> maximumPoolSize,
                                            HystrixProperty<Integer> keepAliveTime, TimeUnit unit,
                                            BlockingQueue<Runnable> workQueue) {
        return existingConcurrencyStrategy != null
                ? existingConcurrencyStrategy.getThreadPool(threadPoolKey, corePoolSize,
                maximumPoolSize, keepAliveTime, unit, workQueue)
                : super.getThreadPool(threadPoolKey, corePoolSize, maximumPoolSize,
                keepAliveTime, unit, workQueue);
    }

    @Override
    public <T> Callable<T> wrapCallable(Callable<T> callable) {
        return existingConcurrencyStrategy != null
                ? existingConcurrencyStrategy
                .wrapCallable(new DelegatingWormholeContextCallable<>(callable, WormholeContextHolder.getContext()))
                : super.wrapCallable(new DelegatingWormholeContextCallable<T>(callable, WormholeContextHolder.getContext()));
    }
}

public class HystrixThreadLocalConfiguration {

    @Autowired(required = false)
    private HystrixConcurrencyStrategy existingConcurrencyStrategy;

    @PostConstruct
    public void init() {
        // Keeps references of existing Hystrix plugins.
        HystrixEventNotifier eventNotifier = HystrixPlugins.getInstance()
                .getEventNotifier();
        HystrixMetricsPublisher metricsPublisher = HystrixPlugins.getInstance()
                .getMetricsPublisher();
        HystrixPropertiesStrategy propertiesStrategy = HystrixPlugins.getInstance()
                .getPropertiesStrategy();
        HystrixCommandExecutionHook commandExecutionHook = HystrixPlugins.getInstance()
                .getCommandExecutionHook();

        HystrixPlugins.reset();

        HystrixPlugins.getInstance().registerConcurrencyStrategy(new ThreadLocalAwareStrategy(existingConcurrencyStrategy));
        HystrixPlugins.getInstance().registerEventNotifier(eventNotifier);
        HystrixPlugins.getInstance().registerMetricsPublisher(metricsPublisher);
        HystrixPlugins.getInstance().registerPropertiesStrategy(propertiesStrategy);
        HystrixPlugins.getInstance().registerCommandExecutionHook(commandExecutionHook);
    }
}

/**
 * 动态数据源上下文
 */
public class DynamicDataSourceContextHolder {
    public static final String PRIMARY_DB = "primary";
    public static final String SHADOW_DB = "shadow";

    private static final ThreadLocal<String> contextHolder = new ThreadLocal<String>() {
        /**
         * 将 master 数据源的 key作为默认数据源的 key
         */
        @Override
        protected String initialValue() {
            return PRIMARY_DB;
        }
    };


    /**
     * 数据源的 key集合，用于切换时判断数据源是否存在
     */
    public static List<Object> dataSourceKeys = new ArrayList<>();

    /**
     * 切换数据源
     *
     * @param key
     */
    public static void setDataSourceKey(String key) {
        contextHolder.set(key);
    }

    /**
     * 获取数据源
     *
     * @return
     */
    public static String getDataSourceKey() {
        return contextHolder.get();
    }

    /**
     * 重置数据源
     */
    public static void clearDataSourceKey() {
        contextHolder.remove();
    }

    /**
     * 判断是否包含数据源
     *
     * @param key 数据源key
     * @return
     */
    public static boolean containDataSourceKey(String key) {
        return dataSourceKeys.contains(key);
    }

    /**
     * 添加数据源keys
     *
     * @param keys
     * @return
     */
    public static boolean addDataSourceKeys(Collection<? extends Object> keys) {
        return dataSourceKeys.addAll(keys);
    }
}

/**
 * 动态数据源实现类
 */
public class DynamicDataSource extends AbstractRoutingDataSource {

    /**
     * 如果不希望数据源在启动配置时就加载好，可以定制这个方法，从任何你希望的地方读取并返回数据源
     * 比如从数据库、文件、外部接口等读取数据源信息，并最终返回一个DataSource实现类对象即可
     */
    @Override
    protected DataSource determineTargetDataSource() {
        //获取当前的上下文
        WormholeContext wormholeContext = WormholeContextHolder.getContext();
        //如果不为空使用影子库
        if (null != wormholeContext) {
            DynamicDataSourceContextHolder.setDataSourceKey(DynamicDataSourceContextHolder.SHADOW_DB);
        } else {
            //为空则使用主数据源
            DynamicDataSourceContextHolder.setDataSourceKey(DynamicDataSourceContextHolder.PRIMARY_DB);
        }
        return super.determineTargetDataSource();
    }

    /**
     * 如果希望所有数据源在启动配置时就加载好，这里通过设置数据源Key值来切换数据，定制这个方法
     */
    @Override
    protected Object determineCurrentLookupKey() {
        return DynamicDataSourceContextHolder.getDataSourceKey();
    }

 
}

public class KeyStringRedisSerializer extends StringRedisSerializer {

    @Resource
    private WormholeIsolationConfiguration isolationConfiguration;

    public byte[] serialize(@Nullable String redisKey) {
        WormholeContext wormholeContext = WormholeContextHolder.getContext();
        if (null != wormholeContext) {
            redisKey = isolationConfiguration.generateIsolationKey(redisKey);
        }
        return super.serialize(redisKey);
    }
}

/**
 * Redis 配置类
 */
@Configuration
@ConditionalOnClass({RedisTemplate.class, RedisOperations.class, RedisConnectionFactory.class})
public class WormholeRedisAutoConfiguration {


    @Bean
    public KeyStringRedisSerializer keyStringRedisSerializer() {
        return new KeyStringRedisSerializer();
    }

    @Bean("redisTemplate")
    public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory factory) {
        RedisTemplate template = new RedisTemplate();
        //使用fastjson序列化
        FastJsonRedisSerializer fastJsonRedisSerializer = new FastJsonRedisSerializer(Object.class);
        // value值的序列化采用fastJsonRedisSerializer
        template.setValueSerializer(fastJsonRedisSerializer);
        template.setHashValueSerializer(fastJsonRedisSerializer);
        // key的序列化采用StringRedisSerializer
        template.setKeySerializer(keyStringRedisSerializer());
        template.setHashKeySerializer(keyStringRedisSerializer());
        template.setConnectionFactory(factory);
        return template;
    }

    @Bean
    public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory factory) throws UnknownHostException {
        StringRedisTemplate template = new StringRedisTemplate();
        template.setKeySerializer(keyStringRedisSerializer());
        template.setHashKeySerializer(keyStringRedisSerializer());
        template.setConnectionFactory(factory);
        return template;
    }
}

public class WormholeRabbitListenerAnnotationBeanPostProcessor extends RabbitListenerAnnotationBeanPostProcessor {

    @Resource
    private WormholeIsolationConfiguration wormholeIsolationConfiguration;

    /**
     * routingKey 前置处理器
     *
     * @param queueName
     * @param routingKey
     * @return
     */
    @Override
    public String preProcessingRoutingKey(String queueName, String routingKey) {
        //如果是影子队列就将routingKey转换为 影子routingKey
        if (wormholeIsolationConfiguration.checkIsolation(queueName) && !wormholeIsolationConfiguration.checkIsolation(routingKey)) {
            return wormholeIsolationConfiguration.generateIsolationKey(routingKey);
        }
        return routingKey;
    }

    /**
     * 处理队列问题，如果来了一个队列就生成一个shadow的队列
     *
     * @param queues
     * @return
     */
    @Override
    public List<String> handelQueues(List<String> queues) {
        List<String> isolationQueues = new ArrayList<>();
        if (null != queues && !queues.isEmpty()) {
            for (String queue : queues) {
                //添加shadow队列
                isolationQueues.add(wormholeIsolationConfiguration.generateIsolationKey(queue));
            }
            queues.addAll(isolationQueues);
        }
        return queues;
    }
}

public class WormholeSimpleRabbitListenerContainerFactory extends SimpleRabbitListenerContainerFactory {

    @Override
    protected SimpleMessageListenerContainer createContainerInstance() {
        SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer();
        simpleMessageListenerContainer.setAfterReceivePostProcessors(message -> {
            //防止线程复用 销毁ThreadLocal
            WormholeContextHolder.invalidContext();
            //获取消息属性标识
            String wormholeRequestContext = message.getMessageProperties().getHeader(WormholeContextHolder.WORMHOLE_REQUEST_MARK);
            if (StringUtils.isNotEmpty(wormholeRequestContext)) {
                WormholeContextHolder.setContext(wormholeRequestContext);
            }
            return message;
        });
        return simpleMessageListenerContainer;
    }
}

public class ShadowRabbitTemplate extends RabbitTemplate {
    public ShadowRabbitTemplate(ConnectionFactory connectionFactory) {
        super(connectionFactory);
    }

    @Autowired
    private WormholeIsolationConfiguration isolationConfiguration;

    @Override
    public void send(final String exchange, final String routingKey,
                     final Message message, @Nullable final CorrelationData correlationData)
            throws AmqpException {
        WormholeContext wormholeContext = WormholeContextHolder.getContext();
        if (null == wormholeContext) {
            super.send(exchange, routingKey, message, correlationData);
        } else {
            message.getMessageProperties().setHeader(WormholeContextHolder.WORMHOLE_REQUEST_MARK, wormholeContext.toString());
            //生成Rabbit 隔离Key
            String wormholeRoutingKey = isolationConfiguration.generateIsolationKey(routingKey);
            //调用父类进行发送
            super.send(exchange, wormholeRoutingKey, message, correlationData);
        }
    }
}

@Aspect
public class WormholeMockSection {

    /**
     * 切点 拦截@WormholeMock的注解
     */
    @Pointcut("@annotation(com.baiyp.wormhole.component.mock.annotation.WormholeMock)")
    public void pointCut() {
    }

    /**
     * 环绕通知
     *
     * @param point
     * @return
     * @throws Throwable
     */
    @Around("pointCut()")
    public Object section(ProceedingJoinPoint point) throws Throwable {
        WormholeContext wormholeContext = WormholeContextHolder.getContext();
        Object[] parameter = point.getArgs();
        //如果没有wormholeContext 就执行正常方法
        if (null == wormholeContext) {
            return point.proceed(parameter);
        }
        //如果存在就执行MOCK方法
        WormholeMock wormholeMock = WormholeMockUtils.getMethodAnnotation(point, WormholeMock.class);
        if (null != wormholeMock) {
            //获取到 Mock 回调类
            WormholeMockCallback wormholeMockCallback = WormholeMockUtils.getWormholeMockCallback(wormholeMock);
            if (null != wormholeMockCallback) {
                return wormholeMockCallback.handelMockData(parameter);
            }
        }
        return null;
    }

}

@Override
//加入注解进行MOCK测试拦截 设置最大耗时
@WormholeMock(maxDelayTime = 10, minDelayTime = 2)
public boolean send(NotifyVO notifyVO) {
    logger.info("开始发送短信通知.....");
    try {
        //模拟发送短信耗时
        Thread.sleep(5);
    } catch (InterruptedException e) {
    }
    return true;
}

/**
 * MVC 自动装配
 */
@Configuration
//当DispatcherServlet存在时该配置类才会被执行到
@ConditionalOnClass(org.springframework.web.servlet.DispatcherServlet.class)
public class WormholeMVCAutoConfiguration {

    @ConditionalOnClass
    @Bean
    public WormholeMVCConfiguration wormholeMVCConfiguration() {
        return new WormholeMVCConfiguration();
    }
}