性能工具之JMeter5.0核心类StandardJMeterEngine源码分析 _源码分析

仓廪实则知礼节，衣食足则知荣辱。这篇文章主要讲述性能工具之JMeter5.0核心类StandardJMeterEngine源码分析相关的知识，希望能为你提供帮助。
概述JMeter 默认单机压测引擎，运行 JMeter 测试，直接用于本地 GUI 和非 GUI 调用，或者RemoteJMeterEngineImpl 在服务器模式下运行时启动。
API地址：https://jmeter.apache.org/api/org/apache/jmeter/engine/StandardJMeterEngine.html
工程位置

文章图片

逻辑关系

文章图片

简要解读：

HashTree是依赖的数据结构；
SearchByClass 用来查找 HashTree 中的所有节点，并把节点实例化为真正的对象，例如图中TestPlan/ThreadGroup/javaSampler/ResultCollector 在 HashTree 中本来都是只是配置，全部通过 SearchByClass 实例化的；
实例化出来的对象如果是 TestStateListener 类型，则会在有生命周期的函数回调，测试前调 testStarted,结束掉 testEnded, 比如 ResultCollector是该类型的一种，在结束的时候回调 testEnded 方法完成 report 的写入；
PreCompiler 用来解析 Arguments, 把 TestPlan 节点中配置的参数作为JMeterVariables 加入到测试线程上线文中；
ThreadGroup 用来用来管理一组线程，包括线程的个数/启动/关闭等；StopTest 作为其内部类对外不可见，作为一个 Runnable，作用是异步停止测试，stopTest方法也是通过该内部类实现的。

主要变量注意关键字 volatile

1. // 灵魂级变量,注意关键字volatile

2. private static volatile StandardJMeterEngine engine;

构造函数有两种构造函数，带参和不带参

1. // 不带参构造函数

2. public StandardJMeterEngine() {

3.this(null);

4. }

5.

6. // 带参构造函数

7. public StandardJMeterEngine(String host) {

8.this.host = host;

9.// Hack to allow external control

10.initSingletonEngine(this);

11. }

主要方法askThreadsToStop
清洁关闭，即等待当前运行的采样器结束

1./**

2.* Clean shutdown ie, wait for end of current running samplers

3.*/

4.public void askThreadsToStop() {

5.if (engine != null) { // Will be null if StopTest thread has started

6.engine.stopTest(false);

7.}

8.}

reset
JMeterEngine 如果运行则停止

1.// 重置。在StandardJMeterEngine中就是直接调用stopTest(true).

2.@Override

3.public void reset() {

4.if (running) {

5.stopTest();

6.}

7.}

configure(HashTree testTree)
配置引擎，HashTree 是 JMeter 执行测试依赖的数据结构，configure 在执行测试之前进行配置测试数据。

1. // HashTree是JMeter执行测试依赖的数据结构，configure在执行测试之前进行配置测试数据

2.// 从HashTree中解析出TestPlan, 获取TestPlan的serialized和tearDownOnShutdown并保存为local属性，同时把整个HashTree也保存到local。

3.// StandardJMeterEngine依赖线程组ThreadGroup, 一个测试中可能会有多个线程组，如果serialized为true，则StandardJMeterEngine会串行的去执行这些线程组，每启动一个ThreadGroup主线程都会等它结束；否则就并行执行所有的线程组。

4.// tearDownOnShutdown与PostThreadGroup配合使用的，这个Special Thread Group专门用来做清理工作

5.

6.@Override

7.public void configure(HashTree testTree) {

8.// Is testplan serialised?

9.SearchByClass< TestPlan> testPlan = new SearchByClass< > (TestPlan.class);

10.testTree.traverse(testPlan);

11.Object[] plan = testPlan.getSearchResults().toArray();

12.if (plan.length == 0) {

13.throw new IllegalStateException("Could not find the TestPlan class!");

14.}

15.TestPlan tp = (TestPlan) plan[0];

16.serialized = tp.isSerialized();

17.tearDownOnShutdown = tp.isTearDownOnShutdown();

18.active = true;

19.test = testTree;

20.}

exit
远程退出由 RemoteJMeterEngineImpl.rexit() 和notifyTestListenersOfEnd() 调用 iff exitAfterTest 为 true; 反过来，run( ) 方法调用，也调用 StopTest 类

1. /**

2.* Remote exit

3.* Called by RemoteJMeterEngineImpl.rexit()

4.* and by notifyTestListenersOfEnd() iff exitAfterTest is true;

5.* in turn that is called by the run() method and the StopTest class

6.* also called

7.*

8.* 是为Remote Test准备的

9.* 如果当前的测试是从一个客户端的JMeter执行远程JMeterEngine的remote samples，则应该调用该exit()方法来关闭远程的测试

10.* 被RemoteJMeterEngineImpl.rexit()调用和exitAfterTest为真时被notifyTestListenersOfEnd()调用

11.*/

12.@Override

13.public void exit() {

14.ClientJMeterEngine.tidyRMI(log); // This should be enough to allow server to exit.

15.if (REMOTE_SYSTEM_EXIT) { // default is false

16.log.warn("About to run System.exit(0) on {}", host);

17.// Needs to be run in a separate thread to allow RMI call to return OK

18.Thread t = new Thread() {

19.@Override

20.public void run() {

21.pause(1000); // Allow RMI to complete

22.log.info("Bye from {}", host);

23.System.out.println("Bye from "+host); // NOSONAR Intentional

24.System.exit(0); // NOSONAR Intentional

25.}

26.};

27.t.start();

28.}

29.}

isActive
isActive 在测试中 JMeterEngine 返回值：
boolean 用于显示引擎是否处于活动状态的标志（在测试运行时为true）。在测试结束时设置为 false。

1. /**

2.*引擎是否有效的标识，在测试结束时设为false

3.*在confgiure()的时候设该值为true，在执行完测试(指的是该JMeterEngine所有ThreadGroup)之后设置为false。

4.*如果active==true，则说明该JMeterEngine已经配置完测试并且还没执行完，我们不能再进行configure或者runTest了；

5.*若active == false, 则该JMeterEngine是空闲的，我们可以重新配置HashTree，执行新的测试.

6.*

7.* @return

8.*/

9.

10.@Override

11.public boolean isActive() {

12.return active;

13.}

engine
操作 engine，initSingletonEngine()、initSingletonEngine()、stopEngineNow()、stopEngine

1./**

2.* Set the shared engine

3.* 操作 engine，initSingletonEngine()、initSingletonEngine()、stopEngineNow()、stopEngine

4.* @param standardJMeterEngine

5.*/

6.private static void initSingletonEngine(StandardJMeterEngine standardJMeterEngine) {

7.StandardJMeterEngine.engine = standardJMeterEngine;

8.}

9.

10.public static void stopEngineNow() {

11.if (engine != null) {// May be null if called from Unit test

12.engine.stopTest(true);

13.}

14.}

15.

16.public static void stopEngine() {

17.if (engine != null) { // May be null if called from Unit test

18.engine.stopTest(false);

19.}

20.}

run
run()，启动测试。
JMeterContextService 清零：numberOfActiveThreads=0, 重置 testStart时间

1. JMeterContextService.startTest();

JMeterContextService.startTest():

1. /**

2.* Method is called by the JMeterEngine class when a test run is started.

3.* Zeroes numberOfActiveThreads.

4.* Saves current time in a field and in the JMeter property "TESTSTART.MS"

5.*/

6.public static synchronized void startTest() {

7.if (testStart == 0) {

8.numberOfActiveThreads = 0;

9.testStart = System.currentTimeMillis();

10.JMeterUtils.setProperty("TESTSTART.MS",Long.toString(testStart)); // $NON-NLS-1$

11.}

12.}

PreCompiler the Tashree，见上面的简要解读

1. try {

2.PreCompiler compiler = new PreCompiler();

3.test.traverse(compiler);

4. } catch (RuntimeException e) {

5.log.error("Error occurred compiling the tree:", e);

6.JMeterUtils.reportErrorToUser("Error occurred compiling the tree: - see log file", e);

7.return; // no point continuing

8. }

利用 SearchByClass 解析所有 TestStateListener 加入到 testList 中

1. SearchByClass< TestStateListener> testListeners = new SearchByClass< > (TestStateListener.class); // TL-S& E

2. test.traverse(testListeners);

3. // Merge in any additional test listeners

4. // currently only used by the function parser

5. testListeners.getSearchResults().addAll(testList);

触发上一步中解析的 testListener 的 testStarted 方法：ResultCollector 会递增 instanceCount,初始化 fileOutput；TestPlan 会设置 FileServer 的basedir，添加 classpath; JavaSampler 会初始化真正要跑的AbstractJavaSamplerClient 类；
利用 SearchByClass 解析所有 ThreadGroup（包括SetupThreadGroup，ThreadGroup, PostThreadGroup）

1. notifyTestListenersOfStart(testListeners);

2.

3. private void notifyTestListenersOfStart(SearchByClass< TestStateListener> testListeners) {

4.for (TestStateListener tl : testListeners.getSearchResults()) {

5.if (tl instanceof TestBean) {

6.TestBeanHelper.prepare((TestElement) tl);

7.}

8.if (host == null) {

9.tl.testStarted();

10.} else {

11.tl.testStarted(host);

12.}

13.}

14. }

实例化一个 ListenerNotifier 实例，用来通知事件发生

1. ListenerNotifier notifier = new ListenerNotifier();

【性能工具之JMeter5.0核心类StandardJMeterEngine源码分析】启动所有 SetupThreadGroup (一般情况下没有 SetupThreadGroup )并等待到都结束

1. if (setupIter.hasNext()) {

2.log.info("Starting setUp thread groups");

3.while (running & & setupIter.hasNext()) {// for each setup thread group

4.AbstractThreadGroup group = setupIter.next();

5.groupCount++;

6.String groupName = group.getName();

7.log.info("Starting setUp ThreadGroup: " + groupCount + " : " + groupName);

8.startThreadGroup(group, groupCount, setupSearcher, testLevelElements, notifier);

9.if (serialized & & setupIter.hasNext()) {

10.log.info("Waiting for setup thread group: " + groupName

11.+ " to finish before starting next setup group");

12.group.waitThreadsStopped();

13.}

14.}

15.log.info("Waiting for all setup thread groups to exit");

16.// wait for all Setup Threads To Exit

17.waitThreadsStopped();

18.log.info("All Setup Threads have ended");

19.groupCount = 0;

20.JMeterContextService.clearTotalThreads();

21. }

进行一次 gc 后开始跑真正的测试，即启动所有的 ThreadGroup，这里会检查 serialized 属性，用来判断是否这些 ThreadGroup 串行执行

1. JMeterUtils.helpGC();

等待所有的ThreadGroup结束

1. while (running & & iter.hasNext()) {// for each thread group

2.AbstractThreadGroup group = iter.next();

3.// ignore Setup and Post here. We could have filtered the searcher.

4.// but then

5.// future Thread Group objects wouldn\'t execute.

6.if (group instanceof SetupThreadGroup || group instanceof PostThreadGroup) {

7.continue;

8.}

9.groupCount++;

10.String groupName = group.getName();

11.log.info("Starting ThreadGroup: " + groupCount + " : " + groupName);

12.startThreadGroup(group, groupCount, searcher, testLevelElements, notifier);

13.if (serialized & & iter.hasNext()) {

14.log.info("Waiting for thread group: " + groupName + " to finish before starting next group");

15.group.waitThreadsStopped();

16.}

17. } // end of thread groups

18. if (groupCount == 0) { // No TGs found

19.log.info("No enabled thread groups found");

20. } else {

21.if (running) {

22.log.info("All thread groups have been started");

23.} else {

24.log.info("Test stopped - no more thread groups will be started");

25.}

26. }

27.

28. // wait for all Test Threads To Exit

29. waitThreadsStopped();

若有 PostThreadGroup（一般没有），执行所有的 PostThreadGroup 并等待至所有 PostThreadGroup 结束

1. if (postIter.hasNext()) {

2.groupCount = 0;

3.JMeterContextService.clearTotalThreads();

4.log.info("Starting tearDown thread groups");

5.if (mainGroups & & !running) { // i.e. shutdown/stopped during main

6.// thread groups

7.running = shutdown & & tearDownOnShutdown; // re-enable for

8.// tearDown if

9.// necessary

10.}

11.while (running & & postIter.hasNext()) {// for each setup thread

12.// group

13.AbstractThreadGroup group = postIter.next();

14.groupCount++;

15.String groupName = group.getName();

16.log.info("Starting tearDown ThreadGroup: " + groupCount + " : " + groupName);

17.startThreadGroup(group, groupCount, postSearcher, testLevelElements, notifier);

18.if (serialized & & postIter.hasNext()) {

19.log.info("Waiting for post thread group: " + groupName

20.+ " to finish before starting next post group");

21.group.waitThreadsStopped();

22.}

23.}

24.waitThreadsStopped(); // wait for Post threads to stop

25. }

触发第三步中解析的 testListener 的 testEnded 方法：JavaSampler 会调用真正跑的 AbstractJavaSamplerClient 的 teardownTest 方法，可以打印该 JavaSamplerClient 测试总共花费的时间；

ResultCollector 用来将测试结果写如文件生成；
reportTestPlan 用来关闭文件。

1. notifyTestListenersOfEnd(testListeners);

2. JMeterContextService.endTest();

startThreadGroup
启动线程组,run 方法中调用

1. private void startThreadGroup(AbstractThreadGroup group, int groupCount, SearchByClass< ?> searcher,

2.List< ?> testLevelElements, ListenerNotifier notifier) {

3.try {

4.int numThreads = group.getNumThreads();

5.JMeterContextService.addTotalThreads(numThreads);

6.boolean onErrorStopTest = group.getOnErrorStopTest();

7.boolean onErrorStopTestNow = group.getOnErrorStopTestNow();

8.boolean onErrorStopThread = group.getOnErrorStopThread();

9.boolean onErrorStartNextLoop = group.getOnErrorStartNextLoop();

10.String groupName = group.getName();

11.log.info("Starting " + numThreads + " threads for group " + groupName + ".");

12.

13.if (onErrorStopTest) {

14.log.info("Test will stop on error");

15.} else if (onErrorStopTestNow) {

16.log.info("Test will stop abruptly on error");

17.} else if (onErrorStopThread) {

18.log.info("Thread will stop on error");

19.} else if (onErrorStartNextLoop) {

20.log.info("Thread will start next loop on error");

21.} else {

22.log.info("Thread will continue on error");

23.}

24.ListedHashTree threadGroupTree = (ListedHashTree) searcher.getSubTree(group);

25.threadGroupTree.add(group, testLevelElements);

26.

27.groups.add(group);

28.group.start(groupCount, notifier, threadGroupTree, this);

29.} catch (JMeterStopTestException ex) { // NOSONAR Reported by log

30.JMeterUtils.reportErrorToUser("Error occurred starting thread group :" + group.getName()

31.+ ", error message:" + ex.getMessage() + ", \\r\\nsee log file for more details", ex);

32.return; // no point continuing

33.}

34. }

waitThreadsStopped
等待线程停止，run 方法中调用

1. /**

2.* Wait for Group Threads to stop

3.*/

4. private void waitThreadsStopped() {

5.// ConcurrentHashMap does not need synch. here

6.for (AbstractThreadGroup threadGroup : groups) {

7.threadGroup.waitThreadsStopped();

8.}

9. }

10.

11. /**

12.* Wait for all Group Threads to stop

13.*/

14. @Override

15. public void waitThreadsStopped() {

16.if (delayedStartup) {

17.waitThreadStopped(threadStarter);

18.}

19.for (Thread t : allThreads.values()) {

20.waitThreadStopped(t);

21.}

22. }

23.

24. /**

25.* Wait for thread to stop

26.* @param thread Thread

27.*/

28. private void waitThreadStopped(Thread thread) {

29.if (thread != null) {

30.while (thread.isAlive()) {

31.try {

32.thread.join(WAIT_TO_DIE);

33.} catch (InterruptedException e) {

34.Thread.currentThread().interrupt();

35.}

36.}

37.}

38. }

removeThreadGroups
移除线程组，在 run 方法里调用

1.private void removeThreadGroups(List< ?> elements) {

2.Iterator< ?> iter = elements.iterator();

3.while (iter.hasNext()) { // Can\'t use for loop here because we remove elements

4.Object item = iter.next();

5.if (item instanceof AbstractThreadGroup || !(item instanceof TestElement)) {

6.iter.remove();

7.}

8.}

9.}

runTest
runTest( )，调用该方法用来执行测试，启动一个线程并触发它的run()方法，若报异常则调用stopTest()，抛出 JMeterEngineException。

1.// 调用该方法用来执行测试,启动一个线程并触发它的run()方法，若报异常则调用stopTest()，抛出JMeterEngineException

2.@Override

3.public void runTest() throws JMeterEngineException {

4.if (host != null){

5.long now=System.currentTimeMillis();

6.System.out.println("Starting the test on host " + host + " @ "+new Date(now)+" ("+now+")"); // NOSONAR Intentional

7.}

8.try {

9.Thread runningThread = new Thread(this, "StandardJMeterEngine");

10.// 启动一个线程并触发它的run()方法

11.runningThread.start();

12.} catch (Exception err) {

13.stopTest();

14.throw new JMeterEngineException(err);

15.}

16.}

stopThread
根据 threadName 停止线程的执行：分两种情况立即停止和非立即停止，根据第二个参数的值决定

1.//根据threadName停止线程的执行：分两种情况立即停止和非立即停止，根据第二个参数的值决定

2.public static boolean stopThread(String threadName) {

3.return stopThread(threadName, false);

4.}

5.

6.public static boolean stopThreadNow(String threadName) {

7.return stopThread(threadName, true);

8.}

9.

10.private static boolean stopThread(String threadName, boolean now) {

11.if (engine == null) {

12.return false; // e.g. not yet started

13.}

14.boolean wasStopped = false;

15.// ConcurrentHashMap does not need synch. here

16.for (AbstractThreadGroup threadGroup : engine.groups) {

17.wasStopped = wasStopped || threadGroup.stopThread(threadName, now);

18.}

19.return wasStopped;

20.}

ThreadGroup.stopThread调用及具体实现代码如下:

1./**

2.* Stop thread called threadName:

3.* < ol>

4.*< li> stop JMeter thread< /li>

5.*< li> interrupt JMeter thread< /li>

6.*< li> interrupt underlying thread< /li>

7.* < /ol>

8.* @param threadName String thread name

9.* @param now boolean for stop

10.* @return true if thread stopped

11.*/

12.@Override

13.public boolean stopThread(String threadName, boolean now) {

14.for (Entry< JMeterThread, Thread> threadEntry : allThreads.entrySet()) {

15.JMeterThread jMeterThread = threadEntry.getKey();

16.if (jMeterThread.getThreadName().equals(threadName)) {

17.stopThread(jMeterThread, threadEntry.getValue(), now);

18.return true;

19.}

20.}

21.return false;

22.}

23.

24./**

25.* Hard Stop JMeterThread thrd and interrupt JVM Thread if interrupt is true

26.* @param jmeterThread {@link JMeterThread}

27.* @param jvmThread {@link Thread}

28.* @param interrupt Interrupt thread or not

29.*/

30.private void stopThread(JMeterThread jmeterThread, Thread jvmThread, boolean interrupt) {

31.jmeterThread.stop();

32.jmeterThread.interrupt(); // interrupt sampler if possible

33.if (interrupt & & jvmThread != null) { // Bug 49734

34.jvmThread.interrupt(); // also interrupt JVM thread

35.}

36.}

stopTest
stopTest(boolean now)测试，若 now 为 true 则停止动作立即执行；若为 false 则停止动作缓刑，它会等待当前正在执行的测试至少执行完一个 iteration。

1.// 停止测试，若now为true则停止动作立即执行；若为false则停止动作缓刑，它会等待当前正在执行的测试至少执行完一个iteration。

2.@Override

3.public synchronized void stopTest(boolean now) {

4.Thread stopThread = new Thread(new StopTest(now));

5.stopThread.start();

6.}

stopTest()立即停止执行测试

1. /**

2.* Stop Test Now

3.*/

4.@Override

5.public synchronized void stopTest() {

6.stopTest(true);

7.}

notifyTestListenersOfStart
测试开始通知监听

1.private void notifyTestListenersOfStart(SearchByClass< TestStateListener> testListeners) {

2.for (TestStateListener tl : testListeners.getSearchResults()) {

3.if (tl instanceof TestBean) {

4.TestBeanHelper.prepare((TestElement) tl);

5.}

6.if (host == null) {

7.tl.testStarted();

8.} else {

9.tl.testStarted(host);

10.}

11.}

12.}

介绍本方法需要了解下 TestStateListener 接口

1. package org.apache.jmeter.testelement;

2.

3. /**

4.* @since 2.8

5.*/

6. public interface TestStateListener {

7.

8./**

9.* 

10.* Called just before the start of the test from the main engine thread.

11.*

12.* This is before the test elements are cloned.

13.*

14.* Note that not all the test

15.* variables will have been set up at this point.

16.* 

17.*

18.* 

19.* 

20.* N.B. testStarted() and testEnded() are called from different threads.

21.* 

22.* 

23.* @see org.apache.jmeter.engine.StandardJMeterEngine#run()

24.*

25.*/

26.void testStarted();

27.

28./**

29.* 

30.* Called just before the start of the test from the main engine thread.

31.*

32.* This is before the test elements are cloned.

33.*

34.* Note that not all the test

35.* variables will have been set up at this point.

36.* 

37.*

38.* 

39.* 

40.* N.B. testStarted() and testEnded() are called from different threads.

41.* 

42.* 

43.* @see org.apache.jmeter.engine.StandardJMeterEngine#run()

44.* @param host name of host

45.*/

46.void testStarted(String host);

47.

48./**

49.* 

50.* Called once for all threads after the end of a test.

51.*

52.* This will use the same element instances as at the start of the test.

53.* 

54.*

55.* 

56.* 

57.* N.B. testStarted() and testEnded() are called from different threads.

58.* 

59.* 

60.* @see org.apache.jmeter.engine.StandardJMeterEngine#stopTest()

61.*

62.*/

63.void testEnded();

64.

65./**

66.* 

67.* Called once for all threads after the end of a test.

68.*

69.* This will use the same element instances as at the start of the test.

70.* 

71.*

72.* 

73.* 

74.* N.B. testStarted() and testEnded() are called from different threads.

75.* 

76.* 

77.* @see org.apache.jmeter.engine.StandardJMeterEngine#stopTest()

78.* @param host name of host

79.*

80.*/

81.

82.void testEnded(String host);

83.

84. }

testStarted:在测试开始之前调用
testEnded:在所有线程测试结束时调用一次

notifyTestListenersOfEnd
测试结束通知监听

1. private void notifyTestListenersOfEnd(SearchByClass< TestStateListener> testListeners) {

2.log.info("Notifying test listeners of end of test");

3.for (TestStateListener tl : testListeners.getSearchResults()) {

4.try {

5.if (host == null) {

6.tl.testEnded();

7.} else {

8.tl.testEnded(host);

9.}

10.} catch (Exception e) {

11.log.warn("Error encountered during shutdown of "+tl.toString(),e);

12.}

13.}

14.if (host != null) {

15.log.info("Test has ended on host {} ", host);

16.long now=System.currentTimeMillis();

17.System.out.println("Finished the test on host " + host + " @ "+new Date(now)+" ("+now+")" // NOSONAR Intentional

18.+(EXIT_AFTER_TEST ? " - exit requested." : ""));

19.if (EXIT_AFTER_TEST){

20.exit();

21.}

22.}

23.active=false;

24.}

单机执行

1. // 加载jmx文件

2. FileServer.getFileServer().setBaseForScript(jmxFile);

3. // 设置jmx脚本文件的工作目录

4. HashTree jmxTree = SaveService.loadTree(jmxFile);

5. // 去掉没用的节点元素，替换掉可以替换的控制器

6. JMeter.convertSubTree(jmxTree);

7.

8. // 初始化默认的压测引擎

9. JMeterEngine engine = new StandardJMeterEngine();

10. engine.configure(jmxTree);

11. engine.runTest();

分布式执行

1. // 分布式执行脚本，StringTokenizer是为了初始化hosts参数

2. // DistributedRunner本质上还是StandardJMeterEngine来执行的压测，使用的是rmi的协议实现的分布式压测。

3. java.util.StringTokenizer st = new java.util.StringTokenizer(remoteHostsString, ","); //$NON-NLS-1$

4. List< String> hosts = new LinkedList< > ();

5. while (st.hasMoreElements()) {

6.hosts.add((String) st.nextElement());

7. }

8.

9. DistributedRunner distributedRunner=new DistributedRunner(this.remoteProps);

10. distributedRunner.setStdout(System.out); // NOSONAR

11. distributedRunner.setStdErr(System.err); // NOSONAR

12. distributedRunner.init(hosts, clonedTree);

13. engines.addAll(distributedRunner.getEngines());

14. distributedRunner.start();