3. 实现舱壁隔离限制故障影响范围
任务说明
本案例要求实现舱壁隔离（Bulkhead Isolation）模式，将系统资源（线程池、连接池等）按业务域划分为独立舱室，使单个舱室的故障（如慢调用、异常堆积）不会耗尽共享资源，从而限制故障影响范围。完成目标：构建一个最小可运行的舱壁隔离框架，支持为不同业务域分配独立线程池，并验证当某一舱室发生故障时，其他舱室仍能正常处理请求。
核心实现
// Bulkhead.java - 舱壁隔离核心类
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
public class Bulkhead {
private final String name;
private final ThreadPoolExecutor executor;
private final Semaphore semaphore;
private final int maxConcurrentCalls;
private final AtomicInteger rejectedCount = new AtomicInteger(0);
public Bulkhead(String name, int coreThreads, int maxThreads, int queueCapacity, int maxConcurrentCalls) {
this.name = name;
this.maxConcurrentCalls = maxConcurrentCalls;
this.semaphore = new Semaphore(maxConcurrentCalls);
this.executor = new ThreadPoolExecutor(
coreThreads, maxThreads, 30L, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(queueCapacity),
new ThreadFactory() {
private final AtomicInteger counter = new AtomicInteger(1);
@Override
public Thread newThread(Runnable r) {
return new Thread(r, name + "-thread-" + counter.getAndIncrement());
}
},
new ThreadPoolExecutor.AbortPolicy()
);
}
public <T> Future<T> submit(Callable<T> task) throws RejectedExecutionException {
return executor.submit(() -> {
if (!semaphore.tryAcquire()) {
rejectedCount.incrementAndGet();
throw new BulkheadFullException(name + " 舱室并发数已达上限");
}
try {
return task.call();
} finally {
semaphore.release();
}
});
}
public int getRejectedCount() { return rejectedCount.get(); }
public int getActiveCount() { return executor.getActiveCount(); }
public String getName() { return name; }
public void shutdown() { executor.shutdown(); }
public static class BulkheadFullException extends RuntimeException {
public BulkheadFullException(String msg) { super(msg); }
}
}// BulkheadDemo.java - 演示故障隔离效果
import java.util.concurrent.*;
public class BulkheadDemo {
public static void main(String[] args) throws Exception {
// 创建两个独立舱室：支付舱室（故障）和订单舱室（正常）
Bulkhead paymentBulkhead = new Bulkhead("payment", 2, 2, 5, 2);
Bulkhead orderBulkhead = new Bulkhead("order", 2, 2, 5, 2);
// 模拟支付舱室故障：每个任务都阻塞5秒（模拟第三方支付接口超时）
for (int i = 0; i < 3; i++) {
final int taskId = i;
paymentBulkhead.submit(() -> {
System.out.println("[" + Thread.currentThread().getName() + "] 支付任务 " + taskId + " 开始（模拟慢调用）");
Thread.sleep(5000);
return "支付结果-" + taskId;
});
}
// 立即向订单舱室提交任务，验证其不受支付舱室故障影响
for (int i = 0; i < 3; i++) {
final int taskId = i;
orderBulkhead.submit(() -> {
System.out.println("[" + Thread.currentThread().getName() + "] 订单任务 " + taskId + " 执行成功");
return "订单结果-" + taskId;
});
}
// 等待2秒后检查两个舱室状态
Thread.sleep(2000);
System.out.println("\n=== 2秒后舱室状态 ===");
System.out.println("支付舱室活动线程数: " + paymentBulkhead.getActiveCount());
System.out.println("订单舱室活动线程数: " + orderBulkhead.getActiveCount());
// 再向支付舱室提交任务，验证其因并发数超限被拒绝
try {
paymentBulkhead.submit(() -> "额外支付任务");
} catch (RejectedExecutionException e) {
System.out.println("支付舱室拒绝新任务: " + e.getMessage());
}
// 验证订单舱室仍可继续处理
orderBulkhead.submit(() -> {
System.out.println("[" + Thread.currentThread().getName() + "] 订单舱室仍可接受新任务");
return "ok";
});
Thread.sleep(1000);
System.out.println("\n=== 最终状态 ===");
System.out.println("支付舱室拒绝任务数: " + paymentBulkhead.getRejectedCount());
System.out.println("订单舱室拒绝任务数: " + orderBulkhead.getRejectedCount());
paymentBulkhead.shutdown();
orderBulkhead.shutdown();
}
}运行与验证
编译运行：
javac Bulkhead.java BulkheadDemo.java
java BulkheadDemo预期输出（关键观测点）：
[payment-thread-1] 支付任务 0 开始（模拟慢调用）
[payment-thread-2] 支付任务 1 开始（模拟慢调用）
[order-thread-1] 订单任务 0 执行成功
[order-thread-2] 订单任务 1 执行成功
[order-thread-1] 订单任务 2 执行成功
=== 2秒后舱室状态 ===
支付舱室活动线程数: 2
订单舱室活动线程数: 0
支付舱室拒绝新任务: 支付舱室并发数已达上限
[order-thread-1] 订单舱室仍可接受新任务
=== 最终状态 ===
支付舱室拒绝任务数: 1
订单舱室拒绝任务数: 0判定标准：当支付舱室因慢调用占满2个并发许可后，新任务被快速拒绝（而非排队等待）；订单舱室全程无拒绝、无阻塞，任务即时完成。这证明故障被限制在支付舱室内，未扩散至订单舱室。
关键设计说明
本实现采用信号量+独立线程池双重隔离机制：每个舱室拥有专属线程池，从资源层面隔离；信号量控制最大并发调用数，当并发数超限时立即抛出异常而非排队等待，实现快速失败。演示中支付舱室因模拟慢调用占满全部并发许可后，后续任务被立即拒绝，而订单舱室因独立资源不受任何影响。实际应用中可将线程池参数（核心线程数、队列容量）和并发上限按业务SLA独立配置，例如支付舱室分配较少线程防止拖垮整体，订单舱室分配充足线程保障核心链路。