部署运维与容错韧性

在 Vert.x 中集成使用 Resilience4j 容错库

本文档将向你展示如何在 Vert.x 响应式应用中使用 Resilience4j 容错保护库。

Resilience4j 是一个备受推崇的轻量级容错框架，实现了常见的系统韧性设计模式：

  • 隔板舱（Bulkhead，并发限流隔离）

  • 断路器（Circuit Breaker，故障快速熔断）

  • 速率限制器（Rate Limiter，流量整形）

  • 重试器（Retry，自动重试失败操作）

  • 超时限制器（Time Limiter，调用超时保护）

在本实战指南中，我们将重点演示断路器（Circuit Breaker）的使用，但本指南的代码仓库中同时包含了上述所有模式针对 Vert.x 异步模型的完整适配器。

本示例所采用的 Resilience4j 2.0 要求运行在 Java 17 或更高版本。

你将构建什么

你将结合 HTTP 客户端使用断路器，以防止不断向频繁返回错误的服务端盲目发起请求。 这主要实现两个核心目标：

  • 防止系统雪崩：避免对本已处于过载崩溃边缘的服务端施加更多无谓负载；

  • 快速失败（Fail Fast）：在下游大概率会失败的情况下，立即给调用方返回降级或错误响应，不再白白浪费连接与等待时间。

该应用主要包含以下核心类：

  1. CircuitBreakerVerticle：演示服务调用的 Verticle；

  2. VertxCircuitBreaker：将 Resilience4j 断路器无缝包装为 Vert.x Future 异步语义的适配器。

前置要求

  • 文本编辑器或 IDE

  • Java 17 或更高版本

  • Maven

创建工程项目

这是你应该使用的 pom.xml 文件配置：

核心适配器实现

由于 Resilience4j 原生针对同步阻塞或标准 CompletableFuture 设计，我们编写了一个优雅的适配器将其接入 Vert.x 的 Future 异步体系：

package io.vertx.howtos.resilience4j;

import io.github.resilience4j.circuitbreaker.CallNotPermittedException;
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import io.vertx.core.internal.ContextInternal;

import java.util.function.Supplier;

public class VertxCircuitBreaker {
  public static <T> Future<T> executeFuture(CircuitBreaker circuitBreaker, Supplier<Future<T>> supplier) {
    ContextInternal ctx = ContextInternal.current();
    Promise<T> promise = ctx != null ? ctx.promise() : Promise.promise();

    if (!circuitBreaker.tryAcquirePermission()) {
      promise.fail(CallNotPermittedException.createCallNotPermittedException(circuitBreaker));
    } else {
      long start = circuitBreaker.getCurrentTimestamp();
      try {
        supplier.get().onComplete(result -> {
          long duration = circuitBreaker.getCurrentTimestamp() - start;
          if (result.failed()) {
            circuitBreaker.onError(duration, circuitBreaker.getTimestampUnit(), result.cause());
            promise.fail(result.cause());
          } else {
            circuitBreaker.onResult(duration, circuitBreaker.getTimestampUnit(), result.result());
            promise.complete(result.result());
          }
        });
      } catch (Exception exception) {
        long duration = circuitBreaker.getCurrentTimestamp() - start;
        circuitBreaker.onError(duration, circuitBreaker.getTimestampUnit(), exception);
        promise.fail(exception);
      }
    }

    return promise.future();
  }

  public static <T> Supplier<Future<T>> decorateFuture(CircuitBreaker circuitBreaker, Supplier<Future<T>> supplier) {
    return () -> executeFuture(circuitBreaker, supplier);
  }
}

Verticle 逻辑实现

在 CircuitBreakerVerticle 中，我们配置断路器规则并将其包裹在 Web Client 请求外部：

package io.vertx.howtos.resilience4j;

import io.github.resilience4j.circuitbreaker.CircuitBreaker;
import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig;
import io.vertx.core.Future;
import io.vertx.core.VerticleBase;
import io.vertx.core.Vertx;
import io.vertx.core.http.HttpResponseExpectation;
import io.vertx.ext.web.Router;
import io.vertx.ext.web.client.WebClient;
import io.vertx.ext.web.codec.BodyCodec;

// tag::class[]
public class CircuitBreakerVerticle extends VerticleBase {
// end::class[]
  // tag::main[]
  public static void main(String[] args) {
    Vertx vertx = Vertx.vertx();
    vertx.deployVerticle(new CircuitBreakerVerticle());
  }
  // end::main[]

  // tag::start[]
  @Override
  public Future<?> start() {
    // tag::circuit-breaker[]
    CircuitBreaker cb = CircuitBreaker.of("my-circuit-breaker", CircuitBreakerConfig.custom()
      .minimumNumberOfCalls(5)
      .build());
    // end::circuit-breaker[]

    // tag::router[]
    Router router = Router.router(vertx);
    WebClient client = WebClient.create(vertx);

    router.get("/").handler(ctx -> {
      VertxCircuitBreaker.executeFuture(cb, () -> {
        return client.get(8080, "localhost", "/does-not-exist")
          .as(BodyCodec.string())
          .send()
          .expecting(HttpResponseExpectation.SC_SUCCESS);
      })
        .onSuccess(response -> ctx.end("Got: " + response.body() + "\n"))
        .onFailure(error -> ctx.end("Failed with: " + error.toString() + "\n"));
    });
    // end::router[]

    // tag::server[]
    return vertx.createHttpServer()
      .requestHandler(router)
      .listen(8080)
      .onSuccess(server -> {
        System.out.println("HTTP server started on port " + server.actualPort());
      });
    // end::server[]
  }
  // end::start[]
// tag::class[]
}
// end::class[]

运行与观察熔断状态

执行构建与启动命令：

./mvnw clean compile exec:java

当目标服务连续抛出异常达到设定的错误阈值时，断路器将瞬间切换为 OPEN（打开状态），后续请求被毫秒级快速阻断；经过等待冷却时间后切换为 HALF_OPEN 进行试探，成功后恢复为 CLOSED。

总结

本文档展示了：

  • 如何将 Resilience4j 核心容错能力适配至 Vert.x 异步 Future 编程模型；

  • 配置断路器实现下游故障感知与快速失败保护。

延伸阅读