欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

SpringCloud Finchley Gateway 缓存请求Body和Form表单的实现

程序员文章站 2024-02-23 21:14:28
在接入spring-cloud-gateway时,可能有需求进行缓存json-body数据或者form-urlencoded数据的情况。 由于spring-cloud...

在接入spring-cloud-gateway时,可能有需求进行缓存json-body数据或者form-urlencoded数据的情况。

由于spring-cloud-gateway是以webflux为基础的响应式架构设计,所以在原有zuul基础上迁移过来的过程中,传统的编程思路,并不适合于reactor stream的开发。

网络上有许多缓存案例,但是在测试过程中出现各种bug问题,在缓存body时,需要考虑整体的响应式操作,才能更合理的缓存数据

下面提供缓存json-body数据或者form-urlencoded数据的具体实现方案,该方案经测试,满足各方面需求,以及避免了网络上其他缓存方案所出现的问题

定义一个gatewaycontext类,用于存储请求中缓存的数据

import lombok.getter;
import lombok.setter;
import lombok.tostring;
import org.springframework.util.linkedmultivaluemap;
import org.springframework.util.multivaluemap;

@getter
@setter
@tostring
public class gatewaycontext {

  public static final string cache_gateway_context = "cachegatewaycontext";

  /**
   * cache json body
   */
  private string cachebody;
  /**
   * cache formdata
   */
  private multivaluemap<string, string> formdata;
  /**
   * cache reqeust path
   */
  private string path;
}

实现globalfilter和ordered接口用于缓存请求数据

1 . 该示例只支持缓存下面3种mediatype

  • application_json--json数据
  • application_json_utf8--json数据
  • application_form_urlencoded--formdata表单数据

2 . 经验总结:

  • 在缓存body时,不能够在filter内部直接进行缓存,需要按照响应式的处理方式,在异步操作路途上进行缓存body,由于body只能读取一次,所以要读取完成后要重新封装新的request和exchange才能保证请求正常传递到下游
  • 在缓存formdata时,formdata也只能读取一次,所以在读取完毕后,需要重新封装request和exchange,这里要注意,如果对formdata内容进行了修改,则必须重新定义header中的content-length已保证传输数据的大小一致
import com.choice.cloud.architect.usergate.option.filterorderenum;
import com.choice.cloud.architect.usergate.support.gatewaycontext;
import io.netty.buffer.bytebufallocator;
import lombok.extern.slf4j.slf4j;
import org.springframework.cloud.gateway.filter.gatewayfilterchain;
import org.springframework.cloud.gateway.filter.globalfilter;
import org.springframework.core.ordered;
import org.springframework.core.io.bytearrayresource;
import org.springframework.core.io.buffer.databuffer;
import org.springframework.core.io.buffer.databufferutils;
import org.springframework.core.io.buffer.nettydatabufferfactory;
import org.springframework.http.httpheaders;
import org.springframework.http.mediatype;
import org.springframework.http.codec.httpmessagereader;
import org.springframework.http.server.reactive.serverhttprequest;
import org.springframework.http.server.reactive.serverhttprequestdecorator;
import org.springframework.util.multivaluemap;
import org.springframework.web.reactive.function.server.handlerstrategies;
import org.springframework.web.reactive.function.server.serverrequest;
import org.springframework.web.server.serverwebexchange;
import reactor.core.publisher.flux;
import reactor.core.publisher.mono;

import java.io.unsupportedencodingexception;
import java.net.urlencoder;
import java.nio.charset.charset;
import java.nio.charset.standardcharsets;
import java.util.list;
import java.util.map;

@slf4j
public class gatewaycontextfilter implements globalfilter, ordered {

  /**
   * default httpmessagereader
   */
  private static final list<httpmessagereader<?>> messagereaders = handlerstrategies.withdefaults().messagereaders();

  @override
  public mono<void> filter(serverwebexchange exchange, gatewayfilterchain chain) {
    /**
     * save request path and serviceid into gateway context
     */
    serverhttprequest request = exchange.getrequest();
    string path = request.getpath().pathwithinapplication().value();
    gatewaycontext gatewaycontext = new gatewaycontext();
    gatewaycontext.getallrequestdata().addall(request.getqueryparams());
    gatewaycontext.setpath(path);
    /**
     * save gateway context into exchange
     */
    exchange.getattributes().put(gatewaycontext.cache_gateway_context,gatewaycontext);
    httpheaders headers = request.getheaders();
    mediatype contenttype = headers.getcontenttype();
    long contentlength = headers.getcontentlength();
    if(contentlength>0){
      if(mediatype.application_json.equals(contenttype) || mediatype.application_json_utf8.equals(contenttype)){
        return readbody(exchange, chain,gatewaycontext);
      }
      if(mediatype.application_form_urlencoded.equals(contenttype)){
        return readformdata(exchange, chain,gatewaycontext);
      }
    }
    log.debug("[gatewaycontext]contenttype:{},gateway context is set with {}",contenttype, gatewaycontext);
    return chain.filter(exchange);

  }


  @override
  public int getorder() {
    return integer.min_value;
  }

  /**
   * readformdata
   * @param exchange
   * @param chain
   * @return
   */
  private mono<void> readformdata(serverwebexchange exchange,gatewayfilterchain chain,gatewaycontext gatewaycontext){
    httpheaders headers = exchange.getrequest().getheaders();
    return exchange.getformdata()
        .doonnext(multivaluemap -> {
          gatewaycontext.setformdata(multivaluemap);
          log.debug("[gatewaycontext]read formdata:{}",multivaluemap);
        })
        .then(mono.defer(() -> {
          charset charset = headers.getcontenttype().getcharset();
          charset = charset == null? standardcharsets.utf_8:charset;
          string charsetname = charset.name();
          multivaluemap<string, string> formdata = gatewaycontext.getformdata();
          /**
           * formdata is empty just return
           */
          if(null == formdata || formdata.isempty()){
            return chain.filter(exchange);
          }
          stringbuilder formdatabodybuilder = new stringbuilder();
          string entrykey;
          list<string> entryvalue;
          try {
            /**
             * remove system param ,repackage form data
             */
            for (map.entry<string, list<string>> entry : formdata.entryset()) {
              entrykey = entry.getkey();
              entryvalue = entry.getvalue();
              if (entryvalue.size() > 1) {
                for(string value : entryvalue){
                  formdatabodybuilder.append(entrykey).append("=").append(urlencoder.encode(value, charsetname)).append("&");
                }
              } else {
                formdatabodybuilder.append(entrykey).append("=").append(urlencoder.encode(entryvalue.get(0), charsetname)).append("&");
              }
            }
          }catch (unsupportedencodingexception e){
            //ignore urlencode exception
          }
          /**
           * substring with the last char '&'
           */
          string formdatabodystring = "";
          if(formdatabodybuilder.length()>0){
            formdatabodystring = formdatabodybuilder.substring(0, formdatabodybuilder.length() - 1);
          }
          /**
           * get data bytes
           */
          byte[] bodybytes = formdatabodystring.getbytes(charset);
          int contentlength = bodybytes.length;
          serverhttprequestdecorator decorator = new serverhttprequestdecorator(
              exchange.getrequest()) {
            /**
             * change content-length
             * @return
             */
            @override
            public httpheaders getheaders() {
              httpheaders httpheaders = new httpheaders();
              httpheaders.putall(super.getheaders());
              if (contentlength > 0) {
                httpheaders.setcontentlength(contentlength);
              } else {
                httpheaders.set(httpheaders.transfer_encoding, "chunked");
              }
              return httpheaders;
            }

            /**
             * read bytes to flux<databuffer>
             * @return
             */
            @override
            public flux<databuffer> getbody() {
              return databufferutils.read(new bytearrayresource(bodybytes),new nettydatabufferfactory(bytebufallocator.default),contentlength);
            }
          };
          serverwebexchange mutateexchange = exchange.mutate().request(decorator).build();
          log.debug("[gatewaycontext]rewrite form data :{}",formdatabodystring);
          return chain.filter(mutateexchange);
        }));
  }

  /**
   * readjsonbody
   * @param exchange
   * @param chain
   * @return
   */
  private mono<void> readbody(serverwebexchange exchange,gatewayfilterchain chain,gatewaycontext gatewaycontext){
    /**
     * join the body
     */
    return databufferutils.join(exchange.getrequest().getbody())
        .flatmap(databuffer -> {
          /**
           * read the body flux<databuffer>
           */
          databufferutils.retain(databuffer);
          flux<databuffer> cachedflux = flux.defer(() -> flux.just(databuffer.slice(0, databuffer.readablebytecount())));
          /**
           * repackage serverhttprequest
           */
          serverhttprequest mutatedrequest = new serverhttprequestdecorator(exchange.getrequest()) {
            @override
            public flux<databuffer> getbody() {
              return cachedflux;
            }
          };
          /**
           * mutate exchage with new serverhttprequest
           */
          serverwebexchange mutatedexchange = exchange.mutate().request(mutatedrequest).build();
          /**
           * read body string with default messagereaders
           */
          return serverrequest.create(mutatedexchange, messagereaders)
              .bodytomono(string.class)
              .doonnext(objectvalue -> {
                gatewaycontext.setcachebody(objectvalue);
                log.debug("[gatewaycontext]read jsonbody:{}",objectvalue);
              }).then(chain.filter(mutatedexchange));
        });
  }

}

在后续filter中,可以直接从serverexchange中获取gatewaycontext,就可以获取到缓存的数据,如果需要缓存其他数据,则可以根据自己的需求,添加到gatewaycontext中即可

复制代码 代码如下:
gatewaycontext gatewaycontext = exchange.getattribute(gatewaycontext.cache_gateway_context);

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。