全链路压测的整体架构设计，以及5种实现方案流量染色方案、数据隔离方案、 ...

业务模块介绍

现在我们对整体的业务进行介绍以及演示

5. 全链路整体架构

上面介绍了为什么需要全链路压测，下面来看下全链路压测的整体架构。

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

5.1 需要应对的问题

5.1.1  业务问题

如何开展全链路压测？在说这个问题前，我们先考虑下，全链路压测有哪些问题比较难解决。

• 涉及的系统太多，牵扯的开发人员太多
  ​                在压测过程中，做一个全链路的压测一般会涉及到大量的系统，在整个压测过程中，光各个产品的人员协调就是一个比较大的工程，牵扯到太多的产品经理和开发人员，如果公司对全链路压测早期没有足够的重视，那么这个压测工作是非常难开展的。
  
• 模拟的测试数据和访问流量不真实
  ​                在压测过程中经常会遇到压测后得到的数据不准确的问题，这就使得压测出的数据参考性不强，为什么会产生这样的问题？主要就是因为压测的环境可能和生成环境存在误差、参数存在不一样的地方、测试数据存在不一样的地方这些因素综合起来导致测试结果的不可信。
  
• 压测生产数据未隔离，影响生产环境
  ​                在全链路压测过程中，压测数据可能会影响到生产环境的真实数据，举个例子，电商系统在生产环境进行全链路压测的时候可能会有很多压测模拟用户去下单，如果不做处理，直接下单的话会导致系统一下子会产生很多废订单，从而影响到库存和生产订单数据，影响到日常的正常运营。
  

5.1.2  技术问题

5.1.2.1 探针的性能消耗

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

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

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

​                对于应用的程序员来说，是不需要知道有跟踪系统这回事的。如果一个跟踪系统想生效，就必须需要依赖应用的开发者主动配合，那么这个跟踪系统也太脆弱了，往往由于跟踪系统在应用中植入代码的bug或疏忽导致应用出问题，这样才是无法满足对跟踪系统“无所不在的部署”这个需求。
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 全链路数据隔离

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

2.3.1 数据库隔离

​                当会员系统访问数据库时，在持久化层同样会根据压测标识进行路由访问压测数据表。数据隔离的手段有多种，比如影子库、影子表，或者影子数据，三种方案的仿真度会有一定的差异，他们的对比如下。
隔离性兼容性安全级别技术难度影子库高高高高影子表中低中中影子数据低低低低5.3.2 消息队列隔离

​                当我们生产的消息扔到MQ之后，接着让消费者进行消费，这个没有问题，压测的数据不能够直接扔到MQ中的，因为它会被正常的消费者消费到的，要做好数据隔离，方案有队列隔离，消息隔离，他们对比如下。
隔离性兼容性安全级别技术难度队列隔离高好高高消息隔离低低低中5.3.3 Redis 隔离

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

6.1  流量染色方案

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

6.1.1 流量识别

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

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

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

6.1.2  MVC接收数据

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

1. /**
2. * 链路跟踪Request设置值
3. */
4. public class MvcWormholeWebInterceptor implements WebRequestInterceptor {
7.     @Override
8.     public void preHandle(WebRequest webRequest) {
9.         //失效上下文，解决Tomcat线程复用问题
10.         WormholeContextHolder.invalidContext();
11.         String wormholeValue = webRequest.getHeader(WormholeContextHolder.WORMHOLE_REQUEST_MARK);
12.         if (StringUtils.isNotEmpty(wormholeValue)) {
13.             WormholeContextHolder.setContext(new WormholeContext(wormholeValue));
14.         }
15.     }
17.     @Override
18.     public void postHandle(WebRequest webRequest, ModelMap modelMap) throws Exception {
20.     }
22.     @Override
23.     public void afterCompletion(WebRequest webRequest, Exception e) throws Exception {
25.     }
26. }

复制代码

6.1.2.2 Tomcat线程复用问题

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

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

6.1.3.1 代码实现

1. public class WormholeFeignRequestInterceptor implements RequestInterceptor {
3.     @Override
4.     public void apply(RequestTemplate requestTemplate) {
5.         WormholeContext wormholeContext = WormholeContextHolder.getContext();
6.         if (null != wormholeContext) {
7.             requestTemplate.header(WormholeContextHolder.WORMHOLE_REQUEST_MARK, wormholeContext.toString());
8.         }
9.     }
10. }

复制代码

6.1.4  Hystrix传递染色标识

6.1.4.1 Hystrix隔离技术

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

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

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

6.1.4.2 Hystrix穿透

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

封装Callable任务

1. public final class DelegatingWormholeContextCallable<V> implements Callable<V> {
2.     private final Callable<V> delegate;
3.     // 用户信息上下文（根据项目实际情况定义ThreadLocal上下文）
4.     private WormholeContext orginWormholeContext;
6.     public DelegatingWormholeContextCallable(Callable<V> delegate,
7.                                              WormholeContext wormholeContext) {
8.         this.delegate = delegate;
9.         this.orginWormholeContext = wormholeContext;
10.     }
12.     public V call() throws Exception {
13.         //防止线程复用销毁ThreadLocal的数据
14.         WormholeContextHolder.invalidContext();
15.         // 将当前的用户上下文设置进Hystrix线程的TreadLocal中
16.         WormholeContextHolder.setContext(orginWormholeContext);
17.         try {
18.             return delegate.call();
19.         } finally {
20.             // 执行完毕，记得清理ThreadLocal资源
21.             WormholeContextHolder.invalidContext();
22.         }
23.     }
25.     public static <V> Callable<V> create(Callable<V> delegate,
26.                                          WormholeContext wormholeContext) {
27.         return new DelegatingWormholeContextCallable<V>(delegate, wormholeContext);
28.     }
29. }

复制代码

实现Hystrix的并发策略类

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

1. public class ThreadLocalAwareStrategy extends HystrixConcurrencyStrategy {
3.     // 最简单的方式就是引入现有的并发策略，进行功能扩展
4.     private final HystrixConcurrencyStrategy existingConcurrencyStrategy;
6.     public ThreadLocalAwareStrategy(
7.             HystrixConcurrencyStrategy existingConcurrencyStrategy) {
8.         this.existingConcurrencyStrategy = existingConcurrencyStrategy;
9.     }
11.     @Override
12.     public BlockingQueue<Runnable> getBlockingQueue(int maxQueueSize) {
13.         return existingConcurrencyStrategy != null
14.                 ? existingConcurrencyStrategy.getBlockingQueue(maxQueueSize)
15.                 : super.getBlockingQueue(maxQueueSize);
16.     }
18.     @Override
19.     public <T> HystrixRequestVariable<T> getRequestVariable(
20.             HystrixRequestVariableLifecycle<T> rv) {
21.         return existingConcurrencyStrategy != null
22.                 ? existingConcurrencyStrategy.getRequestVariable(rv)
23.                 : super.getRequestVariable(rv);
24.     }
26.     @Override
27.     public ThreadPoolExecutor getThreadPool(HystrixThreadPoolKey threadPoolKey,
28.                                             HystrixProperty<Integer> corePoolSize,
29.                                             HystrixProperty<Integer> maximumPoolSize,
30.                                             HystrixProperty<Integer> keepAliveTime, TimeUnit unit,
31.                                             BlockingQueue<Runnable> workQueue) {
32.         return existingConcurrencyStrategy != null
33.                 ? existingConcurrencyStrategy.getThreadPool(threadPoolKey, corePoolSize,
34.                 maximumPoolSize, keepAliveTime, unit, workQueue)
35.                 : super.getThreadPool(threadPoolKey, corePoolSize, maximumPoolSize,
36.                 keepAliveTime, unit, workQueue);
37.     }
39.     @Override
40.     public <T> Callable<T> wrapCallable(Callable<T> callable) {
41.         return existingConcurrencyStrategy != null
42.                 ? existingConcurrencyStrategy
43.                 .wrapCallable(new DelegatingWormholeContextCallable<>(callable, WormholeContextHolder.getContext()))
44.                 : super.wrapCallable(new DelegatingWormholeContextCallable<T>(callable, WormholeContextHolder.getContext()));
45.     }
46. }

复制代码

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

1. public class HystrixThreadLocalConfiguration {
3.     @Autowired(required = false)
4.     private HystrixConcurrencyStrategy existingConcurrencyStrategy;
6.     @PostConstruct
7.     public void init() {
8.         // Keeps references of existing Hystrix plugins.
9.         HystrixEventNotifier eventNotifier = HystrixPlugins.getInstance()
10.                 .getEventNotifier();
11.         HystrixMetricsPublisher metricsPublisher = HystrixPlugins.getInstance()
12.                 .getMetricsPublisher();
13.         HystrixPropertiesStrategy propertiesStrategy = HystrixPlugins.getInstance()
14.                 .getPropertiesStrategy();
15.         HystrixCommandExecutionHook commandExecutionHook = HystrixPlugins.getInstance()
16.                 .getCommandExecutionHook();
18.         HystrixPlugins.reset();
20.         HystrixPlugins.getInstance().registerConcurrencyStrategy(new ThreadLocalAwareStrategy(existingConcurrencyStrategy));
21.         HystrixPlugins.getInstance().registerEventNotifier(eventNotifier);
22.         HystrixPlugins.getInstance().registerMetricsPublisher(metricsPublisher);
23.         HystrixPlugins.getInstance().registerPropertiesStrategy(propertiesStrategy);
24.         HystrixPlugins.getInstance().registerCommandExecutionHook(commandExecutionHook);
25.     }
26. }

复制代码

6.2 数据隔离方案

6.2.1 JDBC数据源隔离

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

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

数据源路由Key持有对象

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

[code]/** * 动态数据源上下文 */public class DynamicDataSourceContextHolder {    public static final String PRIMARY_DB = "primary";    public static final String SHADOW_DB = "shadow";    private static final ThreadLocal contextHolder = new ThreadLocal() {        /**         * 将 master 数据源的 key作为默认数据源的 key         */        @Override        protected String initialValue() {            return PRIMARY_DB;        }    };    /**     * 数据源的 key集合，用于切换时判断数据源是否存在     */    public static List 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