本文目录导读：
我来详细讲解Java并发压测的实操案例,从环境准备到具体实现。
环境准备
Maven依赖
<dependencies>
<!-- HTTP客户端 -->
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
<version>4.5.13</version>
</dependency>
<!-- 计数工具 -->
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-math3</artifactId>
<version>3.6.1</version>
</dependency>
<!-- JSON处理 -->
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.83</version>
</dependency>
<!-- 日志 -->
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>1.2.11</version>
</dependency>
</dependencies>
基础压测框架
1 压测参数配置
@Data
@Builder
public class StressTestConfig {
private int concurrentLevel; // 并发级别
private int totalRequests; // 总请求数
private int duration; // 压测持续时间（秒）
private String targetUrl; // 目标URL
private String httpMethod; // HTTP方法
private Map<String, String> headers; // 请求头
private String requestBody; // 请求体
}
2 压测结果收集
@Data
@Builder
public class StressTestResult {
private int totalRequests;
private int successCount;
private int failCount;
private long totalTime; // 总耗时(ms)
private double avgResponseTime; // 平均响应时间
private double minResponseTime; // 最小响应时间
private double maxResponseTime; // 最大响应时间
private double throughput; // 吞吐量(请求/秒)
private Map<Integer, Integer> statusCodeDistribution; // 状态码分布
private List<Long> responseTimes; // 所有响应时间
}
核心压测实现
1 压测执行器
@Component
@Slf4j
public class StressTestExecutor {
private final HttpClient httpClient;
private final ExecutorService executorService;
public StressTestExecutor() {
// 配置连接池
PoolingHttpClientConnectionManager cm = new PoolingHttpClientConnectionManager();
cm.setMaxTotal(200);
cm.setDefaultMaxPerRoute(200);
this.httpClient = HttpClients.custom()
.setConnectionManager(cm)
.setConnectionTimeToLive(30, TimeUnit.SECONDS)
.evictIdleConnections(60, TimeUnit.SECONDS)
.build();
this.executorService = Executors.newCachedThreadPool();
}
public StressTestResult execute(StressTestConfig config) throws InterruptedException {
// 结果收集
List<CompletableFuture<SingleRequestResult>> futures = new ArrayList<>();
AtomicInteger successCount = new AtomicInteger(0);
AtomicInteger failCount = new AtomicInteger(0);
List<Long> responseTimes = Collections.synchronizedList(new ArrayList<>());
Map<Integer, Integer> statusDistribution = new ConcurrentHashMap<>();
long startTime = System.currentTimeMillis();
// 创建信号量控制并发
Semaphore semaphore = new Semaphore(config.getConcurrentLevel());
// 确定总请求数
int totalRequests = config.getTotalRequests() > 0
? config.getTotalRequests()
: Integer.MAX_VALUE;
// 开始压测
CountDownLatch latch = new CountDownLatch(totalRequests);
for (int i = 0; i < totalRequests; i++) {
final int requestId = i;
CompletableFuture<SingleRequestResult> future = CompletableFuture.supplyAsync(() -> {
try {
semaphore.acquire();
long requestStart = System.currentTimeMillis();
// 执行请求
SingleRequestResult result = executeRequest(config, requestId);
long duration = System.currentTimeMillis() - requestStart;
result.setDuration(duration);
// 统计
if (result.isSuccess()) {
successCount.incrementAndGet();
} else {
failCount.incrementAndGet();
}
responseTimes.add(duration);
statusDistribution.merge(result.getStatusCode(), 1, Integer::sum);
return result;
} catch (Exception e) {
log.error("Request failed: {}", e.getMessage());
failCount.incrementAndGet();
return SingleRequestResult.builder()
.success(false)
.errorMessage(e.getMessage())
.build();
} finally {
semaphore.release();
latch.countDown();
}
}, executorService);
futures.add(future);
}
// 等待所有请求完成
latch.await();
long totalTime = System.currentTimeMillis() - startTime;
// 计算结果
return calculateResult(config, successCount.get(), failCount.get(),
responseTimes, totalTime, statusDistribution);
}
private StressTestResult calculateResult(StressTestConfig config, int success,
int fail, List<Long> responseTimes,
long totalTime,
Map<Integer, Integer> statusDistribution) {
DoubleSummaryStatistics stats = responseTimes.stream()
.mapToDouble(Long::doubleValue)
.summaryStatistics();
return StressTestResult.builder()
.totalRequests(config.getTotalRequests())
.successCount(success)
.failCount(fail)
.totalTime(totalTime)
.avgResponseTime(stats.getAverage())
.minResponseTime(stats.getMin())
.maxResponseTime(stats.getMax())
.throughput((double) (success + fail) / (totalTime / 1000.0))
.statusCodeDistribution(statusDistribution)
.responseTimes(new ArrayList<>(responseTimes))
.build();
}
}
2 单次请求执行
@Slf4j
@Component
public class RequestExecutor {
private final CloseableHttpClient httpClient;
public SingleRequestResult executeRequest(StressTestConfig config, int requestId) {
HttpUriRequest request = buildRequest(config, requestId);
try (CloseableHttpResponse response = httpClient.execute(request)) {
int statusCode = response.getStatusCode();
String responseBody = EntityUtils.toString(response.getEntity());
return SingleRequestResult.builder()
.requestId(requestId)
.success(statusCode >= 200 && statusCode < 500)
.statusCode(statusCode)
.responseBody(responseBody)
.build();
} catch (Exception e) {
log.error("Request execution failed: {}", e.getMessage());
return SingleRequestResult.builder()
.requestId(requestId)
.success(false)
.errorMessage(e.getMessage())
.build();
}
}
private HttpUriRequest buildRequest(StressTestConfig config, int requestId) {
HttpRequestBase request;
switch (config.getHttpMethod().toUpperCase()) {
case "POST":
HttpPost post = new HttpPost(config.getTargetUrl());
if (config.getRequestBody() != null) {
post.setEntity(new StringEntity(config.getRequestBody(), ContentType.APPLICATION_JSON));
}
request = post;
break;
case "PUT":
HttpPut put = new HttpPut(config.getTargetUrl());
if (config.getRequestBody() != null) {
put.setEntity(new StringEntity(config.getRequestBody(), ContentType.APPLICATION_JSON));
}
request = put;
break;
default:
request = new HttpGet(config.getTargetUrl());
}
// 设置请求头
if (config.getHeaders() != null) {
config.getHeaders().forEach(request::setHeader);
}
// 添加请求ID标记
request.setHeader("X-Request-Id", String.valueOf(requestId));
return request;
}
}
高级压测特性
1 阶梯式并发
@Slf4j
public class GradualLoadTest {
public void executeGradualTest(StressTestConfig baseConfig) {
int[] concurrentLevels = {10, 50, 100, 200, 500};
int requestsPerLevel = 1000;
for (int concurrent : concurrentLevels) {
log.info("Starting load test with {} concurrent users", concurrent);
StressTestConfig config = StressTestConfig.builder()
.concurrentLevel(concurrent)
.totalRequests(requestsPerLevel)
.targetUrl(baseConfig.getTargetUrl())
.httpMethod(baseConfig.getHttpMethod())
.headers(baseConfig.getHeaders())
.build();
try {
StressTestResult result = execute(config);
logResult(concurrent, result);
// 预热后的短暂休息
Thread.sleep(2000);
} catch (Exception e) {
log.error("Load test failed at level {}: {}", concurrent, e.getMessage());
}
}
}
private void logResult(int concurrent, StressTestResult result) {
log.info("""
\n=== 并发级别: {} ===
总请求数: {}
成功数: {}
失败数: {}
平均响应时间: {} ms
吞吐量: {} req/s
========================
""",
concurrent,
result.getTotalRequests(),
result.getSuccessCount(),
result.getFailCount(),
String.format("%.2f", result.getAvgResponseTime()),
String.format("%.2f", result.getThroughput())
);
}
}
2 持续压测模式
@Slf4j
public class DurationBasedTest {
private volatile boolean running = true;
public StressTestResult executeForDuration(StressTestConfig config) {
List<Long> responseTimes = Collections.synchronizedList(new ArrayList<>());
AtomicInteger successCount = new AtomicInteger(0);
AtomicInteger failCount = new AtomicInteger(0);
long startTime = System.currentTimeMillis();
long endTime = startTime + (config.getDuration() * 1000L);
// 启动多个线程模拟并发
List<Thread> workers = new ArrayList<>();
for (int i = 0; i < config.getConcurrentLevel(); i++) {
Thread worker = new Thread(() -> {
while (running && System.currentTimeMillis() < endTime) {
long requestStart = System.currentTimeMillis();
try {
SingleRequestResult result = executeRequest(config, 0);
long duration = System.currentTimeMillis() - requestStart;
responseTimes.add(duration);
if (result.isSuccess()) {
successCount.incrementAndGet();
} else {
failCount.incrementAndGet();
}
} catch (Exception e) {
failCount.incrementAndGet();
}
}
});
workers.add(worker);
worker.start();
}
// 等待指定时间
try {
Thread.sleep(config.getDuration() * 1000L);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
running = false;
// 等待所有线程完成
workers.forEach(worker -> {
try {
worker.join(5000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
long totalTime = System.currentTimeMillis() - startTime;
return calculateResult(config, successCount.get(), failCount.get(),
responseTimes, totalTime, new ConcurrentHashMap<>());
}
}
实际压测案例
1 测试REST API
@SpringBootTest
@Slf4j
public class ApiStressTest {
@Autowired
private StressTestExecutor stressTestExecutor;
@Test
public void testCreateUserApi() throws InterruptedException {
StressTestConfig config = StressTestConfig.builder()
.concurrentLevel(100)
.totalRequests(5000)
.targetUrl("http://localhost:8080/api/users")
.httpMethod("POST")
.headers(Map.of(
"Content-Type", "application/json",
"Authorization", "Bearer test-token"
))
.requestBody("""
{
"username": "test_user",
"email": "test@example.com",
"age": 25
}
""")
.build();
StressTestResult result = stressTestExecutor.execute(config);
// 断言
assertThat(result.getFailCount()).isLessThan(100);
assertThat(result.getAvgResponseTime()).isLessThan(500);
assertThat(result.getThroughput()).isGreaterThan(100);
// 生成报告
generateReport(result);
}
@Test
public void testGetUsersApi() throws InterruptedException {
StressTestConfig config = StressTestConfig.builder()
.concurrentLevel(200)
.totalRequests(10000)
.targetUrl("http://localhost:8080/api/users?page=1&size=20")
.httpMethod("GET")
.headers(Map.of(
"Accept", "application/json"
))
.build();
StressTestResult result = stressTestExecutor.execute(config);
// 分析响应时间分布
analyzeResponseTimeDistribution(result);
}
}
2 测试数据库操作
@Slf4j
public class DatabaseStressTest {
@Autowired
private DataSource dataSource;
@Test
public void testConcurrentDatabaseReads() throws InterruptedException {
int concurrentUsers = 50;
int queriesPerUser = 100;
ExecutorService executor = Executors.newFixedThreadPool(concurrentUsers);
CountDownLatch latch = new CountDownLatch(concurrentUsers * queriesPerUser);
AtomicLong totalTime = new AtomicLong(0);
AtomicInteger errors = new AtomicInteger(0);
long startTime = System.currentTimeMillis();
for (int i = 0; i < concurrentUsers; i++) {
executor.submit(() -> {
for (int j = 0; j < queriesPerUser; j++) {
long queryStart = System.currentTimeMillis();
try (Connection conn = dataSource.getConnection();
PreparedStatement stmt = conn.prepareStatement("SELECT * FROM users WHERE id = ?")) {
stmt.setInt(1, ThreadLocalRandom.current().nextInt(1, 10000));
ResultSet rs = stmt.executeQuery();
long duration = System.currentTimeMillis() - queryStart;
totalTime.addAndGet(duration);
} catch (SQLException e) {
errors.incrementAndGet();
} finally {
latch.countDown();
}
}
});
}
latch.await();
long totalDuration = System.currentTimeMillis() - startTime;
log.info("""
\n=== 数据库并发压测结果 ===
并发用户数: {}
总查询数: {}
错误数: {}
总耗时: {} ms
平均查询时间: {} ms
QPS: {}
""",
concurrentUsers,
concurrentUsers * queriesPerUser,
errors.get(),
totalDuration,
String.format("%.2f", totalTime.get() / (double) (concurrentUsers * queriesPerUser)),
String.format("%.2f", (concurrentUsers * queriesPerUser) / (totalDuration / 1000.0))
);
}
}
结果分析工具
1 响应时间分布分析
@Component
@Slf4j
public class ResponseTimeAnalyzer {
public void analyzeDistribution(List<Long> responseTimes) {
// 排序
List<Long> sorted = new ArrayList<>(responseTimes);
Collections.sort(sorted);
// 计算百分位
Map<Integer, Long> percentiles = new TreeMap<>();
percentiles.put(50, percentile(sorted, 50));
percentiles.put(75, percentile(sorted, 75));
percentiles.put(90, percentile(sorted, 90));
percentiles.put(95, percentile(sorted, 95));
percentiles.put(99, percentile(sorted, 99));
log.info("响应时间分布:");
percentiles.forEach((p, time) ->
log.info("P{} = {} ms", p, time));
// 识别峰值
identifyPeaks(responseTimes);
}
private long percentile(List<Long> sorted, double percentile) {
int index = (int) Math.ceil(percentile / 100.0 * sorted.size());
return sorted.get(Math.min(index, sorted.size() - 1));
}
private void identifyPeaks(List<Long> responseTimes) {
double mean = responseTimes.stream()
.mapToLong(Long::longValue)
.average()
.orElse(0);
double stdDev = Math.sqrt(responseTimes.stream()
.mapToDouble(t -> Math.pow(t - mean, 2))
.average()
.orElse(0));
long threshold = (long) (mean + 3 * stdDev);
long peakCount = responseTimes.stream()
.filter(t -> t > threshold)
.count();
log.info("响应时间标准差: {} ms", String.format("%.2f", stdDev));
log.info("异常响应阈值: {} ms", threshold);
log.info("异常响应数量: {}", peakCount);
}
}
2 性能报告生成
@Component
@Slf4j
public class PerformanceReportGenerator {
public void generateReport(StressTestResult result) {
StringBuilder report = new StringBuilder();
report.append("\n========================================\n");
report.append(" 性能压测报告\n");
report.append("========================================\n");
report.append(String.format("测试时间: %s\n", LocalDateTime.now()));
report.append(String.format("总请求数: %d\n", result.getTotalRequests()));
report.append(String.format("成功请求: %d\n", result.getSuccessCount()));
report.append(String.format("失败请求: %d\n", result.getFailCount()));
report.append(String.format("成功率: %.2f%%\n",
(double) result.getSuccessCount() / result.getTotalRequests() * 100));
report.append(String.format("总耗时: %d ms\n", result.getTotalTime()));
report.append("----------------------------------------\n");
report.append("响应时间:\n");
report.append(String.format(" 平均: %.2f ms\n", result.getAvgResponseTime()));
report.append(String.format(" 最小: %.2f ms\n", result.getMinResponseTime()));
report.append(String.format(" 最大: %.2f ms\n", result.getMaxResponseTime()));
report.append("----------------------------------------\n");
report.append(String.format("吞吐量: %.2f requests/s\n", result.getThroughput()));
report.append("========================================\n");
log.info(report.toString());
// 保存到文件
saveReportToFile(report.toString());
}
}
最佳实践建议
- 预热阶段：正式压测前先发送几百个请求预热系统
- 监控系统：压测时监控CPU、内存、网络等系统指标
- 渐进式加压：从低并发开始，逐步增加并发数
- 隔离测试环境：在独立环境进行压测，避免影响生产
- 关注瓶颈：分析响应时间分布，查找性能瓶颈
- 多次测试：每种场景至少测试3次，取平均值
这个完整的压测框架可以根据实际需求进行扩展和定制,支持不同场景的性能测试需求。
