Java并发压测案例如何实操

wen java案例 2026-06-27 220

本文目录导读：

Java并发压测案例如何实操

环境准备
基础压测框架
核心压测实现
高级压测特性
实际压测案例
结果分析工具
最佳实践建议

我来详细讲解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());
    }
}