到目前为止，在本系列中，我们已经了解了 Resilience4j 及其 Retry, RateLimiter 和 TimeLimiter 模块。在本文中，我们将探讨 Bulkhead 模块。我们将了解它解决了什么问题，何时以及如何使用它，并查看一些示例。
代码示例
本文附有 GitHub 上的工作代码示例。
什么是 Resilience4j？
请参阅上一篇文章中的描述，快速了解 Resilience4j 的一般工作原理。
什么是故障隔离？
几年前，我们遇到了一个生产问题，其中一台服务器停止响应健康检查，负载均衡器将服务器从池中取出。
就在我们开始调查这个问题的时候，还有第二个警报——另一台服务器已经停止响应健康检查，也被从池中取出。
几分钟后，每台服务器都停止响应健康探测，我们的服务完全关闭。
我们使用 Redis 为应用程序支持的几个功能缓存一些数据。正如我们后来发现的那样，Redis 集群同时出现了一些问题，它已停止接受新连接。我们使用 Jedis 库连接到 Redis，该库的默认行为是无限期地阻塞调用线程，直到建立连接。
我们的服务托管在 Tomcat 上，它的默认请求处理线程池大小为 200 个线程。因此，通过连接到 Redis 的代码路径的每个请求最终都会无限期地阻塞线程。
几分钟之内，集群中的所有 2000 个线程都无限期地阻塞了——甚至没有空闲线程来响应负载均衡器的健康检查。
该服务本身支持多项功能，并非所有功能都需要访问 Redis 缓存。但是当这一方面出现问题时，它最终影响了整个服务。
这正是故障隔离要解决的问题——它可以防止某个服务区域的问题影响整个服务。
虽然我们的服务发生的事情是一个极端的例子，但我们可以看到缓慢的上游依赖如何影响调用服务的不相关区域。
如果我们在每个服务器实例上对 Redis 设置了 20 个并发请求的限制，那么当 Redis 连接问题发生时，只有这些线程会受到影响。剩余的请求处理线程可以继续为其他请求提供服务。
故障隔离背后的想法是对我们对远程服务进行的并发调用数量设置限制。我们将对不同远程服务的调用视为不同的、隔离的池，并对可以同时进行的调用数量设置限制。
术语舱壁本身来自它在船舶中的使用，其中船舶的底部被分成彼此分开的部分。如果有裂缝，并且水开始流入，则只有该部分会充满水。这可以防止整艘船沉没。
Resilience4j 隔板概念
resilience4j-bulkhead 的工作原理类似于其他 Resilience4j 模块。我们为它提供了我们想要作为函数构造执行的代码——一个进行远程调用的 lambda 表达式或一个从远程服务中检索到的某个值的 Supplier，等等——并且隔板用代码装饰它以控制并发调用数。
Resilience4j 提供两种类型的隔板 – SemaphoreBulkhead 和 ThreadPoolBulkhead。
SemaphoreBulkhead 内部使用
java.util.concurrent.Semaphore 来控制并发调用的数量并在当前线程上执行我们的代码。
ThreadPoolBulkhead 使用线程池中的一个线程来执行我们的代码。它内部使用
java.util.concurrent.ArrayBlockingQueue 和
java.util.concurrent.ThreadPoolExecutor 来控制并发调用的数量。
SemaphoreBulkhead
让我们看看与信号量隔板相关的配置及其含义。
maxConcurrentCalls 确定我们可以对远程服务进行的最大并发调用数。我们可以将此值视为初始化信号量的许可数。
任何尝试超过此限制调用远程服务的线程都可以立即获得 BulkheadFullException 或等待一段时间以等待另一个线程释放许可。这由 maxWaitDuration 值决定。
当有多个线程在等待许可时，fairCallHandlingEnabled 配置确定等待的线程是否以先进先出的顺序获取许可。
最后， writableStackTraceEnabled 配置让我们可以在 BulkheadFullException 发生时减少堆栈跟踪中的信息量。这很有用，因为如果没有它，当异常多次发生时，我们的日志可能会充满许多类似的信息。通常在读取日志时，只知道发生了 BulkheadFullException 就足够了。
ThreadPoolBulkhead
coreThreadPoolSize 、 maxThreadPoolSize 、 keepAliveDuration 和 queueCapacity 是与 ThreadPoolBulkhead 相关的主要配置。ThreadPoolBulkhead 内部使用这些配置来构造一个 ThreadPoolExecutor。
internalThreadPoolExecutor 使用可用的空闲线程之一执行传入的任务。 如果没有线程可以自由执行传入的任务，则该任务将排队等待线程可用时稍后执行。如果已达到 queueCapacity，则远程调用将被拒绝并返回 BulkheadFullException。
ThreadPoolBulkhead 也有 writableStackTraceEnabled 配置来控制 BulkheadFullException 的堆栈跟踪中的信息量。
使用 Resilience4j 隔板模块
让我们看看如何使用 resilience4j-bulkhead 模块中可用的各种功能。
我们将使用与本系列前几篇文章相同的示例。假设我们正在为一家航空公司建立一个网站，以允许其客户搜索和预订航班。我们的服务与 FlightSearchService 类封装的远程服务对话。
SemaphoreBulkhead
使用基于信号量的隔板时，BulkheadRegistry、BulkheadConfig 和 Bulkhead 是我们使用的主要抽象。
BulkheadRegistry 是一个用于创建和管理 Bulkhead 对象的工厂。
BulkheadConfig 封装了 maxConcurrentCalls、maxWaitDuration、writableStackTraceEnabled 和 fairCallHandlingEnabled 配置。每个 Bulkhead 对象都与一个 BulkheadConfig 相关联。
第一步是创建一个 BulkheadConfig：
BulkheadConfig config = BulkheadConfig.ofDefaults();这将创建一个 BulkheadConfig，其默认值为 maxConcurrentCalls(25)、maxWaitDuration(0s)、writableStackTraceEnabled(true) 和 fairCallHandlingEnabled(true)。
假设我们希望将并发调用的数量限制为 2，并且我们愿意等待 2 秒让线程获得许可：
BulkheadConfig config = BulkheadConfig.custom()
.maxConcurrentCalls(2)
.maxWaitDuration(Duration.ofSeconds(2))
.build();然后我们创建一个 Bulkhead：
BulkheadRegistry registry = BulkheadRegistry.of(config);
Bulkhead bulkhead = registry.bulkhead("flightSearchService");现在让我们表达我们的代码以作为 Supplier 运行航班搜索并使用 bulkhead 装饰它：
BulkheadRegistry registry = BulkheadRegistry.of(config);
Bulkhead bulkhead = registry.bulkhead("flightSearchService");最后，让我们调用几次装饰操作来了解隔板的工作原理。我们可以使用 CompletableFuture 来模拟来自用户的并发航班搜索请求：
for (int i=0; i<4; i++) {
CompletableFuture
.supplyAsync(decoratedFlightsSupplier)
.thenAccept(flights -> System.out.println("Received results"));
}输出中的时间戳和线程名称显示，在 4 个并发请求中，前两个请求立即通过：
Searching for flights; current time = 11:42:13 187; current thread = ForkJoinPool.commonPool-worker-3
Searching for flights; current time = 11:42:13 187; current thread = ForkJoinPool.commonPool-worker-5
Flight search successful at 11:42:13 226
Flight search successful at 11:42:13 226
Received results
Received results
Searching for flights; current time = 11:42:14 239; current thread = ForkJoinPool.commonPool-worker-9
Searching for flights; current time = 11:42:14 239; current thread = ForkJoinPool.commonPool-worker-7
Flight search successful at 11:42:14 239
Flight search successful at 11:42:14 239
Received results
Received results第三个和第四个请求仅在 1 秒后就能够获得许可，在之前的请求完成之后。
如果线程无法在我们指定的 2s maxWaitDuration 内获得许可，则会抛出 BulkheadFullException：
Caused by: io.github.resilience4j.bulkhead.BulkheadFullException: Bulkhead 'flightSearchService' is full and does not permit further calls
at io.github.resilience4j.bulkhead.BulkheadFullException.createBulkheadFullException(BulkheadFullException.java:49)
at io.github.resilience4j.bulkhead.internal.SemaphoreBulkhead.acquirePermission(SemaphoreBulkhead.java:164)
at io.github.resilience4j.bulkhead.Bulkhead.lambda$decorateSupplier$5(Bulkhead.java:194)
at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1700)
... 6 more除了第一行，堆栈跟踪中的其他行没有增加太多价值。如果 BulkheadFullException 发生多次，这些堆栈跟踪行将在我们的日志文件中重复。
我们可以通过将 writableStackTraceEnabled 配置设置为 false 来减少堆栈跟踪中生成的信息量：
BulkheadConfig config = BulkheadConfig.custom()
.maxConcurrentCalls(2)
.maxWaitDuration(Duration.ofSeconds(1))
.writableStackTraceEnabled(false)
.build();现在，当 BulkheadFullException 发生时，堆栈跟踪中只存在一行：
Searching for flights; current time = 12:27:58 658; current thread = ForkJoinPool.commonPool-worker-3
Searching for flights; current time = 12:27:58 658; current thread = ForkJoinPool.commonPool-worker-5
io.github.resilience4j.bulkhead.BulkheadFullException: Bulkhead 'flightSearchService' is full and does not permit further calls
Flight search successful at 12:27:58 699
Flight search successful at 12:27:58 699
Received results
Received results与我们见过的其他 Resilience4j 模块类似，Bulkhead 还提供了额外的方法，如 decorateCheckedSupplier()、decorateCompletionStage()、decorateRunnable()、decorateConsumer() 等，因此我们可以在 Supplier 供应商之外的其他结构中提供我们的代码。
ThreadPoolBulkhead
当使用基于线程池的隔板时，
ThreadPoolBulkheadRegistry、ThreadPoolBulkheadConfig 和 ThreadPoolBulkhead 是我们使用的主要抽象。
ThreadPoolBulkheadRegistry 是用于创建和管理 ThreadPoolBulkhead 对象的工厂。
ThreadPoolBulkheadConfig 封装了 coreThreadPoolSize 、 maxThreadPoolSize、 keepAliveDuration 和 queueCapacity 配置。每个 ThreadPoolBulkhead 对象都与一个 ThreadPoolBulkheadConfig 相关联。
第一步是创建一个 ThreadPoolBulkheadConfig：
ThreadPoolBulkheadConfig config =
ThreadPoolBulkheadConfig.ofDefaults();这将创建一个 ThreadPoolBulkheadConfig，其默认值为 coreThreadPoolSize（可用处理器数量 -1）、maxThreadPoolSize（可用处理器最大数量）、keepAliveDuration（20ms）和 queueCapacity（100）。
假设我们要将并发调用的数量限制为 2：
ThreadPoolBulkheadConfig config = ThreadPoolBulkheadConfig.custom()
.maxThreadPoolSize(2)
.coreThreadPoolSize(1)
.queueCapacity(1)
.build();然后我们创建一个 ThreadPoolBulkhead：
ThreadPoolBulkheadRegistry registry = ThreadPoolBulkheadRegistry.of(config);
ThreadPoolBulkhead bulkhead = registry.bulkhead("flightSearchService");现在让我们表达我们的代码以作为 Supplier 运行航班搜索并使用 bulkhead 装饰它：
Supplier<List<Flight>> flightsSupplier =
() -> service.searchFlightsTakingOneSecond(request);
Supplier<CompletionStage<List<Flight>>> decoratedFlightsSupplier =
ThreadPoolBulkhead.decorateSupplier(bulkhead, flightsSupplier);与返回一个 Supplier<List<Flight>> 的
SemaphoreBulkhead.decorateSupplier() 不同，
ThreadPoolBulkhead.decorateSupplier() 返回一个 Supplier<CompletionStage<List<Flight>>。这是因为 ThreadPoolBulkHead 不会在当前线程上同步执行代码。
最后，让我们调用几次装饰操作来了解隔板的工作原理：
for (int i=0; i<3; i++) {
decoratedFlightsSupplier
.get()
.whenComplete((r,t) -> {
if (r != null) {
System.out.println("Received results");
}
if (t != null) {
t.printStackTrace();
}
});
}输出中的时间戳和线程名称显示，虽然前两个请求立即执行，但第三个请求已排队，稍后由释放的线程之一执行：
Searching for flights; current time = 16:15:00 097; current thread = bulkhead-flightSearchService-1
Searching for flights; current time = 16:15:00 097; current thread = bulkhead-flightSearchService-2
Flight search successful at 16:15:00 136
Flight search successful at 16:15:00 135
Received results
Received results
Searching for flights; current time = 16:15:01 151; current thread = bulkhead-flightSearchService-2
Flight search successful at 16:15:01 151
Received results如果队列中没有空闲线程和容量，则抛出 BulkheadFullException：
Exception in thread "main" io.github.resilience4j.bulkhead.BulkheadFullException: Bulkhead 'flightSearchService' is full and does not permit further calls
at io.github.resilience4j.bulkhead.BulkheadFullException.createBulkheadFullException(BulkheadFullException.java:64)
at io.github.resilience4j.bulkhead.internal.FixedThreadPoolBulkhead.submit(FixedThreadPoolBulkhead.java:157)
... other lines omitted ...我们可以使用 writableStackTraceEnabled 配置来减少堆栈跟踪中生成的信息量：
ThreadPoolBulkheadConfig config = ThreadPoolBulkheadConfig.custom()
.maxThreadPoolSize(2)
.coreThreadPoolSize(1)
.queueCapacity(1)
.writableStackTraceEnabled(false)
.build();现在，当 BulkheadFullException 发生时，堆栈跟踪中只存在一行：
Searching for flights; current time = 12:27:58 658; current thread = ForkJoinPool.commonPool-worker-3
Searching for flights; current time = 12:27:58 658; current thread = ForkJoinPool.commonPool-worker-5
io.github.resilience4j.bulkhead.BulkheadFullException: Bulkhead 'flightSearchService' is full and does not permit further calls
Flight search successful at 12:27:58 699
Flight search successful at 12:27:58 699
Received results
Received results上下文传播
有时我们将数据存储在 ThreadLocal 变量中并在代码的不同区域中读取它。我们这样做是为了避免在方法链之间显式地将数据作为参数传递，尤其是当该值与我们正在实现的核心业务逻辑没有直接关系时。
例如，我们可能希望将当前用户 ID 或事务 ID 或某个请求跟踪 ID 记录到每个日志语句中，以便更轻松地搜索日志。对于此类场景，使用 ThreadLocal 是一种有用的技术。
使用 ThreadPoolBulkhead 时，由于我们的代码不在当前线程上执行，因此我们存储在 ThreadLocal 变量中的数据在其他线程中将不可用。
让我们看一个例子来理解这个问题。首先我们定义一个 RequestTrackingIdHolder 类，一个围绕 ThreadLocal 的包装类：
class RequestTrackingIdHolder {
static ThreadLocal<String> threadLocal = new ThreadLocal<>();
static String getRequestTrackingId() {
return threadLocal.get();
}
static void setRequestTrackingId(String id) {
if (threadLocal.get() != null) {
threadLocal.set(null);
threadLocal.remove();
}
threadLocal.set(id);
}
static void clear() {
threadLocal.set(null);
threadLocal.remove();
}
}静态方法可以轻松设置和获取存储在 ThreadLocal 上的值。我们接下来在调用隔板装饰的航班搜索操作之前设置一个请求跟踪 ID：
for (int i=0; i<2; i++) {
String trackingId = UUID.randomUUID().toString();
System.out.println("Setting trackingId " + trackingId + " on parent, main thread before calling flight search");
RequestTrackingIdHolder.setRequestTrackingId(trackingId);
decoratedFlightsSupplier
.get()
.whenComplete((r,t) -> {
// other lines omitted
});
}示例输出显示此值在隔板管理的线程中不可用：
Setting trackingId 98ff99df-466a-47f7-88f7-5e31fc8fcb6b on parent, main thread before calling flight search
Setting trackingId 6b98d73c-a590-4a20-b19d-c85fea783caf on parent, main thread before calling flight search
Searching for flights; current time = 19:53:53 799; current thread = bulkhead-flightSearchService-1; Request Tracking Id = null
Flight search successful at 19:53:53 824
Received results
Searching for flights; current time = 19:53:54 836; current thread = bulkhead-flightSearchService-1; Request Tracking Id = null
Flight search successful at 19:53:54 836
Received results为了解决这个问题，ThreadPoolBulkhead 提供了一个 ContextPropagator。ContextPropagator 是一种用于跨线程边界检索、复制和清理值的抽象。它定义了一个接口，其中包含从当前线程 (retrieve()) 获取值、将其复制到新的执行线程 (copy()) 并最终在执行线程 (clear()) 上进行清理的方法。
让我们实现一个
RequestTrackingIdPropagator：
class RequestTrackingIdPropagator implements ContextPropagator {
@Override
public Supplier<Optional> retrieve() {
System.out.println("Getting request tracking id from thread: " + Thread.currentThread().getName());
return () -> Optional.of(RequestTrackingIdHolder.getRequestTrackingId());
}
@Override
Consumer<Optional> copy() {
return optional -> {
System.out.println("Setting request tracking id " + optional.get() + " on thread: " + Thread.currentThread().getName());
optional.ifPresent(s -> RequestTrackingIdHolder.setRequestTrackingId(s.toString()));
};
}
@Override
Consumer<Optional> clear() {
return optional -> {
System.out.println("Clearing request tracking id on thread: " + Thread.currentThread().getName());
optional.ifPresent(s -> RequestTrackingIdHolder.clear());
};
}
}我们通过在 ThreadPoolBulkheadConfig 上的设置来为 ThreadPoolBulkhead 提供 ContextPropagator：
class RequestTrackingIdPropagator implements ContextPropagator {
@Override
public Supplier<Optional> retrieve() {
System.out.println("Getting request tracking id from thread: " + Thread.currentThread().getName());
return () -> Optional.of(RequestTrackingIdHolder.getRequestTrackingId());
}
@Override
Consumer<Optional> copy() {
return optional -> {
System.out.println("Setting request tracking id " + optional.get() + " on thread: " + Thread.currentThread().getName());
optional.ifPresent(s -> RequestTrackingIdHolder.setRequestTrackingId(s.toString()));
};
}
@Override
Consumer<Optional> clear() {
return optional -> {
System.out.println("Clearing request tracking id on thread: " + Thread.currentThread().getName());
optional.ifPresent(s -> RequestTrackingIdHolder.clear());
};
}
}现在，示例输出显示请求跟踪 ID 在隔板管理的线程中可用：
Setting trackingId 71d44cb8-dab6-4222-8945-e7fd023528ba on parent, main thread before calling flight search
Getting request tracking id from thread: main
Setting trackingId 5f9dd084-f2cb-4a20-804b-038828abc161 on parent, main thread before calling flight search
Getting request tracking id from thread: main
Setting request tracking id 71d44cb8-dab6-4222-8945-e7fd023528ba on thread: bulkhead-flightSearchService-1
Searching for flights; current time = 20:07:56 508; current thread = bulkhead-flightSearchService-1; Request Tracking Id = 71d44cb8-dab6-4222-8945-e7fd023528ba
Flight search successful at 20:07:56 538
Clearing request tracking id on thread: bulkhead-flightSearchService-1
Received results
Setting request tracking id 5f9dd084-f2cb-4a20-804b-038828abc161 on thread: bulkhead-flightSearchService-1
Searching for flights; current time = 20:07:57 542; current thread = bulkhead-flightSearchService-1; Request Tracking Id = 5f9dd084-f2cb-4a20-804b-038828abc161
Flight search successful at 20:07:57 542
Clearing request tracking id on thread: bulkhead-flightSearchService-1
Received resultsBulkhead事件
Bulkhead 和 ThreadPoolBulkhead 都有一个 EventPublisher 来生成以下类型的事件：
- BulkheadOnCallPermittedEvent
- BulkheadOnCallRejectedEvent 和
- BulkheadOnCallFinishedEvent
我们可以监听这些事件并记录它们，例如：
Bulkhead bulkhead = registry.bulkhead("flightSearchService");
bulkhead.getEventPublisher().onCallPermitted(e -> System.out.println(e.toString()));
bulkhead.getEventPublisher().onCallFinished(e -> System.out.println(e.toString()));
bulkhead.getEventPublisher().onCallRejected(e -> System.out.println(e.toString()));示例输出显示了记录的内容：
2020-08-26T12:27:39.790435: Bulkhead 'flightSearch' permitted a call.
... other lines omitted ...
2020-08-26T12:27:40.290987: Bulkhead 'flightSearch' rejected a call.
... other lines omitted ...
2020-08-26T12:27:41.094866: Bulkhead 'flightSearch' has finished a call.Bulkhead 指标
SemaphoreBulkhead
Bulkhead 暴露了两个指标：
- 可用权限的最大数量（
resilience4j.bulkhead.max.allowed.concurrent.calls），和 - 允许的并发调用数（
resilience4j.bulkhead.available.concurrent.calls）。
bulkhead.available 指标与我们在 BulkheadConfig 上配置的 maxConcurrentCalls 相同。
首先，我们像前面一样创建 BulkheadConfig、BulkheadRegistry 和 Bulkhead。然后，我们创建一个 MeterRegistry 并将 BulkheadRegistry 绑定到它：
MeterRegistry meterRegistry = new SimpleMeterRegistry();
TaggedBulkheadMetrics.ofBulkheadRegistry(registry)
.bindTo(meterRegistry);运行几次隔板装饰操作后，我们显示捕获的指标：
Consumer<Meter> meterConsumer = meter -> {
String desc = meter.getId().getDescription();
String metricName = meter.getId().getName();
Double metricValue = StreamSupport.stream(meter.measure().spliterator(), false)
.filter(m -> m.getStatistic().name().equals("VALUE"))
.findFirst()
.map(m -> m.getValue())
.orElse(0.0);
System.out.println(desc + " - " + metricName + ": " + metricValue);};meterRegistry.forEachMeter(meterConsumer);这是一些示例输出：
The maximum number of available permissions - resilience4j.bulkhead.max.allowed.concurrent.calls: 8.0
The number of available permissions - resilience4j.bulkhead.available.concurrent.calls: 3.0ThreadPoolBulkhead
ThreadPoolBulkhead 暴露五个指标：
- 队列的当前长度（
resilience4j.bulkhead.queue.depth）， - 当前线程池的大小（
resilience4j.bulkhead.thread.pool.size）， - 线程池的核心和最大容量（
resilience4j.bulkhead.core.thread.pool.size和resilience4j.bulkhead.max.thread.pool.size），以及 - 队列的容量（
resilience4j.bulkhead.queue.capacity）。
首先，我们像前面一样创建 ThreadPoolBulkheadConfig、
ThreadPoolBulkheadRegistry 和 ThreadPoolBulkhead。然后，我们创建一个 MeterRegistry 并将
ThreadPoolBulkheadRegistry 绑定到它：
MeterRegistry meterRegistry = new SimpleMeterRegistry();
TaggedThreadPoolBulkheadMetrics.ofThreadPoolBulkheadRegistry(registry).bindTo(meterRegistry);运行几次隔板装饰操作后，我们将显示捕获的指标：
The queue capacity - resilience4j.bulkhead.queue.capacity: 5.0
The queue depth - resilience4j.bulkhead.queue.depth: 1.0
The thread pool size - resilience4j.bulkhead.thread.pool.size: 5.0
The maximum thread pool size - resilience4j.bulkhead.max.thread.pool.size: 5.0
The core thread pool size - resilience4j.bulkhead.core.thread.pool.size: 3.0在实际应用中，我们会定期将数据导出到监控系统并在仪表板上进行分析。
实施隔板时的陷阱和良好实践
使隔板成为单例
对给定远程服务的所有调用都应通过同一个 Bulkhead 实例。对于给定的远程服务，Bulkhead 必须是单例。
如果我们不强制执行此操作，我们代码库的某些区域可能会绕过 Bulkhead 直接调用远程服务。为了防止这种情况，远程服务的实际调用应该在一个核心、内部层和其他区域应该使用内部层暴露的隔板装饰器。
我们如何确保未来的新开发人员理解这一意图？ 查看 Tom 的文章，该文章展示了解决此类问题的一种方法，即通过组织包结构来明确此类意图。此外，它还展示了如何通过在 ArchUnit 测试中编码意图来强制执行此操作。
与其他 Resilience4j 模块结合
将隔板与一个或多个其他 Resilience4j 模块（如重试和速率限制器）结合使用会更有效。例如，如果有 BulkheadFullException，我们可能希望在一些延迟后重试。
结论
在本文中，我们学习了如何使用 Resilience4j 的 Bulkhead 模块对我们对远程服务进行的并发调用设置限制。我们了解了为什么这很重要，还看到了一些有关如何配置它的实际示例。
您可以使用 GitHub 上的代码演示一个完整的应用程序。
本文译自： Implementing Bulkhead with Resilience4j – Reflectoring