如何在Vert.x WebClient中立即发送请求？（API基准测试场景）

嘿，我看你在尝试用Vert.x WebClient做REST API的基准测试，想要实现持续立即发送请求的效果，但原来的while(true)写法有个致命问题——Vert.x是基于单线程事件循环的，你把无限循环直接放在start方法里，会彻底阻塞事件线程，导致WebClient的请求回调永远没法执行，整个Verticle都没法正常工作。

下面给你一个完全非阻塞的实现方案，既能完成预热+测试的流程，又能保证请求持续立即发送，同时还能统计吞吐量和延迟：

核心思路 #

1. 用Vert.x的setTimer处理预热和测试时长，完全不阻塞事件循环
2. 控制并发请求数，避免一下子发送过多请求压垮客户端或服务器
3. 在每个请求的回调（成功/失败）里立即发起下一个请求，保持持续发送的状态
4. 用线程安全的原子类统计请求数和延迟数据

完整代码实现 #

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Future;
import io.vertx.ext.web.client.WebClient;
import io.vertx.ext.web.client.WebClientOptions;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

public class BenchmarkVerticle extends AbstractVerticle {

    // 可配置的基准测试参数
    private static final long WARMUP_DURATION = 10_000; // 10秒预热
    private static final long TEST_DURATION = 20_000;   // 20秒正式测试
    private static final String API_ENDPOINT = "/your-target-api";
    private static final String API_HOST = "your-api-server.com";
    private static final int API_PORT = 8080;
    private static final int MAX_CONCURRENT_REQUESTS = 100; // 并发请求上限，按需调整

    private WebClient client;
    private long testStartTime;
    private final AtomicInteger totalRequests = new AtomicInteger(0);
    private final AtomicLong totalLatencyMs = new AtomicLong(0);
    private boolean isTestingActive = false;

    @Override
    public void start(Future<Void> startFuture) {
        // 初始化WebClient，配置连接池和基础参数
        client = WebClient.create(vertx, new WebClientOptions()
                .setDefaultHost(API_HOST)
                .setDefaultPort(API_PORT)
                .setMaxPoolSize(MAX_CONCURRENT_REQUESTS) // 连接池大小匹配并发数
                .setKeepAlive(true) // 开启长连接，提升性能
                .setConnectTimeout(5000) // 连接超时时间
                .setResponseTimeout(10000)); // 响应超时时间

        // 启动预热流程
        System.out.println("开始10秒预热...");
        vertx.setTimer(WARMUP_DURATION, warmupTimerId -> {
            System.out.println("预热完成，启动20秒基准测试...");
            isTestingActive = true;
            testStartTime = System.currentTimeMillis();

            // 初始化并发请求：启动N个请求循环，保持持续发送
            for (int i = 0; i < MAX_CONCURRENT_REQUESTS; i++) {
                sendNextRequest();
            }

            // 设置测试结束定时器
            vertx.setTimer(TEST_DURATION, testTimerId -> {
                isTestingActive = false;
                printBenchmarkResults();
                startFuture.complete(); // 通知Verticle启动完成
            });
        });
    }

    private void sendNextRequest() {
        // 如果测试已结束，停止发送请求
        if (!isTestingActive) {
            return;
        }

        long requestStartTime = System.currentTimeMillis();
        // 这里可以根据需要改成POST/PUT等其他HTTP方法，添加请求体
        client.get(API_ENDPOINT)
                .send()
                .onSuccess(response -> {
                    // 统计成功请求的延迟
                    long latency = System.currentTimeMillis() - requestStartTime;
                    totalRequests.incrementAndGet();
                    totalLatencyMs.addAndGet(latency);
                    // 立即发送下一个请求，保持并发
                    sendNextRequest();
                })
                .onFailure(error -> {
                    // 处理请求失败的情况，也可以统计失败数
                    System.err.printf("请求失败: %s%n", error.getMessage());
                    // 即使失败，也要继续发送下一个请求
                    sendNextRequest();
                });
    }

    private void printBenchmarkResults() {
        long totalTestTimeMs = System.currentTimeMillis() - testStartTime;
        int completedRequests = totalRequests.get();
        long avgLatencyMs = completedRequests > 0 ? totalLatencyMs.get() / completedRequests : 0;
        double throughput = completedRequests * 1000.0 / totalTestTimeMs; // 请求/秒

        System.out.println("\n=== 基准测试结果 ===");
        System.out.printf("总请求数: %d%n", completedRequests);
        System.out.printf("测试总耗时: %dms%n", totalTestTimeMs);
        System.out.printf("吞吐量: %.2f 请求/秒%n", throughput);
        System.out.printf("平均延迟: %dms%n", avgLatencyMs);
    }
}

关键细节说明 #

• 非阻塞定时器：用vertx.setTimer替代阻塞的等待逻辑，完全符合Vert.x的事件驱动模型，不会占用事件循环线程
• 并发控制：MAX_CONCURRENT_REQUESTS参数可以控制同时发送的请求数，避免因请求过多导致客户端连接池耗尽或服务器过载
• 持续发送请求：每个请求完成（无论成功或失败）后立即调用sendNextRequest()，保证请求持续发送，实现"立即发送"的需求
• 线程安全统计：用AtomicInteger和AtomicLong统计数据，确保在多事件循环线程（如果Verticle配置了多实例）下的线程安全
• 可扩展性：如果需要发送POST请求，只需要把client.get()改成client.post()，并通过sendJsonObject()或sendBuffer()添加请求体即可

额外建议 #

• 可以根据服务器的性能调整MAX_CONCURRENT_REQUESTS参数，逐步递增找到最优并发数
• 可以添加失败请求数的统计，让测试结果更全面
• 生产环境中可以把配置参数（比如主机、端口、时长）放到Vert.x的配置文件中，避免硬编码

内容的提问来源于stack exchange，提问作者Nguyễn Tuấn