背景需求
想象一幕场景，我们有一套接口，现在需要适配多版本或者多个调用方，而它们之间对接口的约定可能会存在一些细小的差别，比如说数据结构的不同，字段名的差异或者是状态码的定义不同。为了实现这种需求，如果把这些差异的处理逻辑放到各个接口中，无疑显得重复而啰嗦，不够简洁，也不易于维护。更好的方式，当然是通过对接口的拦截，统一并且透明的对返回内容进行处理。
源码分析
对于在WebMVC中拦截和修改response，大家肯定都很熟悉，直接实现HandlerInterceptor
接口，即可实现对一个controller方法的执行前后进行拦截，从而可以去读取返回内容，修改之后再写回去。
换到WebFlux环境下，就和传统的基于Servlet的阻塞式请求处理流程完全不同，SpringBoot为WebFlux实现了一整套完全不同于WebMVC的流程和API，没有了HandlerInterceptor
，提供了一个WebFilter
接口用于请求流程的控制和处理
public interface WebFilter {
/**
* Process the Web request and (optionally) delegate to the next
* {@code WebFilter} through the given {@link WebFilterChain}.
* @param exchange the current server exchange
* @param chain provides a way to delegate to the next filter
* @return {@code Mono<Void>} to indicate when request processing is complete
*/
Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain);
}
其中ServerWebExchange
里封装了request和response，WebFilterChain
用于触发调用链的下一环节，通过这个我们可以控制调用链是继续往下还是中止，鉴权等操作即可在这一步进行。
那如何去修改response呢？接着往下看，ServerWebExchange
可以通过getResponse
方法，得到一个ServerHttpResponse
对象，我们查看一下这个类的方法
好像并没有哪个方法可以让我们直接读取到返回内容，反而提供两个用于写入内容的方法writeWith
和writeAndFlushWith
。如果我们不需要原内容，仅仅只需要写入一些固定的内容，比如鉴权失败等错误返回，那么到此也就足够了。但是我们的初衷是在原内容的基础上进行修改，所以还是需要想办法取得原内容。
我们换一个角度去想，WebFlux框架是如何将内容写入到ServerHttpResponse
中的，我们编写一个demo，通过打断点的方式，发现是通过HttpMessageWriter
接口的write
方法，将controller方法的返回值进行编码序列化，然后写入到response中。HttpMessageWriter
接口有很多的实现类，对于使用jackson进行json序列化的接口来说，会调用EncoderHttpMessageWriter
实现来进行处理，具体的write逻辑如下：
public class EncoderHttpMessageWriter<T> implements HttpMessageWriter<T> {
@Override
public Mono<Void> write(Publisher<? extends T> inputStream, ResolvableType elementType,
@Nullable MediaType mediaType, ReactiveHttpOutputMessage message, Map<String, Object> hints) {
MediaType contentType = updateContentType(message, mediaType);
Flux<DataBuffer> body = this.encoder.encode(
inputStream, message.bufferFactory(), elementType, contentType, hints);
if (inputStream instanceof Mono) {
return body
.singleOrEmpty()
.switchIfEmpty(Mono.defer(() -> {
message.getHeaders().setContentLength(0);
return message.setComplete().then(Mono.empty());
}))
.flatMap(buffer -> {
Hints.touchDataBuffer(buffer, hints, logger);
message.getHeaders().setContentLength(buffer.readableByteCount());
return message.writeWith(Mono.just(buffer)
.doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release));
})
.doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release);
}
if (isStreamingMediaType(contentType)) {
return message.writeAndFlushWith(body.map(buffer -> {
Hints.touchDataBuffer(buffer, hints, logger);
return Mono.just(buffer).doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release);
}));
}
if (logger.isDebugEnabled()) {
body = body.doOnNext(buffer -> Hints.touchDataBuffer(buffer, hints, logger));
}
return message.writeWith(body);
}
}
此处的message即为ServerHttpResponse
对象，可见框架是通过writeWith
方法进行数据写入的，那我们是否可以自己构造ServerHttpResponse
子类，并重写writeWith
来实现数据读取和修改呢
实现手段
幸运的是，WebFlux允许我们去自定义ServerHttpResponse
，并且应用到请求流程中。
我们回到之前提到的ServerWebExchange
中，这个类提供一个mutate方法
/**
* Return a builder to mutate properties of this exchange by wrapping it
* with {@link ServerWebExchangeDecorator} and returning either mutated
* values or delegating back to this instance.
*/
default Builder mutate() {
return new DefaultServerWebExchangeBuilder(this);
}
返回一个Builder
对象用于构造一个新的ServerWebExchange
对象，在这个过程中，可以传入我们自定义的ServerHttpResponse
对象。
/**
* Builder for mutating an existing {@link ServerWebExchange}.
* Removes the need
*/
interface Builder {
/**
* Configure a consumer to modify the current request using a builder.
* <p>Effectively this:
* <pre>
* exchange.mutate().request(builder-> builder.method(HttpMethod.PUT));
*
* // vs...
*
* ServerHttpRequest request = exchange.getRequest().mutate()
* .method(HttpMethod.PUT)
* .build();
*
* exchange.mutate().request(request);
* </pre>
* @see ServerHttpRequest#mutate()
*/
Builder request(Consumer<ServerHttpRequest.Builder> requestBuilderConsumer);
/**
* Set the request to use especially when there is a need to override
* {@link ServerHttpRequest} methods. To simply mutate request properties
* see {@link #request(Consumer)} instead.
* @see org.springframework.http.server.reactive.ServerHttpRequestDecorator
*/
Builder request(ServerHttpRequest request);
/**
* Set the response to use.
* @see org.springframework.http.server.reactive.ServerHttpResponseDecorator
*/
Builder response(ServerHttpResponse response);
/**
* Set the {@code Mono<Principal>} to return for this exchange.
*/
Builder principal(Mono<Principal> principalMono);
/**
* Build a {@link ServerWebExchange} decorator with the mutated properties.
*/
ServerWebExchange build();
}
对于如何自定义ServerHttpResponse
，WebFlux提供了一个便捷的装饰类ServerHttpResponseDecorator
，用于对原始response对象进行代理，并且可在需要修改的地方进行自定义
public class ServerHttpResponseDecorator implements ServerHttpResponse {
private final ServerHttpResponse delegate;
public ServerHttpResponseDecorator(ServerHttpResponse delegate) {
Assert.notNull(delegate, "Delegate is required");
this.delegate = delegate;
}
public ServerHttpResponse getDelegate() {
return this.delegate;
}
}
最终实现代码如下：
@Component
public class ChangeResponseFilter implements WebFilter {
@Override
public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
ServerHttpResponse original = exchange.getResponse();
// 创建新的response装饰对象，并传入原始response对象，然后重写writeWith方法
ServerHttpResponseDecorator newResponse = new ServerHttpResponseDecorator(original) {
@Override
public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
Mono<DataBuffer> mono = DataBufferUtils.join(body)
.map(buffer -> {
try {
String data = buffer.toString(StandardCharsets.UTF_8);
String mutatedData = mutate(data); // 此处修改返回内容
byte[] bytes = mutatedData.getBytes(StandardCharsets.UTF_8);
getHeaders().setContentLength(bytes.length); // ①
return this.bufferFactory().wrap(bytes); // ②
} finally {
DataBufferUtils.release(buffer); // ③
}
});
return super.writeWith(mono);
}
};
return chain.filter(exchange.mutate().response(newResponse).build());
}
}
其中有几个地方需要注意
① 修改了返回数据之后，需要重新设置Content-Length
头，否则可能造成数据传输不完整的情况
② 获取buffer工厂，使用新的数据重新生成ByteBuffer
对象
③ 对于自己创建的ByteBuffer
对象，在使用完成之后，需要手动release，否则会造成内存泄露
