基于APM实现RPC服务和消息队列的指定消费

 

本文内容是基于公司现有框架整理的一篇专利文章.该框架包含完整的一套DevOps流程,包括工单系统(容器申请、服务部署等)配置中心路由配置中心服务治理平台消息治理平台葛朗台(基于Docker+K8S)等.

该专利的目的为:在业务场景比较复杂,业务流程比较长并且涉及Dubbo消息队列等服务调用时,指定服务路由和消息路由等,从而实现服务和消息的指定消费功能,可方便测试或者开发人员的工作.

转载请说明出处.

 1.背景

　　随着公司业务的发展和增长，一个完整的业务流程涉及到很多服务，各个服务可能同时又因为版本迭代或者其他需求部署了多套环境，导致同一服务存在多个版本，而各个服务之间可能通过RPC服务或者消息队列存在关联，使整个业务流程的服务调用呈现为树状结构，每一个分支是一种可能的调用链路。在测试过程中，测试人员想要按照特定的调用链路进行功能测试变得错综复杂，难以实现。

　　在进行某个服务的功能测试时，很多情况下需要调用指定生产者的RPC服务或者消息被指定消费者消费。对于RPC服务调用来说，消费者服务调用提供者往往会随机调用一个提供者服务；对于消息队列来说，消息会被随机的消费者消费，从而达不到调用指定环境的某个服务或者消息被指定服务的消费者消费的目的，为测试带来很大不便，导致工作效率低。

当前测试过程存在的问题总结如下：

1. 服务调用链路无法指定。服务与服务之间调用链路复杂，无法指定具体的服务调用链路，服务与服务之间随机调用，达不到测试目的。
2. RPC消费者无法指定消费特定的提供者。每个服务存在多个版本，导致RPC服务提供者会存在多个，当RPC消费者调用提供者时，会随机调用一个提供者的服务，无法实现服务的指定消费功能。
3. MQ队列消息无法被指定的消费者消费。MQ生产者生产的消息，会被随机的消费者消费，特定消息被指定的消费者消费无法实现。
4. 完整的服务调用链路无法可视化。对于用户，服务的调用过程完全透明，无法直观的查看服务调用过程，即服务调用的具体节点信息等。

　　针对以上问题，提出了基于应用性能管理(Application Performance Management，简称APM)和java agent技术在对业务代码无侵入、完全透明的情况下，结合路由码(Route Code)，实现RPC服务(下以Dubbo为例)、Kafka、RabbitMQ等消息队列的指定消费功能，也称为路由功能，实现了服务调用按照指定的调用链路执行，极大的提高了测试效率和测试的方便性。

2.平台框架图

 

架构说明：

1. 在通常情况下，业务的测试流程是通过前端H5页面或者通过接口工具等触发测试流程。其中整个业务流程可能的涉及的调用链路包括Http、SpringMVC、Dubbo、RabbitMQ、Kafka等。
2. Squid代理，负责将不管是IP访问还是域名访问，统一构建出Http Header，置入用户IP或用户标识，即路由码。
3. 基于现有的服务调用链路，提出路由码(Route Code)概念，结合APM和java agent技术，在对业务代码无侵入，完全透明的情况下，实现路由码、服务调用链路的层层服务传递和基于路由码的指定消费功能，包括Dubbo指定消费、RabbitMQ队列指定消费和Kafka指定消费。

具体技术方案描述如下：

1. 首先建立路由配置中心(Route Configure Center)，配置路由码(路由码可以是IP或者随机生产的字符串)对应的消费关系等元数据，即携带该路由码的Dubbo消费者或者消息将调用哪些服务提供者或者被哪些消费者消费。通过改造Skywalking(APM)，结合TransmittableThreadLocal(在使用线程池等会池化复用线程的组件情况下，提供ThreadLocal值的传递功能，解决异步执行时上下文传递的问题)，将路由码在服务调用链路过程中向下传递，开发符合业务流程需求的Snowwalking。
2. 通过Snowwalking，开发SpringMVC拦截器。对SpringMVC的DispatcherServlet类进行拦截增强，获取请求的Http Header作为路由码并组装链路信息，放入TransmittableThreadLocal中，便于后继服务调用路由码的向下传递。
3. 通过Snowalking，开发Http的拦截器。对Http请求进行拦截增强，如果TransmittableThreadLocal中存在路由码和链路信息，则将TransmittableThreadLocal内容取出放入Http请求Header中，否则获取当前请求节点IP作为路由码放入TransmittableThreadLocal中，并设置入Http请求的Header中。
4. 通过Snowwalking，开发RabbitMQ拦截器。对RabbitMQ的生产者和消费者分别做拦截增强。对于生产者，对生产类进行拦截增强，将TransmittableThreadLocal内容取出放入消息的Properties中，将路由码和链路信息向下传递。对于消费者，对消费者的消费方法进行拦截增强，当消费者随机拉取一条消息后，将消息Properties中的路由码取出，根据在路由配置中心配置和消息治理平台(Message Governance Platform,简称MGP，用于配置服务之间消息发布、消费关系等元数据)判断当前消费者是否有消费权限，如果有消费权限，则当前消费者可继续消费该消息，否则，该消费者无法消费该消息，将该消息重新入队到消息队列，再重新从消息队列拉取其他消息进行消费，其中还包括处理特殊情况下的消息丢弃和消息重发等。
5. 通过Snowwalking，开发Dubbo拦截器。对Dubbo消费者的调用入口方法进行拦截，根据当前TransmittableThreadLocal中的路由码、路由配置中心路由元数据以及服务治理平台(Service Governance Platform，简称SGP，用于配置各个服务发布、消费关系等元数据)生产消费关系元数据，获取当前Dubbo消费者可消费的生产者，过滤Invoker列表，将符合配置的生产者返回给消费者，从而实现Dubbo服务根据路由码实现指定消费功能。
6. 通过Snowwalking，开发Kafka拦截器。对Kafka的生产者和消费者分别进行拦截增强。对于生产者，对消息发送类进行拦截增强，如果Kafka的版本低于0.11，则从TransmittableThreadLocal中解析出路由码，通过特定格式存入消息的Payload中；如果Kafka的版本高于0.11版本，则将TransmittableThreadLocal中的路由码以及链路信息存入消息的Header中，发送至Kafka。同理，对于消费者，想要实现Kafka的指定消费，前提是通过参数增强或者其他方式，将消费者的group.id隔离成不同的值，使需要执行指定消费的消费者属于不同的group，这样每个消费者都可以消费某个Topic的所有partition。然后对Kafka的消费方法进行拦截增强，如果Kafka的版本低于0.11，则消费服务解析消息Payload，获取路由码，然后通过路由配置中心配置的消费关系，判断当前消费者是否有权限消费该条消息，如果有权限，则消费，否则，跳过该消息，继续执行获取下一条消息进行消费判断；如果Kafka的版本高于0.11版本，则将路由码和链路信息从消息Header中获取出来，解析出路由码，再通过统一路由配置中心配置的消费关系，判断当前消费者是否有权限消费该条消息，逻辑同上。
7. 如果存在服务其他调用方式，同样可以对其进行增强拦截，实现路由码和调用链路的向下传递等，而保证服务调用链路不间断。
8. 通过以上对各种调用链路做拦截增强，即可将路由码结合APM、TransmittableThreadLocal和java agent技术在各种业务场景下进行传递，根据路由配置等元数据实现携带指定路由码服务或者消息的指定消费功能。

3.关键技术点

　　1.路由码(Route Code)传递

　　路由码是指定消费的基础。服务发起者在调用提供者时，会携带路由码，服务提供者提供服务的同时，会通过Snowwalking获取消费者的路由码，并放入TransmittableThreadLocal中，使路由码在链路中进行传递。下图为一种可能的服务调用链路路由码传递时序图：

  

　　2.Snowwalking

　　Snowwalking基于Skywalking改造，结合TransmittableThreadLocal，使路由码在跨线程的情况下，依然可以进行传递。通过Snowwalking对服务链路的各种调用方式进行拦截增强，达到路由码、调用链路信息向下传递和指定消费等功能，对业务代码零侵入且完全透明。

4.效果

　　实现该技术方案之后，测试人员或者开发人员只需要进行简单的配置即可实现RPC服务、RabbitMQ和Kafka等消息队列的指定消费功能，满足了测试需求的同时极大的提高了相关工作人员的工作效率，该技术方案简单高效，针对开发人员和测试人员完全透明且对业务代码无侵入，不会影响生产环境服务和消息的随机消费。

　　下图给出了一种服务指定消费关系。假设路由配置中心配置路由码code1对应的路由链路为B1,C2,D3，则前端H5页面触发测试流程时,通过Squid代理将路由Code(code1)设置到Http请求的Header当中，当SpringMVC接收到Http请求时,就会被SpringMVC plugin拦截,取出Http请求Header中的路由Code(code1)放入TransmittableThreadLocal中.当SpringMVC需要调用Dubbo服务B时,在服务调用之前就会被Dubbo plugin拦截,根据路由Code(code1)和路由Code配置的服务调用关系,过滤服务B的Invoker List,从而过滤出B1为需要指定调用的服务,进而定向调用至服务B1,服务B调用服务C和服务C调用服务D逻辑相同，从而完成RPC服务的指定消费,其他情况同理,不再赘述。