Spring Cloud Gateway全局过滤器WebFlux线程切换？Schedulers.boundedElastic与上下文传递 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

Spring Cloud Gateway 全局过滤器 WebFlux 线程切换与上下文传递

大家好，今天我们来深入探讨 Spring Cloud Gateway 中全局过滤器与 WebFlux 线程模型交互时的一些关键问题，特别是线程切换策略的选择以及上下文传递机制。我们将着重分析 Schedulers.boundedElastic 的使用，并提供具体的代码示例来说明如何在实际应用中进行最佳实践。

1. WebFlux 线程模型简介

WebFlux 是 Spring Framework 提供的响应式编程框架，基于 Reactor 库实现。与传统的 Servlet 线程模型不同，WebFlux 使用非阻塞 I/O 和事件驱动的方式处理请求，充分利用多核 CPU 的优势，提升系统的吞吐量和响应速度。

WebFlux 的核心在于它的异步非阻塞特性。这意味着当一个请求到达时，WebFlux 不会像传统的线程模型那样阻塞当前线程等待 I/O 操作完成，而是将 I/O 操作委托给其他线程池或者事件循环，然后立即返回到工作线程处理其他请求。当 I/O 操作完成后，再通过回调或者事件通知的方式将结果传递回来。

2. Spring Cloud Gateway 与 WebFlux 集成

Spring Cloud Gateway 是一个基于 Spring WebFlux 构建的 API 网关，它利用 WebFlux 的非阻塞特性来实现高性能的请求路由和过滤。Gateway 的核心组件是 GatewayFilter，它可以对请求进行各种处理，例如认证授权、请求转换、流量控制等。

GatewayFilter 在 WebFlux 的上下文中执行，因此需要特别注意线程切换和上下文传递的问题。不恰当的线程切换策略可能会导致性能瓶颈或者数据丢失。

3. 全局过滤器与线程切换

全局过滤器是应用于所有路由的过滤器，它们在请求到达网关之后，路由到目标服务之前执行。在全局过滤器中，我们经常需要执行一些耗时的操作，例如调用外部服务、访问数据库等。为了避免阻塞 WebFlux 的工作线程，我们需要将这些耗时操作提交到其他的线程池执行。

这里就引出了线程切换的问题。在 WebFlux 中，线程切换是通过 subscribeOn() 和 publishOn() 操作符来实现的。

subscribeOn(Scheduler): 指定 Publisher（例如 Mono 或 Flux）的订阅操作在哪个 Scheduler 上执行。它影响的是数据流的源头，决定了数据流的起始线程。
publishOn(Scheduler): 指定 Publisher 的后续操作在哪个 Scheduler 上执行。它影响的是数据流的中间和最终处理线程，可以在数据流的任意位置进行线程切换。

4. Schedulers.boundedElastic() 详解

Schedulers.boundedElastic() 是 Reactor 提供的一个线程池调度器，它适用于执行阻塞 I/O 操作或者 CPU 密集型任务。它的特点是：

弹性伸缩: 它会根据需要创建新的线程，直到达到配置的最大线程数。
线程回收: 空闲线程会被回收，避免资源浪费。
限制并发: 它可以限制并发执行的任务数量，防止系统过载。

Schedulers.boundedElastic() 的默认配置是：

最大线程数：CPU核心数 * 10
线程空闲超时时间：60秒

我们可以通过配置来调整这些参数：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

@Configuration
public class SchedulerConfig {

    @Bean
    public Scheduler customBoundedElastic() {
        return Schedulers.newBoundedElastic(200, 10, "custom-bounded-elastic"); // 200 maxThreads, 10 threadTtlSeconds, name
    }
}

这段代码定义了一个名为 customBoundedElastic 的 Bean，它创建了一个最大线程数为 200，线程空闲超时时间为 10 秒的 Schedulers.boundedElastic() 实例。

5. 在全局过滤器中使用 Schedulers.boundedElastic()

下面是一个使用 Schedulers.boundedElastic() 的全局过滤器的示例：

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.core.Ordered;
import org.springframework.http.HttpHeaders;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.util.UUID;

@Component
public class LoggingGlobalFilter implements GlobalFilter, Ordered {

    private static final Logger logger = LoggerFactory.getLogger(LoggingGlobalFilter.class);

    private final Scheduler boundedElastic;

    public LoggingGlobalFilter() {
        this.boundedElastic = Schedulers.boundedElastic(); // 使用默认配置
    }

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        String traceId = UUID.randomUUID().toString();
        exchange.getAttributes().put("traceId", traceId);

        // 在 boundedElastic 线程池中执行耗时操作
        return Mono.fromRunnable(() -> {
                try {
                    // 模拟耗时操作
                    Thread.sleep(100);
                    HttpHeaders headers = exchange.getRequest().getHeaders();
                    logger.info("Request Headers: {}, traceId: {}", headers, traceId);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt(); // 恢复中断状态
                    logger.error("Interrupted!", e);
                }
            })
            .subscribeOn(boundedElastic) // 将 Mono 的订阅操作提交到 boundedElastic 线程池
            .then(chain.filter(exchange)); // 继续执行过滤器链
    }

    @Override
    public int getOrder() {
        return 0; // 设置过滤器的执行顺序
    }
}

在这个示例中，LoggingGlobalFilter 会记录请求的 Headers 信息。为了避免阻塞 WebFlux 的工作线程，我们将记录 Headers 的操作提交到 Schedulers.boundedElastic() 线程池中执行。

代码解释：

Mono.fromRunnable(() -> { ... }): 将一个 Runnable 包装成 Mono，这意味着 Runnable 中的代码会被异步执行。
.subscribeOn(boundedElastic): 指定 Mono 的订阅操作在 boundedElastic 线程池中执行。这意味着 Runnable 中的代码会在 boundedElastic 线程池的线程中执行。
.then(chain.filter(exchange)): 在 Mono 执行完成后，继续执行过滤器链。

6. 上下文传递

在 WebFlux 中，上下文信息存储在 Context 对象中。Context 是一个不可变的键值对集合，它可以用来传递请求的信息、用户身份、Trace ID 等。

在全局过滤器中，我们需要保证上下文信息在线程切换之后仍然能够正确传递。Reactor 提供了 contextWrite() 和 contextRead() 操作符来实现上下文的传递。

contextWrite(Context): 向 Publisher 的上下文中写入新的键值对。
contextRead(Function<Context, T>): 从 Publisher 的上下文中读取指定键的值。

让我们修改上面的 LoggingGlobalFilter 示例，使其能够从上下文中读取 Trace ID：

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.core.Ordered;
import org.springframework.http.HttpHeaders;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import reactor.util.context.Context;

import java.util.UUID;

@Component
public class LoggingGlobalFilter implements GlobalFilter, Ordered {

    private static final Logger logger = LoggerFactory.getLogger(LoggingGlobalFilter.class);

    private final Scheduler boundedElastic;

    public LoggingGlobalFilter() {
        this.boundedElastic = Schedulers.boundedElastic(); // 使用默认配置
    }

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        String traceId = UUID.randomUUID().toString();

        // 将 Trace ID 写入上下文
        return chain.filter(exchange)
            .contextWrite(context -> context.put("traceId", traceId))
            .then(Mono.deferContextual(contextView -> {
                String id = contextView.get("traceId"); // 从上下文中读取 Trace ID
                return Mono.fromRunnable(() -> {
                    try {
                        // 模拟耗时操作
                        Thread.sleep(100);
                        HttpHeaders headers = exchange.getRequest().getHeaders();
                        logger.info("Request Headers: {}, traceId: {}", headers, id); // 使用从上下文读取的 Trace ID
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        logger.error("Interrupted!", e);
                    }
                }).subscribeOn(boundedElastic);
            }));
    }

    @Override
    public int getOrder() {
        return 0;
    }
}

代码解释：

.contextWrite(context -> context.put("traceId", traceId)): 将 Trace ID 写入上下文。
Mono.deferContextual(contextView -> { ... }): 创建一个延迟执行的 Mono，它可以在订阅时访问上下文。
contextView.get("traceId"): 从上下文中读取 Trace ID。

7. 线程切换策略选择

在 Spring Cloud Gateway 中，选择合适的线程切换策略至关重要。以下是一些常用的线程池调度器：

Scheduler	适用场景	优点	缺点
`Schedulers.immediate()`	同步执行，不切换线程。适用于执行非常快的非阻塞操作。	简单高效，避免线程切换的开销。	不适合执行耗时操作，会阻塞 WebFlux 的工作线程。
`Schedulers.single()`	使用单个线程执行任务。适用于执行顺序敏感的任务。	保证任务的顺序执行，避免并发问题。	吞吐量较低，不适合执行大量并发任务。
`Schedulers.newParallel(String)`	创建一个固定大小的线程池，用于执行并行任务。适用于 CPU 密集型任务。	并行执行任务，提高 CPU 利用率。	如果任务阻塞，会导致线程池中的线程被占用，影响其他任务的执行。
`Schedulers.boundedElastic()`	弹性伸缩的线程池，适用于执行阻塞 I/O 操作或者 CPU 密集型任务。	弹性伸缩，避免资源浪费；限制并发，防止系统过载。	线程切换的开销比 `Schedulers.immediate()` 和 `Schedulers.single()` 大。
`Schedulers.fromExecutorService()`	使用自定义的 `ExecutorService`。适用于需要更精细的线程池控制的场景。	可以根据需要定制线程池的参数，例如线程数、队列大小等。	需要手动管理线程池的生命周期，避免资源泄漏。

选择建议：

对于非常快的非阻塞操作，可以使用 Schedulers.immediate()。
对于顺序敏感的任务，可以使用 Schedulers.single()。
对于 CPU 密集型任务，可以使用 Schedulers.newParallel(String)。
对于阻塞 I/O 操作或者 CPU 密集型任务，可以使用 Schedulers.boundedElastic()。
如果需要更精细的线程池控制，可以使用 Schedulers.fromExecutorService()。

8. 总结与要点回顾

今天我们深入探讨了 Spring Cloud Gateway 中全局过滤器与 WebFlux 线程模型交互时的一些关键问题。我们详细讲解了 Schedulers.boundedElastic() 的使用方法和适用场景，并提供了代码示例来说明如何在实际应用中进行最佳实践。

最后，记住以下几个关键点：

WebFlux 使用非阻塞 I/O 和事件驱动的方式处理请求，充分利用多核 CPU 的优势。
全局过滤器在 WebFlux 的上下文中执行，需要特别注意线程切换和上下文传递的问题。
Schedulers.boundedElastic() 适用于执行阻塞 I/O 操作或者 CPU 密集型任务。
使用 contextWrite() 和 contextRead() 操作符来实现上下文的传递。
选择合适的线程切换策略至关重要，需要根据实际场景进行选择。

希望这次的分享能够帮助大家更好地理解 Spring Cloud Gateway 的线程模型，并在实际应用中编写出高性能、高可靠性的网关服务。

Spring Cloud Gateway 全局过滤器 WebFlux 线程切换与上下文传递

发表回复 取消回复

发表回复取消回复