Spring Cloud Gateway 响应装饰器 writeWith 不执行：一次从 order 到响应式生命周期的深度排查

Novice_programme

367人浏览 · 2026-07-30 01:29:06

Novice_programme · 2026-07-30 01:29:06 发布

开发环境

JDK 版本：21

SpringBoot 版本：3.5.16

SpringCloud 版本：2025.0.0

一、背景与问题

在基于 Spring Cloud Gateway 构建的 API 开放平台中，我实现了一个 GlobalFilter，核心职责包括：

请求鉴权：校验 accessKey/secretKey 签名、nonce、timestamp 等；
响应拦截：通过 ServerHttpResponseDecorator 装饰响应对象，在 writeWith 中读取并记录下游响应体；
调用计数：每次成功调用后，对 user_interface_info 表的调用次数 +1。

核心代码结构如下：

@Component
public class CustomGlobalFilter implements GlobalFilter, Ordered {

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        // 1. 请求日志
        // 2. IP 白名单校验
        // 3. 用户鉴权（ak/sk 签名验证）
        // 4. 接口存在性校验
        // 5. 请求转发 + 响应装饰
        return handleResponse(exchange, chain, interfaceInfo.getId(), invokeUser.getId());
    }

    public Mono<Void> handleResponse(ServerWebExchange exchange, GatewayFilterChain chain,
                                     long interfaceInfoId, long userId) {
        ServerHttpResponse originalResponse = exchange.getResponse();
        DataBufferFactory bufferFactory = originalResponse.bufferFactory();

        ServerHttpResponseDecorator decoratedResponse = new ServerHttpResponseDecorator(originalResponse) {
            @Override
            public Mono<Void> writeWith(@NonNull Publisher<? extends DataBuffer> body) {
                return super.writeWith(
                    Flux.from(body).map(dataBuffer -> {
                        byte[] content = new byte[dataBuffer.readableByteCount()];
                        dataBuffer.read(content);
                        DataBufferUtils.release(dataBuffer);
                        log.info("响应结果: status={}, body={}", getStatusCode(),
                                new String(content, StandardCharsets.UTF_8));
                        return bufferFactory.wrap(content);
                    })
                ).doOnSuccess(v -> {
                    try {
                        innerUserInterfaceInfoService.invokeCount(interfaceInfoId, userId);
                    } catch (Exception e) {
                        log.error("invokeCount error", e);
                    }
                });
            }
        };

        return chain.filter(exchange.mutate().response(decoratedResponse).build());
    }

    @Override
    public int getOrder() {
        return 0;  // ← 问题根源
    }
}

现象：请求正常发出，客户端能收到完整响应（POST 用户名字是Max），handleResponse 方法体同步执行（断点命中），但 writeWith 内的日志和 invokeCount 从未执行。

二、排查过程：排除干扰项

2.1 排除"外层调用方式"问题

最初怀疑外层 filter 方法中 handleResponse 的返回值被丢弃或被 .then(...) 包裹导致装饰器链未被订阅。检查后确认外层是干净的：

return handleResponse(exchange, chain, interfaceInfo.getId(), invokeUser.getId());

✅ 全链路只 chain.filter 一次，且带装饰器。排除。

2.2 排除"writeWith 内部逻辑"问题

删除了 if (body instanceof Flux) 判断（Flux.from(body) 本身就能适配 Mono/任意 Publisher）；
将 invokeCount 从 map（每个 DataBuffer 执行一次）挪到 doOnSuccess（一次响应仅执行一次）；
确认 DataBufferUtils.release(dataBuffer) 正确释放内存。

✅ 内部逻辑无误。排除。

2.3 锁定真凶：`getOrder()` 返回值

在 IDE 中 Ctrl 点进 org.springframework.cloud.gateway.filter.NettyWriteResponseFilter，看到：

public class NettyWriteResponseFilter implements GlobalFilter, Ordered {
    public static final int WRITE_RESPONSE_FILTER_ORDER = -1;

    @Override
    public int getOrder() {
        return WRITE_RESPONSE_FILTER_ORDER;  // -1
    }
}

而我的 filter：getOrder() 返回 0，大于 -1。

三、根因分析：Gateway Filter Chain 的执行模型

3.1 链的排序与执行方向

Spring Cloud Gateway 的 filter 链按 getOrder() 升序排列：

pre 阶段（正向）：order 小的先执行；
then 阶段（回溯）：order 小的后执行（类似洋葱模型）。

当 getOrder() = 0 时，链顺序为：

[NettyWriteResponseFilter(-1)] → [CustomGlobalFilter(0)] → … → [NettyRoutingFilter(转发)]

3.2 NettyWriteResponseFilter 的核心逻辑

// 框架内置，简化版
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
    return chain.filter(exchange)                    // 1.把【收到的 exchange】原样往下传
        .then(Mono.defer(() -> {
            ClientResponse resp = exchange.getAttribute(CLIENT_RESPONSE_ATTR);
            DataBuffer body = ...;
            return exchange.getResponse().writeWith(body);  // 2.用【收到的 exchange】的 response 写回
        }));
}

关键：它把谁往下传、最后用谁的 response 写回，取决于它收到的 exchange 是哪个。

3.3 装饰器为何被绕过

逐步推演 order = 0 时的执行流：

步骤	发生了什么
1	`NettyWriteResponseFilter`（order -1）先执行，手里是原始 exchange E0。调用 `chain.filter(E0)` 往下传，并挂一个 `then`：回溯时用 `E0.getResponse().writeWith(...)` 写回。
2	`CustomGlobalFilter`（order 0）收到 E0，执行 `exchange.mutate().response(decorated).build()` 造出新 exchange E1，调用 `chain.filter(E1)` 往下传。
3	转发 filter 用 E1 发请求，将下游响应存入 attributes（E0、E1 共享同一张 attributes Map）。
4	回溯到 `NettyWriteResponseFilter` 的 `then`：`E0.getResponse().writeWith(...)` —— E0 的 response 是原始的，不是 `decoratedResponse`。
5	框架默认 `writeWith` 从共享 attributes 中取出下游 body，写给客户端。

结果：客户端照常收到完整响应（因为共享 attributes），但装饰器的 writeWith 从未被调用。

这就是"客户端有响应、handleResponse 同步命中、writeWith 一行不跑"这一诡异现象的完整解释。

3.4 修复后的执行流（order = -2）

[CustomGlobalFilter(-2)] → [NettyWriteResponseFilter(-1)] → … → [NettyRoutingFilter(转发)]

步骤	发生了什么
1	`CustomGlobalFilter`（order -2）先执行，mutate 出 E1（带装饰器），`chain.filter(E1)` 往下传。
2	`NettyWriteResponseFilter`（order -1）收到的就是 E1，原样 `chain.filter(E1)` 转发，挂 `then`。
3	转发 filter 发请求，存响应到 attributes。
4	回溯到 `NettyWriteResponseFilter` 的 `then`：`E1.getResponse().writeWith(...)` —— E1 的 response 正是 `decoratedResponse`！
5	装饰器 `writeWith` 被调用 → `map` 改写 body → `doOnSuccess` 执行 `invokeCount` → `super.writeWith` 将最终 body 交给原始 response 写回客户端。

✅ 链路彻底打通。

四、修复方案

@Override
public int getOrder() {
    // 必须严格小于 NettyWriteResponseFilter 的 -1
    // 不要写 -1（撞 order 顺序不稳定），更不要写 0
    return -2;
}

或使用注解：

@Component
@Order(-2)
public class CustomGlobalFilter implements GlobalFilter { ... }

五、验证

修复后发送请求，日志输出：

INFO  CustomGlobalFilter : 请求唯一标识：c2cc7b6a-2
INFO  CustomGlobalFilter : 请求路径：/api/mock-user/user
INFO  CustomGlobalFilter : 请求方法：POST
INFO  CustomGlobalFilter : 客户端地址：IP:127.0.0.1, Port:8414
INFO  CustomGlobalFilter : writeWith entered, body=class reactor.core.publisher.FluxMap
INFO  CustomGlobalFilter : 响应结果: status=200 OK, body=POST 用户名字是Max

writeWith entered 打印 → 装饰器生效；
body=class reactor.core.publisher.FluxMap → 上游给的是 Flux 子类，Flux.from(body) 正确处理；
响应结果 打印 → body 读取、改写、日志全链路正常；
数据库 totalNum +1 → invokeCount 执行成功。

六、writeWith 内部的最佳实践（踩坑总结）

6.1 不要判断 `body instanceof Flux`

// ❌ 错误：Mono 形式的响应体会被挡进 else，误设 500
if (body instanceof Flux) { ... } else { setStatusCode(500); }

// ✅ 正确：Flux.from() 能适配任意 Publisher
return super.writeWith(Flux.from(body).map(...));

6.2 invokeCount 放在 `doOnSuccess`，不要放在 `map`

// ❌ 错误：map 是"每个 DataBuffer 执行一次"
//   - body 为空 → 一次不进 → 漏计
//   - body 分片（压缩/大报文）→ 进多次 → 重复计
fluxBody.map(dataBuffer -> {
    invokeCount(...);  // 可能执行 0 次或 N 次
    ...
});

// ✅ 正确：doOnSuccess 是"整个写入完成执行一次"
super.writeWith(Flux.from(body).map(...))
    .doOnSuccess(v -> {
        invokeCount(...);  // 恰好一次
    });

6.3 释放 DataBuffer

byte[] content = new byte[dataBuffer.readableByteCount()];
dataBuffer.read(content);
DataBufferUtils.release(dataBuffer);  // 必须释放，否则内存泄漏
return bufferFactory.wrap(content);   // 用新 buffer 包装后返回

七、两个"不影响功能但影响质量"的 TODO

7.1 `if (statusCode == HttpStatus.OK)` 是假判断

handleResponse 跑在转发之前，此时下游尚未被调用，getStatusCode() 返回 Spring 默认值 200，该 if 恒为 true。它并没有在过滤下游状态码。

若要真正判断下游响应码，应在 writeWith 内部使用 getStatusCode()（此时才是下游的真实码）。

7.2 pre 阶段的 Dubbo 调用是同步阻塞的

// ❌ 阻塞 Netty event-loop 线程
DubboResult<User> dubboResult = innerUserService.getInvokeUser(accessKey);

filter 方法跑在 reactor-http-nio-* 线程（Netty event-loop）上，同步 RPC 会阻塞整个线程，拖慢网关吞吐。建议改为：

return Mono.fromCallable(() -> innerUserService.getInvokeUser(accessKey))
    .subscribeOn(Schedulers.boundedElastic())
    .flatMap(dubboResult -> {
        // 后续鉴权逻辑...
    });

八、核心经验总结

#	经验
1	Gateway 中做响应后置装饰，第一件事检查 order 是否压过 `NettyWriteResponseFilter` 的 -1。这是所有"writeWith 不执行"问题的头号元凶。
2	`exchange.mutate().build()` 产生的新 exchange 只向下游传播，不会回传给上游 filter。上游 filter 手里永远是它最初收到的那个 exchange。
3	`mutate().build()` 与原 exchange 共享 attributes Map，所以即使装饰器被绕过，客户端仍能收到正确响应——这制造了"一切正常"的假象，极具迷惑性。
4	`writeWith` 的执行时机是回溯阶段（then），不是 pre 阶段。pre 阶段的断点/日志能命中，不代表 writeWith 会被调用。
5	响应式编程中，`map` 是逐元素操作，`doOnSuccess` 是生命周期钩子。"一次请求计一次数"的逻辑必须挂在后者上。
6	不要在 Netty event-loop 线程上做阻塞 I/O（同步 RPC、JDBC），否则一个慢调用就能拖垮整个网关。

九、最终完整代码（修复后）

@Component
public class CustomGlobalFilter implements GlobalFilter, Ordered {

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        // 1. 请求日志
        ServerHttpRequest request = exchange.getRequest();
        log.info("请求路径：{}, 方法：{}", request.getPath().value(), request.getMethod());

        // 2. IP 白名单
        String clientIp = getClientIp(request);
        if (!IP_WHITE_LIST.contains(clientIp)) {
            exchange.getResponse().setStatusCode(HttpStatus.FORBIDDEN);
            return exchange.getResponse().setComplete();
        }

        // 3. 用户鉴权（ak/sk 签名）
        // ... 省略校验逻辑，失败则 setComplete()

        // 4. 接口存在性校验
        // ... 省略，失败则 setComplete()

        // 5. 转发 + 响应装饰
        return handleResponse(exchange, chain, interfaceInfo.getId(), invokeUser.getId());
    }

    public Mono<Void> handleResponse(ServerWebExchange exchange, GatewayFilterChain chain,
                                     long interfaceInfoId, long userId) {
        ServerHttpResponse originalResponse = exchange.getResponse();
        DataBufferFactory bufferFactory = originalResponse.bufferFactory();

        ServerHttpResponseDecorator decoratedResponse = new ServerHttpResponseDecorator(originalResponse) {
            @NotNull
            @Override
            public Mono<Void> writeWith(@NonNull Publisher<? extends DataBuffer> body) {
                return super.writeWith(
                    Flux.from(body).map(dataBuffer -> {
                        byte[] content = new byte[dataBuffer.readableByteCount()];
                        dataBuffer.read(content);
                        DataBufferUtils.release(dataBuffer);
                        log.info("响应结果: status={}, body={}", getStatusCode(),
                                new String(content, StandardCharsets.UTF_8));
                        return bufferFactory.wrap(content);
                    })
                ).doOnSuccess(v -> {
                    try {
                        innerUserInterfaceInfoService.invokeCount(interfaceInfoId, userId);
                        log.info("invokeCount +1, interfaceInfoId={}, userId={}", interfaceInfoId, userId);
                    } catch (Exception e) {
                        log.error("invokeCount error", e);
                    }
                });
            }
        };

        return chain.filter(exchange.mutate().response(decoratedResponse).build());
    }

    @Override
    public int getOrder() {
        return -2;  // 必须 < NettyWriteResponseFilter 的 -1
    }
}