大促之前全链路压测原理解析

1. 全链路压测的意义

'xxs'

上图为 2012 年淘宝核心业务应用关系的拓扑图，还不包含了其他的非核心业务应用，所谓的核心业务就是和交易相关的，和钱相关的业务，这张图大家可能看不清楚，看不清楚才是正常的，因为当时的阿里应用数量之多、应用间关系之混乱靠人工确实已经无法理清楚了。

在真实的业务场景种，每个系统的压力都比较大，而系统之间是有相互依赖关系的，单机压测没有考虑到依赖环节压力都比较大的情况，可能会出现一个不确定的误差

2. 链路压测方案刨析

2.1 线下压测

顾名思义就是在测试环境进行压测，且是针对一些重点项目这种测试手段，因为测试环境硬件资源以及压测数据与线上差别太大并且服务间依赖关系错综复杂，测试环境很难模拟且不够稳定，压测出来的数据指标参考价值不大，难以用测试环境得出的结果推导生产真实容量。

2.2 预生产环境压测

'xxs'

这个一般是将生成环境的硬件以及软件同步复制到与生产环境一份，然后对服务内部的外部调用接口进行拦截，然后进行压测这样可以评估出来生产环境的真实容量以及达到压测的目的，但是成本非常高，需要将生产环境的硬件完全的复制一份，并未维护成本非常高，部署的时候需要同步的在预生产环境进行部署，以及压测代码的更改。

2.3 引流压测

随着业务量的不断增长，考虑到线下测试结果的准确性，开始尝试生产压测，这种压测手段，我们称之为引流压测，事实上没有真正的模拟放大压力进行测试，而是一种通过缩小在线服务集群数的方式来放大单机处理量。比如一个业务系统的集群有100个节点，将其中90个节点模拟下线或转发流量到剩余的10个节点上实施压测。

'xxs'

引流压测的弊端在于，DB承受压力不变，上下游系统的压力不变。压测结果仅能代表单个应用的性能，但往往无法识别链路和架构级的隐患，而且在引流过程中倘若出现异常或突如其来的业务高峰，很容易造成生产故障。

2.4 全链路压测

随着微服务架构的流行，服务按照不同的维度进行拆分，一次请求往往需要涉及到多个服务

互联网应用构建在不同的软件模块集上，这些软件模块，有可能是由不同的团队开发、可能使用不同的编程语言来实现、有可能布在了几千台服务器，横跨多个不同的数据中心。因此，就需要一些可以帮助理解系统行为、用于分析性能问题的工具，以便发生故障的时候，能够快速定位和解决问题，但是他的缺点也很明显就是需要的技术难度很高，需要克服流量染色，数据隔离，日志隔离，风险熔断等技术难题，因位在生产环境压测，所以控制不好风险是非常高的。

所以，在复杂的微服务架构系统中，几乎每一个前端请求都会形成一个复杂的分布式服务调用链路，一个请求完整调用链可能如下图所示：

'xxs'

2.5 四种压测方案对比

	压测效果	技术难度	机器成本	维护成本	风险
线下压测	差	低	低	低	无
预生产压测	好	低	高	高	中
引流压测	差	中	无	低	高
全链路压测	好	高	无	低	高

3. 全链路压测概述

3.1 概念

基于实际的生产业务场景、生产环境，模拟海量的用户请求和数据对整个业务链（通常是核心业务链）进行压力测试，并持续调优的过程。

3.2 解决的问题

解决在业务场景越发复杂化、海量数据冲击下系统整个业务链的可用性、服务能力的瓶颈，以及容量规划等问题。

3.2.1 精确的容量规划

3.2.1.1 为什么需要容量规划

什么时候增减机器、保障系统稳定性、节约成本

容量规划的目的在于让每一个业务系统能够清晰地知道：什么时候该加机器、什么时候应该减机器？双11等大促场景需要准备多少机器，既能保障系统稳定性、又能节约成本

3.2.1.2 容量规划四步走

业务流量预估阶段：通过历史数据分析未来某一个时间点业务的访问量会有多大
系统容量评估阶段：初步计算每一个系统需要分配多少机器
容量的精调阶段：通过全链路压测来模拟大促时刻的用户行为，在验证站点能力的同时对整个站点的容量水位进行精细调整
流量控制阶段：对系统配置限流阈值等系统保护措施，防止实际的业务流量超过预估业务流量的情况下，系统无法提供正常服务流量控制阶段：对系统配置限流阈值等系统保护措施，防止实际的业务流量超过预估业务流量的情况下，系统无法提供正常服务

3.2.2 全链路的性能监控

全链路性能监控 从整体维度到局部维度展示各项指标，将跨应用的所有调用链性能信息集中展现，可方便度量整体和局部性能，并且方便找到故障产生的源头，生产上可极大缩短故障排除时间。

保证系统稳定性：可能提前预估系统存在的各种问题，提前模拟高并发场景，有备无患。
请求链路追踪，故障快速定位：可以通过调用链结合业务日志快速定位错误信息。
精准的容量评估：能够定位到最需要扩容的服务，帮助公司用最低的成本满足业务的性能要求
真实的性能验证：能够在生成环境以最真实的环境来验证系统的真实性能。
数据分析，优化链路：可以得到用户的行为路径，汇总分析应用在很多业务场景。

'xxs'

3.3 如何展开全链路压测

3.3.1 业务模型梳理

首先应该将核心业务和非核心业务进行拆分，确认流量高峰针对的是哪些业务场景和模块，针对性的进行扩容准备。
梳理出对外的接口：使用MOCK（模拟）方式做挡板。
不要污染正常数据：梳理数据处理的每一个环节，确保 mock 数据的处理结果不会写入到正常库里面

3.3.2 数据模型构建

数据的真实性和可用性：可以从生产环境完全移植一份当量的数据包，作为压测的基础数据，然后基于基础数据，通过分析历史数据增长趋势，预估当前可能的数据量
数据隔离：不要污染正常数据，认真梳理数据处理的每一个环节，可以考虑通过压测数据隔离处理，落入影子库，mock 对象等手段，来防止数据污染

3.3.3 压测工具选型

使用分布式压测的手段来进行用户请求模拟，目前有很多的开源工具可以提供分布式压测的方式，比如JMeter、nGrinder、Locust等。

4. 业务模块介绍

'xxs'

5. 全链路整体架构

整体架构如下主要是对压测客户端的压测数据染色，全链路中间件识别出染色数据，并将正常数据和压测数据区分开，进行数据隔离，这里主要涉及到mysql数据库，RabbitMQ，Redis，还需要处理因为hystrix线程池不能通过ThreadLocal传递染色表示的问题。

'xxs'

5.1 需要应对的问题

5.1.1 业务问题

涉及的系统太多，牵扯的开发人员太多

在压测过程中，做一个全链路的压测一般会涉及到大量的系统，在整个压测过程中，光各个产品的人员协调就是一个比较大的工程，牵扯到太多的产品经理和开发人员，如果公司对全链路压测早期没有足够的重视，那么这个压测工作难开展。
模拟的测试数据和访问流量不真实

在压测过程中经常会遇到压测后得到的数据不准确的问题，这就使得压测出的数据参考性不强，主要就是因为压测的环境可能和生成环境存在误差、参数存在不一样的地方、测试数据存在不一样的地方这些因素综合起来导致测试结果的不可信。
压测生产数据未隔离，影响生产环境

在全链路压测过程中，压测数据可能会影响到生产环境的真实数据，举个🌰，电商系统在生产环境进行全链路压测的时候可能会有很多压测模拟用户去下单，如果不做处理，直接下单的话会导致系统一下子会产生很多废订单，从而影响到库存和生产订单数据，影响到日常的正常运营。

5.1.2 技术问题

5.1.2.1 探针的性能消耗

APM组件服务的影响应该做到足够小

服务调用埋点本身会带来性能损耗，这就需要调用跟踪的低损耗，实际中还会通过配置采样率的方式，选择一部分请求去分析请求路径。在一些高度优化过的服务，即使一点点损耗也会很容易察觉到，而且有可能迫使在线服务的部署团队不得不将跟踪系统关停。

采样率: 如1000个请求, 取1次

5.1.2.2 代码的侵入性

即也作为业务组件，应当尽可能少入侵或者无入侵其他业务系统，对于使用方透明，减少开发人员的负担。

5.1.2.3 可扩展性

一个优秀的调用跟踪系统必须支持分布式部署，具备良好的可扩展性，能够支持的组件越多当然越好，或者提供便捷的插件开发API，对于一些没有监控到的组件，应用开发者也可以自行扩展。

5.1.2.4 数据的分析

数据的分析要快，分析的维度尽可能多，跟踪系统能提供足够快的信息反馈，就可以对生产环境下的异常状况做出快速反应。分析的全面，能够避免二次开发。

5.2 全链路压测核心技术

5.2.1 全链路流量染色

做到微服务和中间件的染色标志的穿透

通过压测平台对输出的压力请求打上标识，在订单系统中提取压测标识，确保完整的程序上下文都持有该标识，并且能够穿透微服务以及各种中间件，比如 MQ，hystrix，Fegin等。

5.2.2 全链路服务监控

需要能够实时监控服务的运行状况以及分析服务的调用链，可以采用skywalking进行服务监控和压测分析

5.2.3 全链路日志隔离

做到日志隔离，防止污染生产日志

当订单系统向磁盘或外设输出日志时，若流量是被标记的压测流量，则将日志隔离输出，避免影响生产日志。

5.2.4 全链路风险熔断

流量控制，防止流量超载，导致集群不可用

当订单系统访问会员系统时，通过RPC协议延续压测标识到会员系统，两个系统之间服务通讯将会有白黑名单开关来控制流量流入许可，该方案设计可以一定程度上避免下游系统出现瓶颈或不支持压测所带来的风险，这里可以采用Sentinel来实现风险熔断。

5.3 全链路数据隔离

对各种存储服务以及中间件做到数据隔离，防止数据污染

5.3.1 数据库隔离

当会员系统访问数据库时，在持久化层同样会根据压测标识进行路由访问压测数据表。数据隔离的手段有多种，比如影子库、影子表，或者影子数据，三种方案的仿真度会有一定的差异，他们的对比如下。

	隔离性	兼容性	安全级别	技术难度
影子库	高	高	高	高
影子表	中	低	中	中
影子数据	低	低	低	低

5.3.2 消息队列隔离

当生产的消息扔到MQ之后，接着让消费者进行消费，这个没有问题，压测的数据不能够直接扔到MQ中的，因为它会被正常的消费者消费到的，要做好数据隔离，方案有队列隔离，消息隔离，他们对比如下

	隔离性	兼容性	安全级别	技术难度
队列隔离	高	好	高	高
消息隔离	低	低	低	中

5.3.3 Redis 隔离

通过 key 值来区分，压测流量的 key 值加统一后缀，通过改造RedisTemplate来实现key的路由。

6. 框架实现

6.1 流量染色方案

上面分析了从整体分析了全链路压测用的的核心技术，下面就来实现第一个流量染色。

6.1.1 流量识别

要想压测的流量和数据不影响线上真实的生产数据，就需要线上的集群能识别出压测的流量，只要能识别出压测请求的流量，那么流量触发的读写操作就很好统一去做隔离了。

全链路压测发起的都是Http的请求，只需要要请求头上添加统一的压测请求头。

通过在请求协议中添加压测请求的标识，在不同服务的相互调用时，一路透传下去，这样每一个服务都能识别出压测的请求流量，这样做的好处是与业务完全的解耦，只需要应用框架进行感知，对业务方代码无侵入。

'xxs'

6.1.2 MVC接收数据

客户端传递过来的数据可以通过获取Header的方式获取到，并将其设置进当前的ThreadLocal，交给后面的方法使用。

6.1.2.1 MVC拦截器实现

/**
 * 链路跟踪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 {

    }
}

6.1.2.2 Tomcat线程复用问题

tomcat默认使用线程池来管理线程，一个请求过来，如果线程池里面有空闲的线程，那么会在线程池里面取一个线程来处理该请求，一旦该线程当前在处理请求，其他请求就不会被分配到该线程上，直到该请求处理完成。请求处理完成后，会将该线程重新加入线程池，因为是通过线程池复用线程，就会如果线程内部的ThreadLocal没有清除就会出现问题，需要新的请求进来的时候，清除ThreadLocal。

6.1.3 Fegin传递传递染色标识

项目的微服务是使用Fegin来实现远程调用的，跨微服务传递染色标识是通过MVC拦截器获取到请求Header的染色标识，并放进ThreadLocal中，然后交给Fegin拦截器在发送请求之前从ThreadLocal中获取到染色标识，并放进Fegin构建请求的Header中，实现微服务之间的火炬传递。

'xxs'

6.1.3.1 代码实现

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());
        }
    }
}

6.1.4 Hystrix传递染色标识

6.1.4.1 Hystrix隔离技术

Hystrix 实现资源隔离，主要有两种技术：

信号量

信号量的资源隔离只是起到一个开关的作用，比如，服务 A 的信号量大小为 10，那么就是说它同时只允许有 10 个 tomcat 线程来访问服务 A，其它的请求都会被拒绝，从而达到资源隔离和限流保护的作用。

'xxs'

线程池

线程池隔离技术，是用 Hystrix 自己的线程去执行调用，而信号量隔离技术，是直接让 tomcat 线程去调用依赖服务，信号量隔离，只是一道关卡，信号量有多少，就允许多少个 tomcat 线程通过它，然后去执行。

'xxs'

7.1.4.2 Hystrix穿透

如果使用线程池模式，那么存在一个ThreadLocal变量跨线程传递的问题，即在主线程的ThreadLocal变量，无法在线程池中使用，不过Hystrix内部提供了解决方案。

'xxs'

封装Callable任务

这里使用装饰器模式传递上下文

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);
    }
}

实现Hystrix的并发策略类

因为Hystrix默认的并发策略不支持ThreadLocal传递，可以自定义并发策略类继承HystrixConcurrencyStrategy

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()));
    }
}

Hystrix注入新并发策略并进行刷新

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);
    }
}

6.2 数据隔离方案

6.2.1 JDBC数据源隔离

'xxs'

数据隔离需要对DB，Redis,RabbitMQ进行数据隔离

通过实现Spring动态数据源AbstractRoutingDataSource，通过ThreadLocal识别出来压测数据，如果是压测数据就路由到影子库，如果是正常流量则路由到主库，通过流量识别的改造，各个服务都已经能够识别出压测的请求流量了。

6.2.1.1 代码实现

数据源路由Key持有对象

根据路由Key将选择将操作路由给那个数据源

/**
 * 动态数据源上下文
 */
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);
    }
}

动态数据源实现类

根据路由Key实现数据源的切换

/**
 * 动态数据源实现类
 */
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();
    }

}

6.2.2 Redis 数据源隔离

同时通过ThreadLocal识别出来压测数据，自定义Redis的主键的序列化方式，如果是压测数据则在主键后面加上后缀，这样就可以通过不同主键将Redis数据进行隔离。

6.2.2.1 实现key序列化

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);
    }
}

6.2.2.2 配置序列化器

/**
 * 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;
    }
}

6.2.3 RabbitMQ 数据隔离

'xxs'

6.2.3.1 自动创建影子队列

因为SpringAMQP中的关键方法是私有的，无法通过继承的方式进行实现对以配置好的队列进行扩展，所以需要自定义该类，来实现对自动创建影子队列，并和交换器进行绑定

'xxs'

代码实现

改造RabbitListenerAnnotationBeanPostProcessor类来实现创建MQ影子队列以及将影子Key绑定到影子队列

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;
    }
}

6.2.3.2 传递染色标识

因为MQ是异步通讯，为了传递染色标识，会在发送MQ的时候将染色标识传递过来，MQ接收到之后放进当前线程的ThreadLocal里面，这个需要扩展Spring的SimpleRabbitListenerContainerFactory来实现

'xxs'

代码实现

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;
    }
}

6.2.3.3 发送MQ消息处理

同上，需要传递染色标识，就通过继承RabbitTemplate重写convertAndSend方法来实现传递染色标识。

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);
        }
    }
}

6.3 接口隔离方法

6.3.1 Mock 第三方接口

对于第三方数据接口需要进行隔离，比如短信接口，正常的数据需要发送短信，对于压测数据则不能直接调用接口发送短信，并且需要能够识别出来压测数据，并进行MOCK接口调用。

'xxs'

6.3.1.1 核心类实现

@Aspect
public class WormholeMockSection {

    /**
     * 切点 拦截@WormholeMock的注解
     */
    @Pointcut("@annotation(com.demo.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;
    }

}

6.3.1.2 使用方式

在具体方法上面加上注解就可以使用了

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

6.4 零侵入方案

如果开发的中间件需要各个微服务大量改造，对开发人员来说就是一个灾难，所以这里采用零侵入的springboot starter 来解决

6.4.1 自动装配

使用微服务得@Conditional来完成配置得自动装配，这里用MVC得配置来演示自动装配，其他得都是类似

这样可以最大限度的优化代码并提高很高的可扩展性。

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

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

6.4.1.1 Conditional 简介

@Conditional表示仅当所有指定条件都匹配时，组件才有资格注册。该@Conditional注释可以在以下任一方式使用：

作为任何@Bean方法的方法级注释
作为任何类的直接或间接注释的类型级别注释 @Component，包括@Configuration类
作为元注释，目的是组成自定义构造型注释

6.4.1.2 Conditional派生注解

@Conditional派生了很多注解，下面给个表格列举一下派生注解的用法

@Conditional派生注解	作用(都是判断是否符合指定的条件)
@ConditionalOnJava	系统的java版本是否符合要求
@ConditionalOnBean	有指定的Bean类
@ConditionalOnMissingBean	没有指定的bean类
@ConditionalOnExpression	符合指定的SpEL表达式
@ConditionalOnClass	有指定的类
@ConditionalOnMissingClass	没有指定的类
@ConditionalOnSingleCandidate	容器只有一个指定的bean，或者这个bean是首选bean
@ConditionalOnProperty	指定的property属性有指定的值
@ConditionalOnResource	路径下存在指定的资源
@ConditionalOnWebApplication	系统环境是web环境
@ConditionalOnNotWebApplication	系统环境不是web环境
@ConditionalOnjndi	JNDI存在指定的项

6.4.2 SpringBoot starter

和自动装配一样，Spring Boot Starter的目的也是简化配置，而Spring Boot Starter解决的是依赖管理配置复杂的问题，有了它，当我需要构建一个Web应用程序时，不必再遍历所有的依赖包，一个一个地添加到项目的依赖管理中，而是只需要一个配置spring-boot-starter-web。

6.4.2.1 使用规范

在 Spring Boot starter 开发规范中，项目中会有一个空的名为 xxx-spring-boot-starter 的项目，这个项目主要靠 pom.xml 将所有需要的依赖引入进来。同时项目还会有一个 xxx-spring-boot-autoconfigure 项目，这个项目主要写带 @Configuration 注解的配置类，在这个类或者类中带 @Bean 的方法上。

6.4.2.2 项目使用

在 xxx-spring-boot-starter的项目下的resources文件夹下面新建一个META-INF文件，并在下面创建spring.factories文件，将自动配置类配置进去

org.springframework.boot.autoconfigure.EnableAutoConfiguration=\\
com.demo.wormhole.autoconfiguration.WormholeAutoConfiguration

'xxs'