Spring Cloud Gateway 拦截响应问题分析(数据截断问题)

时间:2023-02-02 起风哥人气:0

Spring Cloud Gateway是Spring 官方基于Spring 5.0、Spring Boot 2.0和Project Reactor等技术开发的网关，Spring Cloud Gateway旨在为微服务架构提供一种简单有效的、统一的 API 路由管理方式。

Spring Cloud Gateway 作为 Spring Cloud 生态系中的网关，其目标是替代 Netflix Zuul，它不仅提供统一的路由方式，并且基于Filter链的方式提供了网关基本的功能，如：安全、监控/埋点和限流等。

Spring Cloud Gateway依赖Spring Boot和Spring WebFlux，基于Netty 运行。不能在传统的 servlet 容器中工作也不能构建成war包。

关于Spring Cloud Gateway 核心概念

1、Route

Route 是网关的基础元素，由 ID、目标 URI、断言、过滤器组成。当请求到达网关时，由 Gateway HandlerMapping 通过断言进行路由匹配（Mapping），断言为真时匹配到路由。

2、Predicate

Predicate 是 Java 8 中提供的一个函数。输入类型是 Spring Framework ServerWebExchange。它允许开发人员匹配来自 HTTP 的请求，例如请求头或者请求参数。简单来说它就是匹配条件。

3、Filter

Filter是Gateway 中的过滤器，可以在请求发出前后进行一些业务上的处理。

Spring Cloud Gateway 拦截响应

最近因为上链路追踪后发现如果在服务层将异常拦截掉，在链路追踪界面上就不会显示异常链路信息，除了服务异常意外，系统的异常不会触发链路error，所以对服务层做了个变更，将所有未处理异常直接捕获后统一封装完抛出，这个时候就需要在网关层统一处理那么网关需要对响应数据进行拦截如果是 9999错误码，则封装后返回，如果是其它响应码或者其它数据直接返回。

起初写法测试后发现数据过长时被截断。

 return super.writeWith(fluxBody.map(dataBuffer -> {
	
 
	byte[] content = new byte[dataBuffer.readableByteCount()];
	
	dataBuffer.read(content);
	// 释放掉内存
	DataBufferUtils.release(dataBuffer);
	String str = new String(content, Charset.forName("UTF-8"));
 
	originalResponse.getHeaders().setContentLength(str.getBytes().length);
	log.error("gateway catch service exception error:"+ str);
	
	JsonResult result = new JsonResult();
	result.setCode(ErrorCode.SYS_EXCEPTION.getCode());
	result.setMessage(ErrorCode.SYS_EXCEPTION.getMsg());
	
	return bufferFactory.wrap(str.getBytes());
}));

查询api后发现存在一个DataBufferFactory可以一次性join完所有数据后拼接就不会产生截断问题。

DataBufferFactory dataBufferFactory = new DefaultDataBufferFactory();

于是修改代码如下

package com.server.gateway.filters;
 
import java.nio.charset.Charset;
import java.util.concurrent.atomic.AtomicReference;
 
import org.apache.http.protocol.HTTP;
import org.reactivestreams.Publisher;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.core.Ordered;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.http.server.reactive.ServerHttpResponseDecorator;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
 
import com.framework.common.enums.ErrorCode;
import com.framework.common.web.JsonResult;
 
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
 
@Component
@Slf4j
public class WrapperResponseGlobalFilter implements GlobalFilter, Ordered{
 
	@Override
	public int getOrder() {
		// -1 is response write filter, must be called before that
		return -2;
	}
 
	@Override
	public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
		ServerHttpResponse originalResponse = exchange.getResponse();
		DataBufferFactory bufferFactory = originalResponse.bufferFactory();
		ServerHttpResponseDecorator decoratedResponse = new ServerHttpResponseDecorator(originalResponse) {
			@Override
			public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
				AtomicReference<String> bodyRef = new AtomicReference<>();
				if (body instanceof Flux) {
					Flux<? extends DataBuffer> fluxBody = (Flux<? extends DataBuffer>) body;
					
					return super.writeWith(fluxBody.buffer().map(dataBuffers -> {
						
						DataBufferFactory dataBufferFactory = new DefaultDataBufferFactory();
						DataBuffer join = dataBufferFactory.join(dataBuffers);
						
						byte[] content = new byte[join.readableByteCount()];
						
						join.read(content);
						// 释放掉内存
						DataBufferUtils.release(join);
						String str = new String(content, Charset.forName("UTF-8"));
		
						originalResponse.getHeaders().setContentLength(str.getBytes().length);
						log.error("gateway catch service exception error:"+ str);
						
						JsonResult result = new JsonResult();
				        result.setCode(ErrorCode.SYS_EXCEPTION.getCode());
				        result.setMessage(ErrorCode.SYS_EXCEPTION.getMsg());
				        
						return bufferFactory.wrap(str.getBytes());
					}));
					
				}
				// if body is not a flux. never got there.
	                return super.writeWith(body);
	            }
	        };
	        // replace response with decorator
	        return chain.filter(exchange.mutate().response(decoratedResponse).build());
	    }
 
}

经过如上修改后链路追踪可以实现哪个服务出错就立马出现链路异常马上可以定位到哪个服务出现了未处理异常

加载全部内容