Spring-Cloud-Gateway 從升級到放棄

本文轉載自查看原文 2019-03-06 16:31 11529 wrk/ 高並發/ 過濾器/ java 高級/ 長度丟失/ response/ spring-cloud-gateway/ 性能

1 為什么要升級為spring-cloud-gateway？

Spring Cloud Gateway features:

Built on Spring Framework 5, Project Reactor and Spring Boot 2.0
Able to match routes on any request attribute.
Predicates and filters are specific to routes.
Hystrix Circuit Breaker integration.
Spring Cloud DiscoveryClient integration
Easy to write Predicates and Filters
Request Rate Limiting
Path Rewriting

這是官方說的，spring gateway相對spring zuul要新很多，應用也更加自由，開發體驗更好。但是我在測試中發現，spring-cloud-gateway相對於zuul來說，更加出眾的還是其性能，當然最后讓我放棄的也是因為這一點。

網上的朋友也有做一些gateway和zuul的性能比較，大多的結論也是gateway要優於zuul，同時也更加穩定。

但是我們不能輕信，所以我也做了測試,這部分測試內容若不感興趣可以跳過，zuul就不測試了。

2.spring-cloud-gateway的初步測試

　　step.1：測試准備：

　　　　1.gateway版本：2.0.1

　　　　2.服務主機：10.1.4.32，16G內存，4核虛擬機

　　　　3.測試客戶端：10.1.4.34，16G內存，4核虛擬機

　　　　4.測試工具wrk

　　step.2：建立gateway工程並寫兩個測試http接口，

　　　　1.http://10.1.4.32:14077/hello [POST]

　　　　2.http://10.1.4.32:14077/test [GET]

　　step.3：開始測試

　　step.4：測試結果　　　

[wrk@localhost wrk]$ ./wrk  -t 15 -c500 -d 10 --latency  http://10.1.4.32:14077/test
Running 10s test @ http://10.1.4.32:14077/test
  15 threads and 500 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency     3.38ms    3.26ms 113.45ms   95.76%
    Req/Sec    10.84k     1.44k   26.48k    89.50%
  Latency Distribution
     50%    2.79ms
     75%    3.51ms
     90%    4.21ms
     99%   17.23ms
  1625714 requests in 10.10s, 131.79MB read
Requests/sec: 160961.07
Transfer/sec:     13.05MB

以及：

[wrk@localhost wrk]$ ./wrk  -t 15 -c500 -d 10 --latency -s scripts/gateway.lua  http://10.1.4.32:14077/hello
Running 10s test @ http://10.1.4.32:14077/hello
  15 threads and 500 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency     5.21ms    3.96ms 255.59ms   96.75%
    Req/Sec     6.62k   604.79    13.72k    88.48%
  Latency Distribution
     50%    4.78ms
     75%    5.55ms
     90%    6.32ms
     99%   14.87ms
  994374 requests in 10.10s, 539.59MB read
Requests/sec:  98471.14
Transfer/sec:     53.43MB

說明，如果測試結果差別較大可能是因為測試工具的問題。

結果顯示，POST方法的性能TPS達到了10W/s,而GET方法的性能TPS達到了16W/s。

這看起來很不可思議，因為正常的微服務，能達到2W/s的性能已經是良好，達到10W實在是不可思議。但是前面說了spring-cloud-gateway引入了Spring Reactor反應式編程，應對的便是這種高並發需求。

當然，即便spring-cloud-gateway給了我們很大驚喜，但是如果因此就引入了spring-cloud-gateway，那還是會有些草率，畢竟gateway是用來干什么的？是路由和過濾。繼續測試。

　　step.5：加上路由和過濾器,在配置文件中加入下面內容

spring:
  cloud:
    gateway:
      routes:
      - id: test
        uri: http://10.1.4.32:14077/test
        predicates:
        - Path=/tt
        filters:
        - AddRequestParameter=foo, bar

表示，給test方法加入了路由，並且加入了官方提供的過濾器：AddRequestParameter=foo, bar

　　step.6：測試，並附測試結果：

[wrk@localhost wrk]$ ./wrk  -t 15 -c500 -d 10 --latency  http://10.1.4.32:14077/tt
Running 10s test @ http://10.1.4.32:14077/tt
  15 threads and 500 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency    18.99ms   12.15ms 122.69ms   70.84%
    Req/Sec     1.82k   155.77     2.36k    73.94%
  Latency Distribution
     50%   17.03ms
     75%   25.49ms
     90%   35.02ms
     99%   57.13ms
  274529 requests in 10.10s, 22.25MB read
Requests/sec:  27182.88
Transfer/sec:      2.20MB

性能只剩27000/s，貌似降低了很多，但是比起zuul仍然快了不少。因為在這台機器上，測試zuul或許都不能到達2W。

那么，是不是就應該使用spring-cloud-gateway了？

3.開始使用spring-cloud-gateway

在使用上spring-cluod-gateway之后，我開始編輯自己的過濾器，需求要求寫兩個過濾器，修改請求體和響應體。

因為需要對特定的請求使用過濾器，所以這里使用gateway-filter，有些代碼官方有，有些網友提供，兩個過濾器代碼大致如下：

解密過濾器，pre：

package com.newland.dc.ctid.fileter;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.newland.dc.common.vo.RequestHeaderVo;
import com.newland.dc.ctid.entity.dto.RequestDto;
import io.netty.buffer.ByteBufAllocator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.gateway.filter.GatewayFilter;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.factory.AbstractGatewayFilterFactory;
import org.springframework.core.Ordered;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.NettyDataBufferFactory;
import org.springframework.core.style.ToStringCreator;
import org.springframework.http.MediaType;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpRequestDecorator;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.nio.CharBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;

/**
 * Created by garfield on 2019/2/26.
 */
@Component
public class DecryptGatewayFilterFactory extends AbstractGatewayFilterFactory<DecryptGatewayFilterFactory.Config>{

    private static Logger log = LoggerFactory.getLogger(DecryptGatewayFilterFactory.class);

    public static final String DECRYPT_HEADER = "decrypt_header";


    public DecryptGatewayFilterFactory() {
        super(Config.class);
    }


    private Gson gson = new GsonBuilder().serializeNulls().create();


    @Override
    @SuppressWarnings("unchecked")
    public GatewayFilter apply(Config config) {
        return new DecryptGatewayFilter(config);
    }

    public class DecryptGatewayFilter implements GatewayFilter, Ordered {
        Config config;

        DecryptGatewayFilter(Config config) {
            this.config = config;
        }
        @Override
        public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
                log.debug(config.toString());
                ServerHttpRequest request = exchange.getRequest();
                MediaType contentType = request.getHeaders().getContentType();

                boolean postRequest = "POST".equalsIgnoreCase(request.getMethodValue()) && !contentType.toString().contains("multipart/form-data");
                //判斷是否為POST請求
                if (postRequest) {

                    Flux<DataBuffer> body = request.getBody();
                    AtomicReference<String> bodyRef = new AtomicReference<>();//緩存讀取的request body信息
                    body.subscribe(dataBuffer -> {
                        CharBuffer charBuffer = StandardCharsets.UTF_8.decode(dataBuffer.asByteBuffer());
                        DataBufferUtils.release(dataBuffer);
                        bodyRef.set(charBuffer.toString());
                    });//讀取request body到緩存
                    String bodyStr = bodyRef.get();//獲取request body
                    log.debug(bodyStr);//這里是我們需要做的操作
                    RequestDto requestDto = gson.fromJson(bodyStr, RequestDto.class);
                    log.debug("decrypt filter");
                    //save header to response header
                    RequestHeaderVo headerVo = requestDto.getHeader();
                    headerVo.setAppVersion("1000");
                    //此處可以傳遞一些變量
                    exchange.getResponse().getHeaders().add(DECRYPT_HEADER, gson.toJson(headerVo));

                    DataBuffer bodyDataBuffer = stringBuffer(bodyStr);
                    Flux<DataBuffer> bodyFlux = Flux.just(bodyDataBuffer);

                    request = new ServerHttpRequestDecorator(request){
                        @Override
                        public Flux<DataBuffer> getBody() {
                            return bodyFlux;
                        }
                    };//封裝我們的request
                }
                return chain.filter(exchange.mutate().request(request.mutate().header("a","200").build()).build());
            };

        @Override
        public int getOrder() {
            return -10;
        }
    }

    public static DataBuffer stringBuffer(String value) {
        byte[] bytes = value.getBytes(StandardCharsets.UTF_8);

        NettyDataBufferFactory nettyDataBufferFactory = new NettyDataBufferFactory(ByteBufAllocator.DEFAULT);
        DataBuffer buffer = nettyDataBufferFactory.allocateBuffer(bytes.length);
        buffer.write(bytes);
        return buffer;
    }


    @Override
    public ServerHttpRequest.Builder mutate(ServerHttpRequest request) {
        return null;
    }

    public static class Config {
        private boolean decrypt;


        public boolean isDecrypt() {
            return decrypt;
        }

        public void setDecrypt(boolean decrypt) {
            this.decrypt = decrypt;
        }

        @Override
        public String toString() {
            return new ToStringCreator(this)
                    .append("decrypt", decrypt)
                    .toString();
        }
    }


    @Override
    public List<String> shortcutFieldOrder() {
        return Arrays.asList("decrypt");
    }
}

View Code

加密過濾器，使用源碼的提供的修改方法，post：

package com.newland.dc.ctid.fileter;

import com.google.gson.Gson;
import com.newland.dc.common.vo.RequestHeaderVo;
import com.newland.dc.ctid.entity.dto.RequestDto;
import org.springframework.cloud.gateway.filter.GatewayFilter;
import org.springframework.cloud.gateway.filter.factory.AbstractGatewayFilterFactory;
import org.springframework.cloud.gateway.filter.factory.rewrite.ModifyResponseBodyGatewayFilterFactory;
import org.springframework.core.style.ToStringCreator;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;

import java.util.Arrays;
import java.util.List;

/**
 * @Auther: garfield
 * @Date: 2019/3/5 15:33
 * @Description:
 */
@Component
public class AnGatewayFilterFactory extends AbstractGatewayFilterFactory<AnGatewayFilterFactory.Config> {


    private Gson gson = new Gson();

    public AnGatewayFilterFactory() {
        super(Config.class);
    }

    @Override
    public GatewayFilter apply(Config config) {

        ModifyResponseBodyGatewayFilterFactory m1 = new ModifyResponseBodyGatewayFilterFactory(null);
        ModifyResponseBodyGatewayFilterFactory.Config c1 = new ModifyResponseBodyGatewayFilterFactory.Config();
        c1.setInClass(String.class);
        c1.setOutClass(String.class);
        c1.setNewContentType("application/json");

        c1.setRewriteFunction((exchange, body) -> {
            ServerWebExchange ex = (ServerWebExchange) exchange;
            //此處更改響應體
            RequestHeaderVo requestHeaderVo = new RequestHeaderVo();
            RequestDto requestDto = gson.fromJson(body.toString(), RequestDto.class);
            requestDto.setHeader(requestHeaderVo);
            body = gson.toJson(requestDto);
            return Mono.just(body);
        });
        return m1.apply(c1);
    }

    public static class Config {
        private boolean decrypt;


        public boolean isDecrypt() {
            return decrypt;
        }

        public void setDecrypt(boolean decrypt) {
            this.decrypt = decrypt;
        }

        @Override
        public String toString() {
            return new ToStringCreator(this)
                    .append("encrypt", decrypt)
                    .toString();
        }
    }

    @Override
    public List<String> shortcutFieldOrder() {
        return Arrays.asList("encrypt");
    }
}

View Code

這里需要轉移一下話題，這個過濾器修改其實有幾種方法，可以自己寫，也可以應用源碼提供的例子。上面的兩種寫法已經測試都能使用，其實我還有兩種方式，大同小異就是了，但也准備貼出來，也記錄一下問題：

下面這個其實就是源碼中的例子，只不過不引用，自己寫：

    @Override
        @SuppressWarnings("unchecked")
        public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {

            ServerHttpResponseDecorator responseDecorator = new ServerHttpResponseDecorator(exchange.getResponse()) {
                @Override
                public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
                    ServerHttpRequest request = exchange.getRequest();

                    MediaType originalResponseContentType = exchange.getAttribute(ORIGINAL_RESPONSE_CONTENT_TYPE_ATTR);
                    HttpHeaders httpHeaders = new HttpHeaders();
                    httpHeaders.setContentType(originalResponseContentType);
                    ResponseAdapter responseAdapter = new ResponseAdapter(body, httpHeaders);
                    DefaultClientResponse clientResponse = new DefaultClientResponse(responseAdapter, ExchangeStrategies.withDefaults());

                    Mono<DataBuffer> modifiedBody = clientResponse.bodyToMono(DataBuffer.class).map(encrypt(config, new RequestHeaderVo()));

                    BodyInserter bodyInserter = BodyInserters.fromPublisher(modifiedBody, DataBuffer.class);
                    CachedBodyOutputMessage outputMessage = new CachedBodyOutputMessage(exchange, exchange.getResponse().getHeaders());
                    return bodyInserter.insert(outputMessage, new BodyInserterContext())
                            .then(Mono.defer(() -> {
                                Flux<DataBuffer> messageBody = outputMessage.getBody();
                                HttpHeaders headers = getDelegate().getHeaders();
                                if (headers.getContentLength() < 0 && !headers.containsKey(HttpHeaders.TRANSFER_ENCODING)) {
                                    messageBody = messageBody.doOnNext(data -> headers.setContentLength(data.readableByteCount()));
                                }
                                return this.getDelegate().writeWith(messageBody);
                            }));
                }

                @Override
                public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> body) {
                    return writeWith(Flux.from(body)
                            .flatMapSequential(p -> p));
                }
            };
            return chain.filter(exchange.mutate().response(responseDecorator).build());

        }

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

    }

    private Function<DataBuffer, DataBuffer> encrypt(Config config, RequestHeaderVo headerVo) {
        if (config.encrypt) {
            return (i) -> {
                InputStream inputStream = i.asInputStream();

                byte[] bytes = new byte[0];
                try {
                    bytes = new byte[inputStream.available()];


                    inputStream.read(bytes);
                } catch (IOException e) {
                    e.printStackTrace();
                }
                //進行我們的操作
                String body = new String(bytes);
                log.debug("this is response encrypt");
                log.debug(body);
                log.debug(headerVo.toString());
//                body = encryptService.responseEncrypt(body, headerVo);

                //進行我們的操作
                return i.write(TokenGatewayFilterFactory.stringBuffer(body));
//                return i.write(new String(body).getBytes());

            };


        } else {
            return i -> i;
        }

    }

View Code

這種例子中，發現修改response body的時候，會引起代碼進入NioEventLoop類中的run方法，死循環無法退出，我也不清楚為什么，修改需謹慎。

另一種，跟這位網友寫得差不多，只不過我沒測試就是了：https://www.jianshu.com/p/9f00e0e1681c

        @Override
        @SuppressWarnings("unchecked")
        public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
            return chain.filter(exchange.mutate().response(new ServerHttpResponseDecorator(exchange.getResponse()) {
                @Override
                public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
                    DataBufferFactory bufferFactory = exchange.getResponse().bufferFactory();
                    if (getStatusCode().equals(HttpStatus.OK) && body instanceof Flux) {
                        Flux<? extends DataBuffer> fluxBody = Flux.first(body);
                        return super.writeWith(fluxBody.map(dataBuffer -> {
                            System.out.println(dataBuffer.readableByteCount());

                            byte[] content = new byte[dataBuffer.readableByteCount()];
                            dataBuffer.read(content);
                            //釋放掉內存
                            DataBufferUtils.release(dataBuffer);
                            //responseData就是下游系統返回的內容,可以查看修改
                            String responseData = new String(content, Charset.forName("UTF-8"));

                            log.debug("響應內容:{}", responseData);
                            log.debug("this is response encrypt");
                            System.out.println(responseData);

                            byte[] newContent = responseData.getBytes();
//                body = encryptService.responseEncrypt(body, headerVo);
                            byte[] uppedContent = new String(newContent, Charset.forName("UTF-8")).getBytes();
                            return bufferFactory.wrap(uppedContent);
                        }));
                    } else {
                        log.error("響應code異常:{}", getStatusCode());
                    }
                    return super.writeWith(body);
                }
            }).build());
        }


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

View Code

這個方法會出現問題，body的截取長度經常沒有完全。

我本來是到這個網址下面尋找答案，作者是這樣回復的：

　　上面只是簡單的樣例，FIux是發送多個數據的，當報文長時會拆分，處理一次只能拿到一部分報文，可以使用Flux.toArray方法將數據聚合后處理，也可以參照https://www.jianshu.com/p/9b781fb1aaa0里面的響應處理。

確實是這個問題，所以我們也可以仿照他的另外一個例子寫，大家可以到他的簡書博客中去看，值得提醒的是，他的例子中，版本也是2.0.1,若是版本改為2.1以上，就不能用哦！

這里蠻貼一下：

package com.newland.dc.ctid.fileter;

import com.google.gson.Gson;
import com.newland.dc.common.vo.RequestHeaderVo;
import com.newland.dc.ctid.service.SecurityService;
import com.newland.dc.log.kafka.KafkaLog;
import org.reactivestreams.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.gateway.filter.GatewayFilter;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.NettyWriteResponseFilter;
import org.springframework.cloud.gateway.filter.factory.AbstractGatewayFilterFactory;
import org.springframework.cloud.gateway.support.*;
import org.springframework.core.Ordered;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.style.ToStringCreator;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseCookie;
import org.springframework.http.client.reactive.ClientHttpResponse;
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.MultiValueMap;
import org.springframework.web.reactive.function.client.ExchangeStrategies;
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.IOException;
import java.io.InputStream;
import java.util.*;
import java.util.function.BiFunction;
import java.util.function.Function;

import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.ORIGINAL_RESPONSE_CONTENT_TYPE_ATTR;


/**
 * @Auther: garfield
 * @Date: 2019/2/28 上午10:45
 * @Description:
 */
@Component
public class EncryptGatewayFilterFactory extends AbstractGatewayFilterFactory<EncryptGatewayFilterFactory.Config> {

    private static Logger log = LoggerFactory.getLogger(EncryptGatewayFilterFactory.class);


    @Autowired
    private SecurityService encryptService;

    public EncryptGatewayFilterFactory() {
        super(Config.class);
    }


    //    private Gson gson = new GsonBuilder().serializeNulls().create();
    private Gson gson = new Gson();

    @Value("${server.host:10.10.10.10}")
    private String serverHost;

    @Value("${server.port}")
    private String serverPort;


    @Override
    @SuppressWarnings("unchecked")
    public GatewayFilter apply(Config config) {
        return new EncryptGatewayFilter(config);
    }


    @Override
    public ServerHttpRequest.Builder mutate(ServerHttpRequest request) {
        return null;
    }

    public static class Config {

        private boolean encrypt;

        public boolean isEncrypt() {
            return encrypt;
        }

        public Config setEncrypt(boolean encrypt) {
            this.encrypt = encrypt;
            return this;
        }

        @Override
        public String toString() {
            return new ToStringCreator(this)
                    .append("encrypt", encrypt)
                    .toString();
        }
    }

    @Override
    public List<String> shortcutFieldOrder() {
        return Arrays.asList("encrypt");
    }

    public class EncryptGatewayFilter implements GatewayFilter, Ordered {
        Config config;

        EncryptGatewayFilter(Config config) {
            this.config = config;
        }

        @Override
        @SuppressWarnings("unchecked")
        public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
            String trace = exchange.getRequest().getHeaders().getFirst("trace");
            ServerRequest serverRequest = new DefaultServerRequest(exchange);
            return serverRequest.bodyToMono(String.class).flatMap(reqBody -> {
                //重寫原始請求
                ServerHttpRequestDecorator decorator = new ServerHttpRequestDecorator(exchange.getRequest()) {
                    @Override
                    public HttpHeaders getHeaders() {
                        HttpHeaders httpHeaders = new HttpHeaders();
                        httpHeaders.putAll(super.getHeaders());
                        return httpHeaders;
                    }

                    @Override
                    public Flux<DataBuffer> getBody() {
                        //打印原始請求日志
                        log.info("[Trace:{}]-gateway request:headers=[{}],body=[{}]", trace, getHeaders(), reqBody);
                        return Flux.just(reqBody).map(bx -> exchange.getResponse().bufferFactory().wrap(bx.getBytes()));
                    }
                };
                //重寫原始響應
                BodyHandlerServerHttpResponseDecorator responseDecorator = new BodyHandlerServerHttpResponseDecorator(
                        initBodyHandler(exchange), exchange.getResponse());

                return chain.filter(exchange.mutate().request(decorator).response(responseDecorator).build());
            });
        }

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

    }

    public interface BodyHandlerFunction
            extends BiFunction<ServerHttpResponse, Publisher<? extends DataBuffer>, Mono<Void>> {
    }

    protected BodyHandlerFunction initBodyHandler(ServerWebExchange exchange) {
        return (resp, body) -> {
            //攔截
            MediaType originalResponseContentType = exchange.getAttribute(ORIGINAL_RESPONSE_CONTENT_TYPE_ATTR);
            HttpHeaders httpHeaders = new HttpHeaders();
            httpHeaders.setContentType(originalResponseContentType);
            DefaultClientResponseAdapter clientResponseAdapter = new DefaultClientResponseAdapter(body, httpHeaders);
            Mono<String> bodyMono = clientResponseAdapter.bodyToMono(String.class);
            //此處可以獲得前面放置的參數
            return bodyMono.flatMap((respBody) -> {
//                打印返回響應日志
                System.out.println(respBody);
                return resp.writeWith(Flux.just(respBody).map(bx -> resp.bufferFactory().wrap(bx.getBytes())));
            }).then();
        };
    }

 
    public static class DefaultClientResponseAdapter extends DefaultClientResponse {

        /**
         * @param body
         * @param httpHeaders
         */
        public DefaultClientResponseAdapter(Publisher<? extends DataBuffer> body,
                                            HttpHeaders httpHeaders) {
            this(new ResponseAdapter(body, httpHeaders),
                    ExchangeStrategies.withDefaults());
        }

        /**
         * @param response
         * @param strategies
         */
        public DefaultClientResponseAdapter(ClientHttpResponse response,
                                            ExchangeStrategies strategies) {
            super(response, strategies);
        }

        /**
         * ClientHttpResponse 適配器
         */
        static class ResponseAdapter implements ClientHttpResponse {
            /**
             * 響應數據
             */
            private final Flux<DataBuffer> flux;
            /**
             *
             */
            private final HttpHeaders headers;

            public ResponseAdapter(Publisher<? extends DataBuffer> body,
                                   HttpHeaders headers) {
                this.headers = headers;
                if (body instanceof Flux) {
                    flux = (Flux) body;
                } else {
                    flux = ((Mono) body).flux();
                }
            }

            @Override
            public Flux<DataBuffer> getBody() {
                return flux;
            }

            @Override
            public HttpHeaders getHeaders() {
                return headers;
            }

            @Override
            public HttpStatus getStatusCode() {
                return null;
            }

            @Override
            public int getRawStatusCode() {
                return 0;
            }

            @Override
            public MultiValueMap<String, ResponseCookie> getCookies() {
                return null;
            }
        }
    }

    class BodyHandlerServerHttpResponseDecorator extends ServerHttpResponseDecorator {

        /**
         * body 處理攔截器
         */
        private BodyHandlerFunction bodyHandler = initDefaultBodyHandler();

        /**
         * 構造函數
         *
         * @param bodyHandler
         * @param delegate
         */
        public BodyHandlerServerHttpResponseDecorator(BodyHandlerFunction bodyHandler, ServerHttpResponse delegate) {
            super(delegate);
            if (bodyHandler != null) {
                this.bodyHandler = bodyHandler;
            }
        }

        @Override
        public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
            //body 攔截處理器處理響應
            return bodyHandler.apply(getDelegate(), body);
        }

        @Override
        public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> body) {
            return writeWith(Flux.from(body).flatMapSequential(p -> p));
        }

        /**
         * 默認body攔截處理器
         *
         * @return
         */
        private BodyHandlerFunction initDefaultBodyHandler() {
            return (resp, body) -> resp.writeWith(body);
        }
    }
}

View Code

那么萬事具備，代碼都寫好了，我們又需要進行性能測試。這邊要記住，我用的是官方的那個例子，其他的寫法也用過了，但是結果差不多。

4.再一次測試spring-cloud-gateway 網關路由性能

　　step.1：性能測試，改一下配置，表示加入了過濾器。這里為什么只有一個過濾器，因為這個過濾器問題比較大，過程就略過了。

      - id: an
        uri: http://10.1.4.32:14077/hello
        predicates:
        - Path=/an
        filters:
        - An

　　經過多次測試，其他的過濾器都還好，只有修改response body的過濾器，嚴重影響性能，且有讀寫錯誤。

　　step.2：測試，以及測試結果

[wrk@localhost wrk]$ ./wrk  -t 15 -c500 -d 10 --latency -s scripts/gateway.lua  http://10.1.4.32:14077/an
Running 10s test @ http://10.1.4.32:14077/an
  15 threads and 500 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency     1.03s   488.75ms   2.00s    60.62%
    Req/Sec    26.59     13.84    80.00     67.60%
  Latency Distribution
     50%  931.54ms
     75%    1.45s
     90%    1.76s
     99%    1.97s
  3848 requests in 10.10s, 1.64MB read
  Socket errors: connect 0, read 0, write 0, timeout 458
Requests/sec:    381.05
Transfer/sec:    166.71KB

結果多出一行，socket錯誤，而且還是超時,而且，日志中也存在錯誤：

2019-03-06T16:09:33,396|INFO ||AsyncResolver-bootstrap-executor-0||||Resolving eureka endpoints via configuration
2019-03-06T16:10:38,268|ERROR||reactor-http-server-epoll-18||||Unhandled failure: Connection has been closed, response already set (status=200)
2019-03-06T16:10:38,268|WARN ||reactor-http-server-epoll-18||||Handling completed with error: Connection has been closed
2019-03-06T16:10:38,269|ERROR||reactor-http-server-epoll-18||||Unhandled failure: null, response already set (status=200)
2019-03-06T16:10:38,269|WARN ||reactor-http-server-epoll-18||||Handling completed with error: null
2019-03-06T16:10:38,294|ERROR||reactor-http-server-epoll-18||||Unhandled failure: syscall:write(..) failed: 斷開的管道, response already set (status=null)
2019-03-06T16:10:38,294|WARN ||reactor-http-server-epoll-18||||Handling completed with error: syscall:write(..) failed: 斷開的管道
2019-03-06T16:10:38,306|ERROR||reactor-http-server-epoll-23||||Unhandled failure: syscall:write(..) failed: 斷開的管道, response already set (status=null)
2019-03-06T16:10:38,306|WARN ||reactor-http-server-epoll-23||||Handling completed with error: syscall:write(..) failed: 斷開的管道
2019-03-06T16:14:33,397|INFO ||AsyncResolver-bootstrap-executor-0||||Resolving eureka endpoints via configuration

這個問題很嚴重了，因為單個請求的時候，並不會報錯，這個錯誤只發生在高並發壓測下，無法追蹤。最重要的是，我們看到性能只剩下300/s，這是萬萬不能接受的，生產更不能接收。

這個問題很難解釋，因為我們采用的是官方提供的寫法，我們回頭看官方的修改response 類，好吧，不用看了，因為：

package org.springframework.cloud.gateway.filter.factory.rewrite;
/**
 * This filter is BETA and may be subject to change in a future release.
 */
public class ModifyResponseBodyGatewayFilterFactory
        extends AbstractGatewayFilterFactory<ModifyResponseBodyGatewayFilterFactory.Config> {

官方已經說了，這是測試版本，不頂用。

不死心，又想起了gateway提供的GlobalFilter，將剛才的代碼寫到全局過濾器中再試試，但是結果相同！

涼涼...

跪求結論跟我不同的啟發文檔，或者只能等下一版本了。

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 spring-cloud-gateway報錯 spring-cloud-gateway限流 spring-cloud-gateway降級 Spring-Cloud-Gateway 基礎（一） spring-cloud-gateway 服務網關 spring-cloud-gateway動態路由 spring-cloud-gateway動態路由 spring-cloud-gateway聚合swagger文檔 spring-cloud-gateway靜態路由 spring-cloud-gateway負載普通web項目