Dubbo 压测插件 2.0 —— 基于普通 API 调用

文 | 聂风 on 测试

插件已开源，详见 gatling-dubbo：https://github.com/youzan/gatling-dubbo.git

上一篇《Dubbo压测插件的实现——基于Gatling》中，我们介绍了基于 Dubbo 泛化调用实现的 Gatling Dubbo 压测插件，使用泛化调用发起 Dubbo 压测请求，consumer 端不需要拿到 provider 端的 API 包，使用上很便利，但是众所周知，Dubbo 泛化调用的性能不如普通 API 调用，虽然可以优化并使之达到与普通 API 调用相近的性能，但仍存在一些局限性。生产中除了网关等特殊应用外，一般很少使用泛化调用，如果以泛化调用的性能来表征生产中普通 API 调用的性能，其压测结论很难令人信服。做压测的时候，一般要求各种条件如环境等都尽可能保持一致。所以，我们又开发了基于普通 API 调用的 Gatling Dubbo 压测插件，即 gatling-dubbo2.0。此外，依托于 Gatling 强大的基础能力， gatling-dubbo2.0 相比于 Jmeter 还存在以下几方面的优势：

• 更强的场景编排能力，支持多场景同时编排，如仿真电商业务中同时存在普通下单、团购、秒杀等多种交易类型的场景

• 支持设置场景内流量模型，如漏斗模型，仿真用户从商品浏览 -> 加入购物车 -> 下单 -> 支付过程中的各级转化率

• 不需要安装额外插件，原生支持设置压力模型，如设置压测需要达到的目标 RPS，甚至逐级加压进行梯度压力测试

• 更低的资源消耗，更高的并发能力

  
  

一、插件主要组成

Action 和 ActionBuild
执行部分，这里的作用是发起 Dubbo 请求，校验请求结果并记录日志以便后续生成压测报告。ActionBuild 则为 DSL 使用 Action 的辅助类

Check 和 CheckBuild
校验部分，全链路压测中我们使用 json path 校验 HTTP 请求结果，这里我们实现了一样的校验方法，而且，对于一些不规范的返回结果（如返回了基本数据类型），还增加了自定义校验方法。CheckBuild 则为 DSL 使用 Check 的辅助类。

DSL
插件的领域特定语言，提供简单易用的 API 方便编写 Dubbo 压测脚本。

1.1 Action

DubboAction 包含了发起 Dubbo 请求、请求结果校验以及压力控制逻辑，需要扩展 Gatling 的 ExitableAction 并实现 execute 方法。

DubboAction 的入参 f 是一个函数，从压测脚本传入，函数负责组织 Dubbo 请求，从 session 中取值并动态构造请求参数。这一过程类似于使用 Jmeter 压测 Java 接口，即扩展 AbstractJavaSamplerClient。所以，gatling-dubbo 2.0 也支持非 dubbo 的其他 java 调用压测，因为 f 怎么写的控制权完全掌握在写压测脚本的人手里（本质上，远程调用和本地调用的客户端使用方式上并没有区别）。

所有虚拟用户以并发方式执行 execute 方法，每个用户又以异步方式执行 Dubbo 请求，且无论请求是否正确返回，都需要记录相应的成功或失败日志，失败可能是由于请求失败了，也可能是请求成功了，但是校验请求结果失败了。下一步就是准备发起新的 Dubbo 请求，如果开启了 Rps 阀门（throttled），则会根据当前的 Rps 和 Rps 阀门阈值动态调整发送请求的频率，在施压机（consumer）未达到性能瓶颈的情况下，可以很稳定的保持在设置的 Rps 目标值上进行压测。如果 Rps 阀门未开启，则直接发起新的 Dubbo 请求（通过 AKKA Message 触发）。

1. class DubboAction[A]( requestName: Expression[String],

2. f: (Session) => A,

3. val executor: ExecutorService,

4. val objectMapper: ObjectMapper,

5. checks: List[DubboCheck],

6. coreComponents: CoreComponents,

7. throttled: Boolean,

8. val next: Action

9. ) extends ExitableAction with NameGen {

10. ......

11. override def execute(session: Session): Unit = recover(session) {

12. requestName(session) map { reqName =>

13. val startTime = System.currentTimeMillis()

14. val fu = Future {

15. try {

16. f(session)

17. } finally {

18. }

19. }

20. 
    

21. fu.onComplete {

22. case Success(result) =>

23. val endTime = System.currentTimeMillis()

24. val resultJson = objectMapper.writeValueAsString(result)

25. val (newSession, error) = Check.check(resultJson, session, checks)

26. error match {

27. case None =>

28. statsEngine.logResponse(session, reqName, ResponseTimings(startTime, endTime), Status("OK"), None, None)

29. throttle(newSession(session))

30. case Some(Failure(errorMessage)) =>

31. statsEngine.logResponse(session, reqName, ResponseTimings(startTime, endTime), Status("KO"), None, Some(errorMessage))

32. throttle(newSession(session).markAsFailed)

33. }

34. 
    

35. case UFailure(e) =>

36. val endTime = System.currentTimeMillis()

37. statsEngine.logResponse(session, reqName, ResponseTimings(startTime, endTime), Status("KO"), None, Some(e.getMessage))

38. throttle(session.markAsFailed)

39. }

40. }

41. }

42. 
    

43. private def throttle(s: Session): Unit = {

44. if (throttled) {

45. coreComponents.throttler.throttle(s.scenario, () => next ! s)

46. } else {

47. next ! s

48. }

49. }

50. }

DubboActionBuilder 负责创建线程池并实例化 DubboAction：

1. case class DubboActionBuilder[A](requestName: Expression[String], f: (Session) => A, checks: List[DubboCheck], threadPoolSize: Int) extends ActionBuilder {

2. override def build(ctx: ScenarioContext, next: Action): Action = {

3. import ctx._

4. val executor = Executors.newFixedThreadPool(threadPoolSize)

5. val objectMapper: ObjectMapper = new ObjectMapper()

6. new DubboAction[A](requestName, f, executor, objectMapper, checks, coreComponents, throttled, next)

7. }

8. }

LambdaProcessBuilder 提供了设置 check 条件的 DSL 和 设置线程池大小的 DSL：

有赞的施压机是 4 核 8Gb 内存的，我们为其设置的默认线程池大小为 200，与 Dubbo 应用部署环境一致。你可以使用 DSL threadPoolSize(threadPoolSize: Int) 按照你的机器配置设置一个合适的线程池大小。如果施压机成了性能瓶颈，你可以考虑将其改造成集群来施压，具体可参考《有赞全链路压测引擎的设计与实现》

1. case class DubboProcessBuilder[A](requestName: Expression[String], f: (Session) => A, checks: List[DubboCheck] = Nil, threadPoolSize: Int = 200) extends DubboCheckSupport {

2. def check(dubboChecks: DubboCheck*): DubboProcessBuilder[A] = copy[A](checks = checks ::: dubboChecks.toList)

3. 
   

4. def threadPoolSize(threadPoolSize: Int): DubboProcessBuilder[A] = copy[A](threadPoolSize = threadPoolSize)

5. 
   

6. def build(): ActionBuilder = DubboActionBuilder[A](requestName, f, checks, threadPoolSize)

7. }

1.2 Check

全链路压测中，我们使用 json path 校验 HTTP 请求结果，Dubbo 压测插件中，我们也实现了基于 json path 的校验方法：

1. package object dubbo {

2. type DubboCheck = Check[String]

3. 
   

4. val DubboStringExtender: Extender[DubboCheck, String] =

5. (check: DubboCheck) => check

6. 
   

7. val DubboStringPreparer: Preparer[String, String] =

8. (result: String) => Success(result)

9. }

1. trait DubboJsonPathOfType {

2. self: DubboJsonPathCheckBuilder[String] =>

3. 
   

4. def ofType[X: JsonFilter](implicit extractorFactory: JsonPathExtractorFactory) = new DubboJsonPathCheckBuilder[X](path, jsonParsers)

5. }

6. 
   

7. object DubboJsonPathCheckBuilder {

8. val CharsParsingThreshold = 200 * 1000

9. 
   

10. def preparer(jsonParsers: JsonParsers): Preparer[String, Any] =

11. response => {

12. if (response.length() > CharsParsingThreshold || jsonParsers.preferJackson)

13. jsonParsers.safeParseJackson(response)

14. else

15. jsonParsers.safeParseBoon(response)

16. }

17. 
    

18. def jsonPath(path: Expression[String])(implicit extractorFactory: JsonPathExtractorFactory, jsonParsers: JsonParsers) =

19. new DubboJsonPathCheckBuilder[String](path, jsonParsers) with DubboJsonPathOfType

20. }

21. 
    

22. class DubboJsonPathCheckBuilder[X: JsonFilter](

23. private[check] val path: Expression[String],

24. private[check] val jsonParsers: JsonParsers

25. )(implicit extractorFactory: JsonPathExtractorFactory)

26. extends DefaultMultipleFindCheckBuilder[DubboCheck, String, Any, X](

27. DubboStringExtender,

28. DubboJsonPathCheckBuilder.preparer(jsonParsers)

29. ) {

30. import extractorFactory._

31. 
    

32. def findExtractor(occurrence: Int) = path.map(newSingleExtractor[X](_, occurrence))

33. def findAllExtractor = path.map(newMultipleExtractor[X])

34. def countExtractor = path.map(newCountExtractor)

35. }

但有时候存在一些不规范的情况，dubbo 接口的返回结果并不能直接转化为 json，如返回了基本数据类型，所以我们还提供了自定义校验方法，可以将这样的返回结果转化为 String 类型，并使用字符串比较、正则表达式匹配等方法校验返回结果：

1. case class DubboCustomCheck(func: String => Boolean, failureMessage: String = "Dubbo check failed") extends DubboCheck {

2. override def check(response: String, session: Session)(implicit cache: mutable.Map[Any, Any]): Validation[CheckResult] = {

3. func(response) match {

4. case true => CheckResult.NoopCheckResultSuccess

5. case _ => Failure(failureMessage)

6. }

7. }

8. }

DubboCheckSupport 则提供了 json path、 custom 两种检验方式的 DSL

1. trait DubboCheckSupport {

2. def jsonPath(path: Expression[String])(implicit extractorFactory: JsonPathExtractorFactory, jsonParsers: JsonParsers) =

3. DubboJsonPathCheckBuilder.jsonPath(path)

4. 
   

5. def custom = DubboCustomCheck

6. }

Dubbo 压测脚本中可以设置一个或多个 check 来校验请求结果

1.3 DSL

DubboDsl 提供顶层 DSL，隐式方法 dubboProcessBuilder2ActionBuilderScala 用于自动从 DubboProcessBuilder 构造 ActionBuilder

1. trait DubboDsl extends DubboCheckSupport {

2. def dubbo[A](requestName: Expression[String], f: (Session) => A) = DubboProcessBuilder[A](requestName, f)

3. 
   

4. implicit def dubboProcessBuilder2ActionBuilder[A](builder: DubboProcessBuilder[A]): ActionBuilder = builder.build()

5. }

二、示例

2.1 压测脚本示例

1. class Mix extends Simulation {

2. val application = new ApplicationConfig()

3. application.setName("gatling-dubbo")

4. 
   

5. // 初始化 AService

6. val referenceAService = new ReferenceConfig[AService]

7. referenceAService.setApplication(application)

8. referenceAService.setUrl("dubbo://IP:PORT/com.xxx.service.AService")

9. referenceAService.setInterface(classOf[AService])

10. val aService = referenceAService.get()

11. 
    

12. // 初始化 BService

13. val referenceBService = new ReferenceConfig[BService]

14. referenceBService.setApplication(application)

15. referenceBService.setUrl("dubbo://IP:PORT/com.yyy.service.BService")

16. referenceBService.setInterface(classOf[BService])

17. val bService = referenceBService.get()

18. 
    

19. // 设置数据源

20. val jsonFileFeeder = jsonFile("data.json").shuffle.circular

21. val mixScenario = scenario("scenario of mix")

22. .forever("tripsCount") {

23. feed(jsonFileFeeder)

24. .randomSwitch(11d -> exec(

25. dubbo("com.xxx.service.AService.aMethod", fAMethod)

26. .check(jsonPath("$.success").is("true"))

27. )

28. )

29. .randomSwitch(4d -> exec(

30. dubbo("com.yyy.service.BService.bMethod", fBMethod)

31. .check(jsonPath("$.success").is("true"))

32. )

33. )

34. .randomSwitch(5d -> exec(

35. ......

36. )

37. ......

38. )

39. }

40. 
    

41. setUp(mixScenario.inject(constantUsersPerSec(100) during (10 seconds)).throttle(reachRps(1000) in (1 seconds), holdFor(120 seconds)))

42. 
    

43. // 设置 aMethod 的请求参数并调用

44. def fAMethod(session: Session): Object = {

45. val aParam = new AParam()

46. aParam.setName("A Name");

47. // 从 session 中获取动态参数并设置

48. aParam.setAId(session.attributes("aId").asInstanceOf[Integer].toLong);

49. aService.aMethod(aParam);

50. }

51. 
    

52. // 设置 bMethod 的请求参数并调用

53. def fBMethod(session: Session): Object = {

54. val bParam = new BParam()

55. bParam.setAge(26)

56. // 从 session 中获取动态参数并设置

57. bParam.setBId(session.attributes("bId").asInstanceOf[Integer].toLong)

58. bService.bMethod(bParam);

59. }

60. 
    

61. def fXxx(session: Session): Object = {

62. ......

63. }

64. }

randomSwitch 的作用：
以上示例其实是 gatling-dubbo 在有赞的一个典型使用场景，即评估一个应用的单实例性能。按生产环境真实的接口调用比例请求各个接口（该比例由场景执行各个请求的概率分布模拟），这样的压测结果就可以真实反映生产环境应用的单实例性能，并为容量报警、生产扩容等提供参考依据。

2.2 压测数据示例

1. [

2. {

3. "aId": 160,

4. "bId": 859296

5. },

6. {

7. "aId": 160,

8. "bId": 1019040

9. },

10. {

11. "aId": 160,

12. "bId": 1221792

13. },

14. ......

15. ]

压测数据使用 Json 数组保存，其中每一个 Json 对象都包含了一次压测请求所需的所有动态参数，且为了方便通过 session 设置动态参数，Json 对象中不再嵌套其他 Json 对象。

2.3 压测报告示例

1、应用基线性能评估，用于精准扩容：

2、中心化限流效果验证：

有赞测试组在持续招人中，大量岗位空缺，只要你来，就能帮你点亮全栈开发技能树，有意向换工作的同学可以发简历到 sunjun@youzan.com

相关阅读

1. Mock服务插件在接口测试中的设计与应用

2. 有赞前端质量保障体系

3. Dubbo压测插件的实现——基于Gatling

4. 有赞全链路压测实战

-The End-

Vol.196

微商城｜零售｜美业 | 教育

微信公众号：有赞coder    微博：@有赞技术

技术博客：tech.youzan.com

The bigger the dream, 

the more important the team.