springbootgateway记录请求和响应日志

　　springboot gateway 记录请求和响应日志
　　spring cloud gateway是基于webflux的项目，因而不能跟使用spring mvc一样直接获取request body，因此需要重新构造再转发。
　　如果我们在spring cloud gateway 封装之前读取了一次request body，比如打印request body日志，在下游获取数据的时候会出现错误：[spring cloud] [error] java.lang.IllegalStateException: Only one connection receive subscriber allowed. 因为request body只能读取一次，它是属于消费类型的。
　　出现这样的原因是InputStream的read()方法内部有一个postion，标志当前流被读取到的位置，每读取一次，该标志就会移动一次，如果读到最后，read()会返回-1，表示已经读取完了。如果想要重新读取则需要调用reset()方法，position就会移动到上次调用mark的位置，mark默认是0，所以就能从头再读了。调用reset()方法的前提是已经重写了reset()方法，当然能否reset也是有条件的，它取决于markSupported()方法是否返回true，InputStream默认不实现reset()，并且markSupported()默认也是返回false。综上，InputStream默认不实现reset的相关方法，而ServletInputStream也没有重写reset的相关方法，这样就无法重复读取流，这就是我们从request对象中获取的输入流就只能读取一次的原因。
　　直接上代码 import lombok.Data;  import java.io.Serializable; import java.util.Date;  @Data public class GatewayLog implements Serializable {     private static final long serialVersionUID = 1983879536575766072L;     /**访问实例*/     private String targetServer;     /**请求路径*/     private String requestPath;     /**请求方法*/     private String requestMethod;     /**协议 */     private String schema;     /**请求体*/     private String requestBody;     /**响应体*/     private String responseData;     /**请求ip*/     private String ip;    /**请求时间*/     private Date requestTime;    /**响应时间*/     private Date responseTime;     /**执行时间*/     private long executeTime;     /**返回码*/     private long code;     /**返回数据类型*/     private String responseContentType;     /**请求数据类型*/     private String requestContentType;     /**请求用户id*/     private String userId;   }
　　public interface AccessLogService {      void saveAccessLog(GatewayLog gatewayLog);  }
　　import cn.hutool.core.collection.CollectionUtil; import com.shouwei.gateway.entity.GatewayLog; import com.shouwei.gateway.service.AccessLogService; import com.shouwei.gateway.utils.IpUtils; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang.StringUtils; import org.reactivestreams.Publisher; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.cloud.gateway.filter.GatewayFilterChain; import org.springframework.cloud.gateway.filter.GlobalFilter; import org.springframework.cloud.gateway.filter.factory.rewrite.CachedBodyOutputMessage; import org.springframework.cloud.gateway.route.Route; import org.springframework.cloud.gateway.support.BodyInserterContext; import org.springframework.cloud.gateway.support.ServerWebExchangeUtils; import org.springframework.core.Ordered; import org.springframework.core.ResolvableType; import org.springframework.core.io.buffer.*; import org.springframework.http.HttpHeaders; import org.springframework.http.MediaType; import org.springframework.http.codec.HttpMessageReader; import org.springframework.http.codec.multipart.FormFieldPart; import org.springframework.http.codec.multipart.Part; import org.springframework.http.server.reactive.ServerHttpRequest; import org.springframework.http.server.reactive.ServerHttpRequestDecorator; import org.springframework.http.server.reactive.ServerHttpResponse; import org.springframework.http.server.reactive.ServerHttpResponseDecorator; import org.springframework.stereotype.Component; import org.springframework.util.LinkedMultiValueMap; import org.springframework.util.MultiValueMap; import org.springframework.web.reactive.function.BodyInserter; import org.springframework.web.reactive.function.BodyInserters; import org.springframework.web.reactive.function.server.HandlerStrategies; import org.springframework.web.reactive.function.server.ServerRequest; import org.springframework.web.server.ServerWebExchange; import org.springframework.web.util.UriComponentsBuilder; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono;  import java.io.UnsupportedEncodingException; import java.nio.charset.StandardCharsets; import java.util.*; import java.util.regex.Matcher; import java.util.regex.Pattern;  /**  * /**  * 全局拦截器，作用所有的微服务  * 1. 对请求的API调用过滤，记录接口的请求时间，方便日志审计、告警、分析等运维操作  * 2. 后期可以扩展对接其他日志系统  */ @Slf4j @Component public class AccessLogFilter implements GlobalFilter, Ordered {      /**      * default HttpMessageReader.      */     private static final List> MESSAGE_READERS = HandlerStrategies.withDefaults().messageReaders();       @Autowired     private AccessLogService accessLogService;      private final List> messageReaders = HandlerStrategies.withDefaults().messageReaders();       /**      * 顺序必须是<-1，否则标准的NettyWriteResponseFilter将在您的过滤器得到一个被调用的机会之前发送响应      * 也就是说如果不小于 -1 ，将不会执行获取后端响应的逻辑      *      * @return      */     @Override     public int getOrder() {         return -100;     }      @Override     @SuppressWarnings(＂unchecked＂)     public Mono filter(ServerWebExchange exchange, GatewayFilterChain chain) {          ServerHttpRequest request = exchange.getRequest();          // 请求路径         String requestPath = request.getPath().pathWithinApplication().value();          Route route = getGatewayRoute(exchange);          String ipAddress = IpUtils.getIpAddress(request);          GatewayLog gatewayLog = new GatewayLog();         gatewayLog.setSchema(request.getURI().getScheme());         gatewayLog.setRequestMethod(request.getMethodValue());         gatewayLog.setRequestPath(requestPath);         gatewayLog.setTargetServer(route.getId());         gatewayLog.setRequestTime(new Date());         gatewayLog.setIp(ipAddress);          MediaType mediaType = request.getHeaders().getContentType();         gatewayLog.setRequestContentType(mediaType.getType() + ＂/＂ + mediaType.getSubtype());         //json格式         if (MediaType.APPLICATION_JSON.isCompatibleWith(mediaType)) {             return writeBodyLog(exchange, chain, gatewayLog);         }          //form-data格式         else if (MediaType.MULTIPART_FORM_DATA.isCompatibleWith(mediaType) || MediaType.APPLICATION_FORM_URLENCODED.isCompatibleWith(mediaType)) {             return readFormData(exchange, chain, gatewayLog);         }          //其他格式         else {             return writeBasicLog(exchange, chain, gatewayLog);         }     }      private Mono writeBasicLog(ServerWebExchange exchange, GatewayFilterChain chain, GatewayLog accessLog) {          return DataBufferUtils.join(exchange.getRequest().getBody()).flatMap(dataBuffer -> {             DataBufferUtils.retain(dataBuffer);             final Flux cachedFlux = Flux.defer(() -> Flux.just(dataBuffer.slice(0, dataBuffer.readableByteCount())));             final ServerHttpRequest mutatedRequest = new ServerHttpRequestDecorator(exchange.getRequest()) {                 @Override                 public Flux getBody() {                     return cachedFlux;                 }                  @Override                 public MultiValueMap getQueryParams() {                     return UriComponentsBuilder.fromUri(exchange.getRequest().getURI()).build().getQueryParams();                 }              };              StringBuilder builder = new StringBuilder();             MultiValueMap queryParams = exchange.getRequest().getQueryParams();             if (CollectionUtil.isNotEmpty(queryParams)) {                 for (Map.Entry> entry : queryParams.entrySet()) {                     builder.append(entry.getKey()).append(＂=＂).append(StringUtils.join(entry.getValue(), ＂,＂));                 }             }              accessLog.setRequestBody(builder.toString());              //获取响应体             ServerHttpResponseDecorator decoratedResponse = recordResponseLog(exchange, accessLog);              return chain.filter(exchange.mutate().request(mutatedRequest).response(decoratedResponse).build())                     .then(Mono.fromRunnable(() -> {                         // 打印日志                         writeAccessLog(accessLog);                     }));          });      }      /**      * 读取form-data数据      *      * @param exchange      * @param chain      * @param accessLog      * @return      */     private Mono readFormData(ServerWebExchange exchange, GatewayFilterChain chain, GatewayLog accessLog) {          return DataBufferUtils.join(exchange.getRequest().getBody()).flatMap(dataBuffer -> {             DataBufferUtils.retain(dataBuffer);             final Flux cachedFlux = Flux.defer(() -> Flux.just(dataBuffer.slice(0, dataBuffer.readableByteCount())));             final ServerHttpRequest mutatedRequest = new ServerHttpRequestDecorator(exchange.getRequest()) {                 @Override                 public Flux getBody() {                     return cachedFlux;                 }                  @Override                 public MultiValueMap getQueryParams() {                     return UriComponentsBuilder.fromUri(exchange.getRequest().getURI()).build().getQueryParams();                 }             };              final HttpHeaders headers = exchange.getRequest().getHeaders();             if (headers.getContentLength() == 0) {                 return chain.filter(exchange);             }              ResolvableType resolvableType;             if (MediaType.MULTIPART_FORM_DATA.isCompatibleWith(headers.getContentType())) {                 resolvableType = ResolvableType.forClassWithGenerics(MultiValueMap.class, String.class, Part.class);             } else {                 //解析 application/x-www-form-urlencoded                 resolvableType = ResolvableType.forClass(String.class);             }              return MESSAGE_READERS.stream().filter(reader -> reader.canRead(resolvableType, mutatedRequest.getHeaders().getContentType())).findFirst()                     .orElseThrow(() -> new IllegalStateException(＂no suitable HttpMessageReader.＂)).readMono(resolvableType, mutatedRequest, Collections.emptyMap()).flatMap(resolvedBody -> {                         if (resolvedBody instanceof MultiValueMap) {                             LinkedMultiValueMap map = (LinkedMultiValueMap) resolvedBody;                             if (CollectionUtil.isNotEmpty(map)) {                                  StringBuilder builder = new StringBuilder();                                  final Part bodyPartInfo = (Part) ((MultiValueMap) resolvedBody).getFirst(＂body＂);                                 if (bodyPartInfo instanceof FormFieldPart) {                                     String body = ((FormFieldPart) bodyPartInfo).value(); //                                    log.info(＂body ==== ＂ + body);                                     builder.append(＂body=＂).append(body);                                 }                                  final Part uidPartInfo = (Part) ((MultiValueMap) resolvedBody).getFirst(＂uid＂);                                 if (uidPartInfo instanceof FormFieldPart) {                                     String uid = ((FormFieldPart) uidPartInfo).value(); //                                    log.info(＂uid ==== ＂ + uid);                                     accessLog.setUserId(uid);                                     if (builder.length() > 0) {                                         builder.append(＂&uid=＂).append(uid);                                     } else {                                         builder.append(＂uid=＂).append(uid);                                     }                                  }                                  final Part timeStampPartInfo = (Part) ((MultiValueMap) resolvedBody).getFirst(＂timeStamp＂);                                 if (timeStampPartInfo instanceof FormFieldPart) {                                     String timeStamp = ((FormFieldPart) timeStampPartInfo).value(); //                                    log.info(＂timeStamp ==== ＂ + timeStamp );                                     if (builder.length() > 0) {                                         builder.append(＂&timeStamp=＂).append(timeStamp);                                     } else {                                         builder.append(＂timeStamp=＂).append(timeStamp);                                     }                                  }                                  final Part tokenPartInfo = (Part) ((MultiValueMap) resolvedBody).getFirst(＂token＂);                                 if (tokenPartInfo instanceof FormFieldPart) {                                     String token = ((FormFieldPart) tokenPartInfo).value(); //                                    log.info(＂token ==== ＂ + token);                                     if (builder.length() > 0) {                                         builder.append(＂&token=＂).append(token);                                     } else {                                         builder.append(＂token=＂).append(token);                                     }                                  }                                  final Part signPartInfo = (Part) ((MultiValueMap) resolvedBody).getFirst(＂sign＂);                                 if (signPartInfo instanceof FormFieldPart) {                                     String sign = ((FormFieldPart) signPartInfo).value(); //                                    log.info(＂sign ==== ＂ + sign);                                     if (builder.length() > 0) {                                         builder.append(＂&sign=＂).append(sign);                                     } else {                                         builder.append(＂sign=＂).append(sign);                                     }                                 }                                  accessLog.setRequestBody(builder.toString());                             }                         } else {                             accessLog.setRequestBody((String) resolvedBody);                         }                          //获取响应体                         ServerHttpResponseDecorator decoratedResponse = recordResponseLog(exchange, accessLog);                          return chain.filter(exchange.mutate().request(mutatedRequest).response(decoratedResponse).build())                                 .then(Mono.fromRunnable(() -> {                                     // 打印日志                                     writeAccessLog(accessLog);                                 }));                     });          });     }       /**      * 解决 request body 只能读取一次问题，      * 参考: org.springframework.cloud.gateway.filter.factory.rewrite.ModifyRequestBodyGatewayFilterFactory      *      * @param exchange      * @param chain      * @param gatewayLog      * @return      */     @SuppressWarnings(＂unchecked＂)     private Mono writeBodyLog(ServerWebExchange exchange, GatewayFilterChain chain, GatewayLog gatewayLog) {         ServerRequest serverRequest = ServerRequest.create(exchange, messageReaders);          Mono modifiedBody = serverRequest.bodyToMono(String.class)                 .flatMap(body -> {                     gatewayLog.setRequestBody(body);                     return Mono.just(body);                 });          // 通过 BodyInserter 插入 body(支持修改body), 避免 request body 只能获取一次         BodyInserter bodyInserter = BodyInserters.fromPublisher(modifiedBody, String.class);         HttpHeaders headers = new HttpHeaders();         headers.putAll(exchange.getRequest().getHeaders());         // the new content type will be computed by bodyInserter         // and then set in the request decorator         headers.remove(HttpHeaders.CONTENT_LENGTH);          CachedBodyOutputMessage outputMessage = new CachedBodyOutputMessage(exchange, headers);          return bodyInserter.insert(outputMessage, new BodyInserterContext())                 .then(Mono.defer(() -> {                     // 重新封装请求                     ServerHttpRequest decoratedRequest = requestDecorate(exchange, headers, outputMessage);                      // 记录响应日志                     ServerHttpResponseDecorator decoratedResponse = recordResponseLog(exchange, gatewayLog);                      // 记录普通的                     return chain.filter(exchange.mutate().request(decoratedRequest).response(decoratedResponse).build())                             .then(Mono.fromRunnable(() -> {                                 // 打印日志                                 writeAccessLog(gatewayLog);                             }));                 }));     }      /**      * 打印日志      *      * @param gatewayLog 网关日志      * @author javadaily      * @date 2021/3/24 14:53      */     private void writeAccessLog(GatewayLog gatewayLog) {         accessLogService.saveAccessLog(gatewayLog);     }       private Route getGatewayRoute(ServerWebExchange exchange) {         return exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_ROUTE_ATTR);     }       /**      * 请求装饰器，重新计算 headers      *      * @param exchange      * @param headers      * @param outputMessage      * @return      */     private ServerHttpRequestDecorator requestDecorate(ServerWebExchange exchange, HttpHeaders headers,                                                        CachedBodyOutputMessage outputMessage) {         return new ServerHttpRequestDecorator(exchange.getRequest()) {             @Override             public HttpHeaders getHeaders() {                 long contentLength = headers.getContentLength();                 HttpHeaders httpHeaders = new HttpHeaders();                 httpHeaders.putAll(super.getHeaders());                 if (contentLength > 0) {                     httpHeaders.setContentLength(contentLength);                 } else {                     // TODO: this causes a ＂HTTP/1.1 411 Length Required＂ // on                     // httpbin.org                     httpHeaders.set(HttpHeaders.TRANSFER_ENCODING, ＂chunked＂);                 }                 return httpHeaders;             }              @Override             public Flux getBody() {                 return outputMessage.getBody();             }         };     }       /**      * 记录响应日志      * 通过 DataBufferFactory 解决响应体分段传输问题。      */     private ServerHttpResponseDecorator recordResponseLog(ServerWebExchange exchange, GatewayLog gatewayLog) {         ServerHttpResponse response = exchange.getResponse();         DataBufferFactory bufferFactory = response.bufferFactory();          return new ServerHttpResponseDecorator(response) {             @Override             public Mono writeWith(Publisher<? extends DataBuffer> body) {                 if (body instanceof Flux) {                     Date responseTime = new Date();                     gatewayLog.setResponseTime(responseTime);                     // 计算执行时间                     long executeTime = (responseTime.getTime() - gatewayLog.getRequestTime().getTime());                      gatewayLog.setExecuteTime(executeTime);                      // 获取响应类型，如果是 json 就打印                     String originalResponseContentType = exchange.getAttribute(ServerWebExchangeUtils.ORIGINAL_RESPONSE_CONTENT_TYPE_ATTR); //                    log.info(＂originalResponseContentType =========== ＂ + originalResponseContentType);                      gatewayLog.setResponseContentType(originalResponseContentType);                      gatewayLog.setCode(this.getStatusCode().value());  //                    if (ObjectUtils.equals(this.getStatusCode(), HttpStatus.OK) //                            && !StringUtil.isNullOrEmpty(originalResponseContentType) //                            && originalResponseContentType.contains(＂application/json＂)) {                      Flux<? extends DataBuffer> fluxBody = Flux.from(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 responseResult = new String(content, StandardCharsets.UTF_8); //                            log.info(＂responseResult =========== ＂ + responseResult);                         gatewayLog.setResponseData(responseResult);                          return bufferFactory.wrap(content);                     }));                 } //                }                  // if body is not a flux. never got there.                 return super.writeWith(body);             }         };     }   }
　　用到的依赖     org.springframework.cloud     spring-cloud-starter-gateway     2.2.5.RELEASE          org.projectlombok       lombok       1.18.22            cn.hutool       hutool-all       5.2.5