Vert.x学习之Resilience4j原理与用法解读

<dependency>
    <groupId>io.github.resilience4j</groupId>
    <artifactId>resilience4j-vertx</artifactId>
    <version>2.0.0</version>
</dependency>

VertxOptions options = new VertxOptions()
    .setWorkerPoolSize(10); // 专用Worker线程池
Vertx vertx = Vertx.vertx(options);

// 配置断路器
CircuitBreakerConfig config = CircuitBreakerConfig.custom()
    .failureRateThreshold(50) // 失败率阈值50%
    .waitDurationInOpenState(Duration.ofSeconds(5)) // Open状态持续时间
    .permittedNumberOfCallsInHalfOpenState(3) // 半开状态允许的请求数
    .build();

CircuitBreaker circuitBreaker = CircuitBreaker.of("myService", config);

// 在Worker线程中执行服务调用
vertx.executeBlocking(promise -> {
    try {
        String result = circuitBreaker.executeSupplier(() -> {
            // 模拟阻塞操作（如HTTP请求）
            Thread.sleep(1000);
            return "Success";
        });
        promise.complete(result);
    } catch (Exception e) {
        promise.fail(e);
    }
}, false, res -> {
    if (res.succeeded()) {
        System.out.println("Result: " + res.result());
    } else {
        System.out.println("Failed: " + res.cause().getMessage());
    }
});

// 使用Vertx的Future.timeout（底层由Event Loop线程调度，但业务逻辑在Worker线程中）
Future<String> future = vertx.executeBlocking(promise -> {
    try {
        // 模拟长时间操作
        Thread.sleep(2000);
        promise.complete("Done");
    } catch (InterruptedException e) {
        promise.fail(e);
    }
}, false);

// 设置超时（独立线程管理）
future.onComplete(ar -> {
    if (ar.failed() && ar.cause() instanceof TimeoutException) {
        System.out.println("Operation timed out");
    }
});

// 更推荐的方式：使用Resilience4j的TimeLimiter（完全独立线程）
TimeLimiter timeLimiter = TimeLimiter.of(Duration.ofSeconds(1));
CompletableFuture<String> timedFuture = timeLimiter.executeFutureSupplier(() -> 
    CompletableFuture.supplyAsync(() -> {
        // 在Worker线程中执行
        try {
            Thread.sleep(2000);
            return "Done";
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }, vertx.getOrCreateContext().owner().newInstance());
);

timedFuture.whenComplete((result, ex) -> {
    if (ex != null) {
        System.out.println("Timed out or failed: " + ex.getMessage());
    } else {
        System.out.println("Result: " + result);
    }
});

// 限流配置
RateLimiterConfig rateLimiterConfig = RateLimiterConfig.custom()
    .limitForPeriod(5) // 每秒允许的请求数
    .limitRefreshPeriod(Duration.ofSeconds(1))
    .timeoutDuration(Duration.ofMillis(100)) // 获取令牌超时时间
    .build();

RateLimiter rateLimiter = RateLimiter.of("myService", rateLimiterConfig);

// 重试配置
RetryConfig retryConfig = RetryConfig.custom()
    .maxAttempts(3)
    .waitDuration(Duration.ofMillis(500))
    .build();

Retry retry = Retry.of("myService", retryConfig);

// 组合使用（限流 + 重试 + 断路器）
Supplier<String> decoratedSupplier = Retry.decorateSupplier(
    retry,
    RateLimiter.decorateSupplier(
        rateLimiter,
        CircuitBreaker.decorateSupplier(
            circuitBreaker,
            () -> {
                // 在Worker线程中执行
                if (Math.random() > 0.7) {
                    throw new RuntimeException("Random failure");
                }
                return "Success";
            }
        )
    )
);

vertx.executeBlocking(promise -> {
    try {
        String result = decoratedSupplier.get();
        promise.complete(result);
    } catch (Exception e) {
        promise.fail(e);
    }
}, false, res -> {
    if (res.succeeded()) {
        System.out.println("Final result: " + res.result());
    } else {
        System.out.println("Final failure: " + res.cause().getMessage());
    }
});