提升高并发系统鲁棒性三件套 - Resilience4j、MDC、Semaphore

kinglyjn      2023-11-15     阅读次

提升高并发系统鲁棒性三件套 - Resilience4j、MDC、Semaphore

1. 1. 限流与熔断
   1. 1.1. Resilience4j 简介
   2. 1.2. 典型的案例
   3. 1.3. 集成到 spring
2. 2. 链路追踪
   1. 2.1. MDC 简介
   2. 2.2. 具体的实现
3. 3. 有限资源的治理

在实际环境中，光有高并发是不够的，系统的稳定性才是王道。高性能后端系统在上线前必须具备完备的防御工事。这其中包括限流与熔断、链路追踪、资源治理等等。本文我们将介绍这些 “防御工事” 的实现。

限流与熔断

Resilience4j 简介

Resilience4j 是 Java 8+ 的轻量级容错（Fault Tolerance）库，专为微服务/高并发场景设计，用来给 “不稳定调用”（远程 API、DB、LLM、第三方服务）加弹性保护，是 Netflix Hystrix 停止维护后的官方推荐替代。

• 轻量无重依赖：核心只依赖 Vavr/SLF4J，不绑 Archaius/Netty，Spring Boot 3 直接集成；Resilience4j 2.x 要求 Java 17。
• 装饰器模式：把容错逻辑”包”在业务函数外，不侵入代码。
• 模块化：熔断/限流/隔离各是独立 jar，只用需要的，不引入全家桶。
• Registry 管理：每个实例独立配置，支持 Micrometer/Prometheus 埋点。

典型的案例

使用 Resilience4j 对 “异步调用” 链进行增强。目标如下：

一个实际的场景是：比如业务系统获取薪资任务调用的是一个外部的薪资中心 API。这个 API 如果每分钟只允许你调 100 次（超额会报错），需要如何在你的虚拟线程代码中加入一个限流器 (RateLimiter)来防止触发 API 的封禁。为了实现这一点，我们需要给它加三层 “防弹衣”：

• RateLimiter（限流器）： 限制请求频率（例如：每秒最多 50 次）。
• CircuitBreaker（断路器）：当失败率超过 50% 时，打开断路器，直接切断请求 10 秒（快速失败，保护后端）。
• Retry（重试）：对于网络抖动导致的异常，自动重试 3 次，间隔 500ms。
• 舱壁模式 (Bulkhead)：为核心业务（用户、订单）和边缘业务（推荐）分配独立的限流窗口。

首先，引入依赖，以及设置必要的配置：

<dependencies>
    <dependency>
        <groupId>io.github.resilience4j</groupId>
        <artifactId>resilience4j-all</artifactId>
        <version>2.2.0</version>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>1.18.44</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
        <version>2.0.18</version>
    </dependency>
    <dependency>
        <groupId>ch.qos.logback</groupId>
        <artifactId>logback-classic</artifactId>
        <version>1.5.34</version>
    </dependency>
</dependencies>

resources/logback.xml：

<?xml version="1.0" encoding="UTF-8"?>
<configuration>

    <!-- 定义日志输出格式 -->
    <property name="LOG_PATTERN" value="%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n" />

    <!-- 控制台输出 -->
    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>${LOG_PATTERN}</pattern>
        </encoder>
    </appender>

    <!-- 设置全局日志级别为 INFO -->
    <root level="INFO">
        <appender-ref ref="CONSOLE" />
    </root>
</configuration>

业务类和测试方法：

import io.github.resilience4j.bulkhead.Bulkhead;
import io.github.resilience4j.bulkhead.BulkheadConfig;
import io.github.resilience4j.circuitbreaker.CallNotPermittedException;
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig;
import io.github.resilience4j.decorators.Decorators;
import io.github.resilience4j.ratelimiter.RateLimiter;
import io.github.resilience4j.ratelimiter.RateLimiterConfig;
import io.github.resilience4j.retry.Retry;
import io.github.resilience4j.retry.RetryConfig;
import lombok.extern.slf4j.Slf4j;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;

@Slf4j
public class SalaryService {

    // 1. 配置限流：每秒最多 50 次请求
    private final RateLimiter rateLimiter = RateLimiter.of("salaryService", RateLimiterConfig.custom()
                                                                                             .limitForPeriod(50)
                                                                                             .limitRefreshPeriod(Duration.ofSeconds(1))
                                                                                             .timeoutDuration(Duration.ofMillis(100)) // 获取令牌超时时间
                                                                                             .build());
	   
    /**
     * 2. 配置断路器：失败率达 50% 时开启
     *    CircuitBreaker 的三态：
     *    - CLOSED：正常调用，滑动窗口统计失败率
     *    - OPEN：失败率超阈值（默认50%）→ 直接快速失败，不真调用，等 waitDurationInOpenState
     *    - HALF_OPEN：冷却后放少量探测请求，成功则回 CLOSED，失败则回 OPEN
     */
    private final CircuitBreaker circuitBreaker = CircuitBreaker.of("salaryService", CircuitBreakerConfig.custom()
                                                                                                         .failureRateThreshold(50)
                                                                                                         .waitDurationInOpenState(Duration.ofSeconds(10)) // 开启后 10 秒尝试恢复
                                                                                                         .slidingWindowSize(10) // 最近 10 次请求进行采样
                                                                                                         .build());
  
    // 3. 配置重试：遇到异常重试 3 次，间隔 500ms
    private final Retry retry = Retry.of("salaryService", RetryConfig.custom()
                                                                     .maxAttempts(3)
                                                                     .waitDuration(Duration.ofMillis(500))
                                                                     .build());

    /**
     * 4. 定义舱壁：为薪资查询（假设它是边缘业务）分配一个只能容纳 100 个并发执行的“舱室”
     *    如果没有舱壁：大量请求在等待薪资 API 返回，这会消耗大量的系统资源（即使是虚拟线程，也需要消耗内存和连接池），最终导致整个 Owlias 系统响应变慢。
     *    有了舱壁：你设置了 maxConcurrentCalls(10)。第 11 个请求进来时，会被直接拒绝并降级。结果就是：薪资服务挂了，但你的订单服务、用户服务依然能流畅运行。
     */
    private final Bulkhead salaryBulkhead = Bulkhead.of("salaryService", BulkheadConfig.custom()
                                                                                       .maxConcurrentCalls(100) // 哪怕有 1000 个线程来，也只允许 100 个同时执行
                                                                                       .maxWaitDuration(Duration.ofMillis(50)) // 超过 50ms 没拿到坑位的直接丢弃
                                                                                       .build());

    public Double callSalaryApi(Long empId) {
        Supplier<Double> supplier = () -> fetchSalary(empId); // 👈🏻

        // 使用 Decorators 链式编排
        return Decorators.ofSupplier(supplier)
                         .withBulkhead(salaryBulkhead) // 挂载舱壁是容量控制，它管的是正在处理的任务总数是多少，它通过空间维度保护系统。
                         .withRetry(retry)
                         .withCircuitBreaker(circuitBreaker)
                         .withRateLimiter(rateLimiter) // 限流是频率控制，它管的是一秒钟允许进多少人，它通过时间维度保护系统。
                         .withFallback(List.of(CallNotPermittedException.class, Exception.class),
                                 ex -> {
                                     System.err.println("API 降级处理: " + ex.getMessage());
                                     return 0.0; // 发生任何异常或被限流/断路时返回默认值
                                 })
                         .get();
    }


    /**
     * 被装饰的 service api
     */
    public Double fetchSalary(Long empId) {
        try {
            // 改成 100 毫秒，这样在 5 秒的等待时间内它能执行完
            TimeUnit.MILLISECONDS.sleep(100);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        // 模拟业务逻辑
        log.info("execute fetchSalary...");
        return 50000.0;
    }


    public static void main(String[] args) {
        SalaryService salaryService = new SalaryService();
        // 使用虚拟线程模拟高并发（Java 21+）
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            AtomicInteger successCount = new AtomicInteger(0);
            AtomicInteger fallbackCount = new AtomicInteger(0);

            // 发起 60 个并发请求，限流器配置是每秒 50 个
            int totalRequests = 60;
            for (int i = 0; i < totalRequests; i++) {
                executor.submit(() -> {
                    Double result = salaryService.callSalaryApi(1L);
                    if (result == 0.0) {
                        fallbackCount.incrementAndGet();
                    } else {
                        successCount.incrementAndGet();
                    }
                });
            }

            // 等待一段时间让任务执行
            executor.shutdown();
            executor.awaitTermination(5, TimeUnit.SECONDS);

            // 由于配置了每秒限流 50 次，60 次请求里至少有 10 次应该是触发了降级或异常
            System.out.println("成功次数: " + successCount.get());
            System.out.println("降级次数: " + fallbackCount.get());
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}

控制台输出：

14:04:53.623 [virtual-40] INFO  demo01.SalaryService - execute fetchSalary...
14:04:53.623 [virtual-34] INFO  demo01.SalaryService - execute fetchSalary...
14:04:53.628 [virtual-41] INFO  demo01.SalaryService - execute fetchSalary...
...
14:04:53.639 [virtual-81] INFO  demo01.SalaryService - execute fetchSalary...
14:04:53.641 [virtual-28] INFO  demo01.SalaryService - execute fetchSalary...
成功次数: 50
降级次数: 10
API 降级处理: RateLimiter 'salaryService' does not permit further calls
API 降级处理: RateLimiter 'salaryService' does not permit further calls
API 降级处理: RateLimiter 'salaryService' does not permit further calls
API 降级处理: RateLimiter 'salaryService' does not permit further calls
API 降级处理: RateLimiter 'salaryService' does not permit further calls
API 降级处理: RateLimiter 'salaryService' does not permit further calls
API 降级处理: RateLimiter 'salaryService' does not permit further calls
API 降级处理: RateLimiter 'salaryService' does not permit further calls
API 降级处理: RateLimiter 'salaryService' does not permit further calls
API 降级处理: RateLimiter 'salaryService' does not permit further calls

集成到 spring

依赖配置变成：

<dependencies>
    <!--spring-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
        <version>3.5.16</version>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-aop</artifactId>
        <version>3.5.16</version>
    </dependency><!--关键：AOP 必须手动导入-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <version>3.5.16</version>
        <scope>test</scope>
    </dependency>

    <!--resilience4j-->
    <dependency>
        <groupId>io.github.resilience4j</groupId>
        <artifactId>resilience4j-spring-boot3</artifactId>
        <version>2.2.0</version>
    </dependency>

    <!--others-->
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
        <version>2.0.18</version>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>1.18.44</version>
    </dependency>
</dependencies>

application.yml

resilience4j:
  circuitbreaker:
    instances:
      salaryService:
        failureRateThreshold: 50
        waitDurationInOpenState: 10s
        slidingWindowSize: 10
  ratelimiter:
    instances:
      salaryService:
        limitForPeriod: 50
        limitRefreshPeriod: 1s
        timeoutDuration: 100ms
  bulkhead:
    instances:
      salaryService:
        maxConcurrentCalls: 10
        maxWaitDuration: 50ms

业务类：

import io.github.resilience4j.bulkhead.annotation.Bulkhead;
import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
import io.github.resilience4j.ratelimiter.annotation.RateLimiter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
public class SalaryService {

    /**
     * 被装饰的 service api
     */
    @CircuitBreaker(name = "salaryService", fallbackMethod = "fallback")
    @RateLimiter(name = "salaryService")
    @Bulkhead(name = "salaryService")
    public Double fetchSalary(Long empId) {
        try {
            // 改成 100 毫秒，这样在 5 秒的等待时间内它能执行完
            TimeUnit.MILLISECONDS.sleep(100);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        // 模拟业务逻辑
        log.info("execute fetchSalary...");
        return 50000.0;
    }

    // 降级方法必须在同一个类中，且参数匹配
    public Double fallback(Long empId, Throwable t) {
        log.error("API 降级处理: {}", t.getMessage());
        return 0.0;
    }
}

测试类：

@SpringBootTest(classes = App.class)
public class SalaryServiceTest {

    @Resource
    private SalaryService salaryService;

    @Test
    public void test01() {
        // 使用虚拟线程模拟高并发（Java 21+）
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            AtomicInteger successCount = new AtomicInteger(0);
            AtomicInteger fallbackCount = new AtomicInteger(0);

            // 发起 60 个并发请求，限流器配置是每秒 50 个
            int totalRequests = 60;
            for (int i = 0; i < totalRequests; i++) {
                executor.submit(() -> {
                    Double result = salaryService.fetchSalary(1L);
                    if (result == 0.0) {
                        fallbackCount.incrementAndGet();
                    } else {
                        successCount.incrementAndGet();
                    }
                });
            }

            // 等待一段时间让任务执行
            executor.shutdown();
            executor.awaitTermination(5, TimeUnit.SECONDS);

            // 由于配置了每秒限流 50 次，60 次请求里至少有 10 次应该是触发了降级或异常
            System.out.println("成功次数: " + successCount.get());
            System.out.println("降级次数: " + fallbackCount.get());
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}

链路追踪

MDC 简介

在异步多线程环境中，最痛苦的是 “日志散落在各处，找不到请求的起点”。接下来我们引入 MDC (Mapped Diagnostic Context)。目标是在 CompletableFuture 或 VirtualThread 任务启动时，将 TraceID 从主线程传递到异步线程。确保无论任务怎么切换，日志里都能带上同一个 TraceID。这是排查线上异步 Bug 的唯一方案。

MDC 是 SLF4J (Simple Logging Facade for Java) 的一部分，而 SLF4J 又是所有 Java 项目几乎必装的日志门面。只要你的项目中有 slf4j-api 这个包，你就已经拥有了 org.slf4j.MDC 类。虽然 MDC 本身是 slf4j-api 的一部分，但要让它真正生效并显示在日志中，你需要一个具体的日志实现框架，比如 Logback 或者 Logback（spring 自带）。

<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-api</artifactId>
    <version>2.0.18</version>
</dependency>
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <version>1.18.44</version>
</dependency>

具体的实现

首先，日志配置需要显式指定 traceId：

<?xml version="1.0" encoding="UTF-8"?>
<configuration>

    <!-- 定义日志输出格式，指定 traceId -->
    <property name="LOG_PATTERN" value="%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} - [TraceID: %X{traceId}] - %msg%n" />

    <!-- 控制台输出 -->
    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>${LOG_PATTERN}</pattern>
        </encoder>
    </appender>

    <!-- 设置全局日志级别为 INFO -->
    <root level="INFO">
        <appender-ref ref="CONSOLE" />
    </root>
</configuration>

要实现 “全链路 TraceID”，我们采用装饰器模式（Decorator Pattern），在任务提交时手动捕获主线程的 MDC 并注入到任务中。

@Slf4j
@Service
public class UserService {
    private final ThreadPoolExecutor executor = new ThreadPoolExecutor(4, 4, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>());

    /**
     * 业务方法
     */
    public void executeAsyncFunc(Runnable task) {
        log.info("executeAsyncFunc 开始执行...");
        Map<String, String> context = MDC.getCopyOfContextMap();
        executor.submit(() -> {
            try {
                // 任务启动时，还原上下文
                MDC.setContextMap(context);
                log.info("mdc setContextMap...");
                task.run();
            } finally {
                // 任务结束时，务必清理，防止虚拟线程池复用污染
                log.info("mdc clear...");
                MDC.clear();
            }
        });
    }
}

测试类：

@Slf4j
@SpringBootTest(classes = App.class)
public class UserServiceTest {

    @Resource
    private UserService userService;

    @Test
    public void test01() {
        // 放入一个 TraceID
        MDC.put("traceId", UUID.randomUUID().toString().replace("-", ""));
        try {
            userService.executeAsyncFunc(() -> log.info("测试方法开始执行"));
        } finally {
            MDC.clear(); // 测试结束清理
        }
    }
}

日志输出：

2023-11-15 15:18:10.905 [main] INFO  demo01.service.UserService - [TraceID: dd74db8d0c0449a192642f5f3512233f] - executeAsyncFunc 开始执行...
2023-11-15 15:18:10.906 [pool-2-thread-1] INFO  demo01.service.UserService - [TraceID: dd74db8d0c0449a192642f5f3512233f] - mdc setContextMap...
2023-11-15 15:18:10.906 [pool-2-thread-1] INFO  service.UserServiceTest - [TraceID: dd74db8d0c0449a192642f5f3512233f] - 测试方法开始执行
2023-11-15 15:18:10.906 [pool-2-thread-1] INFO  demo01.service.UserService - [TraceID: dd74db8d0c0449a192642f5f3512233f] - mdc clear...

有限资源的治理

我们知道，即使有虚拟线程，数据库连接池也不是无限的。 我们可以在使用 Executors.newVirtualThreadPerTaskExecutor() 的同时，结合 Semaphore 对特定高风险资源（如：调用第三方支付接口、慢 SQL 查询）进行 “并发度限流”。

在虚拟线程场景下，Semaphore 是控制 “资源并发访问” 的最轻量级工具。它不同于 RateLimiter（控制频率），它是直接控制 “在途任务数”。你可以将 Semaphore 视为 “连接池的守护门卫”。

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

@Slf4j
public class DatabaseAccessLayer {
    
    // 假设数据库连接池最大支持 50 个并发
    private final Semaphore dbSemaphore = new Semaphore(50);
    // 虚拟线程池
    private final ExecutorService vtExecutor = Executors.newVirtualThreadPerTaskExecutor();

    public void executeQuery(String sql) {
        vtExecutor.submit(() -> {
            try {
                // 1. 在获取连接之前，先申请信号量许可
                dbSemaphore.acquire();

                try {
                    // 2. 执行慢 SQL
                    log.info("Executing slow SQL...");
                    // database.execute(sql);
                } finally {
                    // 3. 必须在 finally 中释放，确保资源回收
                    dbSemaphore.release();
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
    }
}

这样，即使你的 Web 层通过虚拟线程接收了 5000 个请求，这个 Semaphore 会确保只有 50 个请求能真正触达数据库，其余的 4950 个请求会被阻塞在 acquire() 这一步，直到前面的任务完成。如果没有这个限制，当 SQL 变慢时，所有虚拟线程都会挂在数据库连接上，连接池耗尽，随后所有业务（包括那些简单的查询）都会因为拿不到连接而超时。你还可以使用 dbSemaphore.tryAcquire(500, TimeUnit.MILLISECONDS)。如果超过 500ms 还没拿到许可，直接抛出 “系统繁忙” 异常，而不是让用户无限等待。

如果想用 Resilience4j 来管理数据库访问，可以这样配置：

resilience4j:
  bulkhead:
    instances:
      dbQuery:
        maxConcurrentCalls: 50
        maxWaitDuration: 500ms # 等待 500ms 拿不到资源就放弃

然后在代码中：

@Bulkhead(name = "dbQuery", type = Bulkhead.Type.SEMAPHORE)
public void performSlowQuery() {
    // 访问数据库
}

标题：

提升高并发系统鲁棒性三件套 - Resilience4j、MDC、Semaphore

作者：

kinglyjn

声明：

引用 本文 请务必注明出处，原创不易，感谢尊重。

java, core

java