壓力測試神器stresstester源碼分析

stresstester-1.0.jar是早期淘寶的一個壓力測試工具，很方便開發人員進行本地代碼的壓力測試，其他專門壓力測試工具也有很多，如:jmeter  loadrunner 等等，本篇文章主要講一下stresstester的源碼設計

先來張類圖

從上圖可以看出StressTestUtils是和開發者打交道的最主要的類；下面來個例子

package com.taobao.stresstester;

import com.taobao.stresstester.core.StressTask;
import java.io.PrintStream;

public class Example
{
public static void main(String[] args)
{
StressTestUtils.testAndPrint(100, 1000, new StressTask()
{
public Object doTask()
throws Exception
{
System.out.println(「Do my task.」);
return null;
}
});
}
}

下面我來介紹主要原理吧：

1.開啓多線程去跑目標代碼（每個線程訪問的次數=總請求次數/併發數）

public Stre***esult test(int concurrencyLevel, int totalRequests, StressTask stressTask, int warmUpTime)
{
if (stressTask == null) {
stressTask = this.emptyTestService;
}
warmUp(warmUpTime, stressTask);
int everyThreadCount = totalRequests / concurrencyLevel;
CyclicBarrier threadStartBarrier = new CyclicBarrier(concurrencyLevel);
CountDownLatch threadEndLatch = new CountDownLatch(concurrencyLevel);
AtomicInteger failedCounter = new AtomicInteger();

StressContext stressContext = new StressContext();
stressContext.setTestService(stressTask);
stressContext.setEveryThreadCount(everyThreadCount);
stressContext.setThreadStartBarrier(threadStartBarrier);
stressContext.setThreadEndLatch(threadEndLatch);
stressContext.setFailedCounter(failedCounter);

ExecutorService executorService = Executors.newFixedThreadPool(concurrencyLevel);

List<StressThreadWorker> workers = new ArrayList(concurrencyLevel);
for (int i = 0; i < concurrencyLevel; i++)
{
StressThreadWorker worker = new StressThreadWorker(stressContext, everyThreadCount);
workers.add(worker);
}
for (int i = 0; i < concurrencyLevel; i++)
{
StressThreadWorker worker = (StressThreadWorker)workers.get(i);
executorService.submit(worker);
}
try
{
threadEndLatch.await();
}
catch (InterruptedException e)
{
log.error(「InterruptedException」, e);
}
executorService.shutdownNow();
int realTotalRequests = everyThreadCount * concurrencyLevel;
int failedRequests = failedCounter.get();
Stre***esult stre***esult = new Stre***esult();

SortResult sortResult = getSortedTimes(workers);
List<Long> allTimes = sortResult.allTimes;

stre***esult.setAllTimes(allTimes);
List<Long> trheadTimes = sortResult.trheadTimes;
long totalTime = ((Long)trheadTimes.get(trheadTimes.size() – 1)).longValue();

stre***esult.setTestsTakenTime(totalTime);
stre***esult.setFailedRequests(failedRequests);
stre***esult.setTotalRequests(realTotalRequests);
stre***esult.setConcurrencyLevel(concurrencyLevel);
stre***esult.setWorkers(workers);

return stre***esult;
}

爲什麼要用2個同步輔助類CyclicBarrier，CountDownLatch

//設置線程集合點，等所有現存啓動完畢在一起請求任務

CyclicBarrier threadStartBarrier = new CyclicBarrier(concurrencyLevel);

//控制所有線程做完任務後狀態，數量爲0的時間，所有任務執行完畢

CountDownLatch threadEndLatch = new CountDownLatch(concurrencyLevel);

2.記錄每次調用代碼的時間，放到一個集合中

SortResult sortResult = getSortedTimes(workers);
List<Long> allTimes = sortResult.allTimes;

3.計算打印出tps /平均耗時/最短耗時/最長耗時

TPS=併發數/平均的相應時間

平均響應時間=測試總時間/總請求次數

package com.taobao.stresstester.core;

import java.io.IOException;
import java.io.Writer;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class SimpleResultFormater
implements Stre***esultFormater
{
protected static Logger log = LoggerFactory.getLogger(SimpleResultFormater.class);

public void format(Stre***esult stre***esult, Writer writer)
{
long testsTakenTime = stre***esult.getTestsTakenTime();
int totalRequests = stre***esult.getTotalRequests();
int concurrencyLevel = stre***esult.getConcurrencyLevel();

float takes = StatisticsUtils.toMs(testsTakenTime);

List<Long> allTimes = stre***esult.getAllTimes();
long totaleTimes = StatisticsUtils.getTotal(allTimes);

float tps = 1.0E+009F * (concurrencyLevel * (totalRequests / (float)totaleTimes));

float averageTime = StatisticsUtils.getAverage(totaleTimes,
totalRequests);

float onTheadAverageTime = averageTime / concurrencyLevel;

int count_50 = totalRequests / 2;
int count_66 = totalRequests * 66 / 100;
int count_75 = totalRequests * 75 / 100;
int count_80 = totalRequests * 80 / 100;
int count_90 = totalRequests * 90 / 100;
int count_95 = totalRequests * 95 / 100;
int count_98 = totalRequests * 98 / 100;
int count_99 = totalRequests * 99 / 100;

long longestRequest = ((Long)allTimes.get(allTimes.size() – 1)).longValue();
long shortestRequest = ((Long)allTimes.get(0)).longValue();

StringBuilder view = new StringBuilder();
view.append(」 Concurrency Level:\t」).append(concurrencyLevel)
.append(「–併發數」);
view.append(「\r\n Time taken for tests:\t」).append(takes).append(」 ms」)
.append(「–測試耗時」);
view.append(「\r\n Complete Requests:\t」).append(totalRequests)
.append(「–完成測試次數」);
view.append(「\r\n Failed Requests:\t」)
.append(stre***esult.getFailedRequests()).append(「–失敗次數」);
view.append(「\r\n Requests per second:\t」).append(tps).append(「–QPS」);
view.append(「\r\n Time per request:\t」)
.append(StatisticsUtils.toMs(averageTime)).append(」 ms」)
.append(「–平均耗時」);
view.append(「\r\n Time per request:\t」)
.append(StatisticsUtils.toMs(onTheadAverageTime))
.append(」 ms (across all concurrent requests)」)
.append(「–平均耗時，忽略併發影響」);
view.append(「\r\n Shortest request:\t」)
.append(StatisticsUtils.toMs(shortestRequest)).append(」 ms」)
.append(「–最短耗時」);

StringBuilder certainTimeView = view; certainTimeView .append(「\r\n Percentage of the requests served within a certain time (ms)」); certainTimeView.append(「\r\n 50%\t」) .append(StatisticsUtils.toMs(((Long)allTimes.get(count_50)).longValue())) .append(「–50% 的耗時在0.005703毫秒以下」); certainTimeView.append(「\r\n 66%\t」).append( StatisticsUtils.toMs(((Long)allTimes.get(count_66)).longValue())); certainTimeView.append(「\r\n 75%\t」).append( StatisticsUtils.toMs(((Long)allTimes.get(count_75)).longValue())); certainTimeView.append(「\r\n 80%\t」).append( StatisticsUtils.toMs(((Long)allTimes.get(count_80)).longValue())); certainTimeView.append(「\r\n 90%\t」).append( StatisticsUtils.toMs(((Long)allTimes.get(count_90)).longValue())); certainTimeView.append(「\r\n 95%\t」).append( StatisticsUtils.toMs(((Long)allTimes.get(count_95)).longValue())); certainTimeView.append(「\r\n 98%\t」).append( StatisticsUtils.toMs(((Long)allTimes.get(count_98)).longValue())); certainTimeView.append(「\r\n 99%\t」).append( StatisticsUtils.toMs(((Long)allTimes.get(count_99)).longValue())); certainTimeView.append(「\r\n 100%\t」) .append(StatisticsUtils.toMs(longestRequest)) .append(」 (longest request)」).append(「–最長的耗時」); try { writer.write(view.toString()); } catch (IOException e) { log.error(「IOException:」, e); } } }