概述 JMeter 默认单机压测引擎，运行 JMeter 测试，直接用于本地 GUI 和非 GUI 调用，或者RemoteJMeterEngineImpl 在服务器模式下运行时启动。 API地址：https://jmeter.apache.org/api/org/apache/jmeter/engi
概述
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. // 灵魂级变量,注意关键字volatile2. 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&E2. 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）
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();启动所有 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 group2. 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 用来关闭文件。
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()，抛出JMeterEngineException2. @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. * <p>
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. * </p>
17. *
18. * <p>
19. * <b>
20. * N.B. testStarted() and testEnded() are called from different threads.
21. * </b>
22. * </p>
23. * @see org.apache.jmeter.engine.StandardJMeterEngine#run()
24. *
25. */
26. void testStarted();
27.
28. /**
29. * <p>
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. * </p>
37. *
38. * <p>
39. * <b>
40. * N.B. testStarted() and testEnded() are called from different threads.
41. * </b>
42. * </p>
43. * @see org.apache.jmeter.engine.StandardJMeterEngine#run()
44. * @param host name of host
45. */
46. void testStarted(String host);
47.
48. /**
49. * <p>
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. * </p>
54. *
55. * <p>
56. * <b>
57. * N.B. testStarted() and testEnded() are called from different threads.
58. * </b>
59. * </p>
60. * @see org.apache.jmeter.engine.StandardJMeterEngine#stopTest()
61. *
62. */
63. void testEnded();
64.
65. /**
66. * <p>
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. * </p>
71. *
72. * <p>
73. * <b>
74. * N.B. testStarted() and testEnded() are called from different threads.
75. * </b>
76. * </p>
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();
StringTokenizer 是为了初始化hosts参数使用的。 DistributedRunner 本质上还是 StandardJMeterEngine 来执行的压测，使用的是 RMI 的协议实现的分布式压测。