load-balancing - 具有WebClient的功能区负载平衡器与其余模板不同(未正确平衡)

Question

我尝试将WebClient与LoadBalancerExchangeFilterFunction结合使用:
WebClient配置:

@Bean
public WebClient myWebClient(final LoadBalancerExchangeFilterFunction lbFunction) {
    return WebClient.builder()
            .filter(lbFunction)
            .defaultHeader(ACCEPT, APPLICATION_JSON_VALUE)
            .defaultHeader(CONTENT_ENCODING, APPLICATION_JSON_VALUE)
            .build();
} 

然后，我注意到对基础服务的调用没有适本地实现负载平衡-每个实例的RPS始终存在差异。

然后，我尝试回到 RestTemplate。而且工作正常。

配置 RestTemplate:

private static final int CONNECT_TIMEOUT_MILLIS = 18 * DateTimeConstants.MILLIS_PER_SECOND;
private static final int READ_TIMEOUT_MILLIS = 18 * DateTimeConstants.MILLIS_PER_SECOND;

@LoadBalanced
@Bean
public RestTemplate restTemplateSearch(final RestTemplateBuilder restTemplateBuilder) {
    return restTemplateBuilder
            .errorHandler(errorHandlerSearch())
            .requestFactory(this::bufferedClientHttpRequestFactory)
            .build();
}

private ClientHttpRequestFactory bufferedClientHttpRequestFactory() {
    final SimpleClientHttpRequestFactory requestFactory = new SimpleClientHttpRequestFactory();
    requestFactory.setConnectTimeout(CONNECT_TIMEOUT_MILLIS);
    requestFactory.setReadTimeout(READ_TIMEOUT_MILLIS);
    return new BufferingClientHttpRequestFactory(requestFactory);
}

private ResponseErrorHandler errorHandlerSearch() {
    return new DefaultResponseErrorHandler() {
        @Override
        public boolean hasError(ClientHttpResponse response) throws IOException {
            return response.getStatusCode().is5xxServerError();
        }
    };
}

最高使用11:25的 WebClient配置进行负载平衡，然后切换回 RestTemplate:



有原因为何存在差异，为什么我可以使用 WebClient在每个实例上使用相同数量的RPS？提示可能是较旧的实例比新的实例收到更多的请求。

我尝试了一些调试，并调用了相同的(默认值为 ZoneAwareLoadBalancer)逻辑。

Answer 1

我做了简单的POC，所有工作都与Web客户端和用于默认配置的rest模板完全相同。

其余服务器代码:

@SpringBootApplication
internal class RestServerApplication

fun main(args: Array<String>) {
    runApplication<RestServerApplication>(*args)
}

class BeansInitializer : ApplicationContextInitializer<GenericApplicationContext> {
    override fun initialize(context: GenericApplicationContext) {
        serverBeans().initialize(context)
    }
}

fun serverBeans() = beans {
    bean("serverRoutes") {
        PingRoutes(ref()).router()
    }
    bean<PingHandler>()
}

internal class PingRoutes(private val pingHandler: PingHandler) {
    fun router() = router {
        GET("/api/ping", pingHandler::ping)
    }
}

class PingHandler(private val env: Environment) {
    fun ping(serverRequest: ServerRequest): Mono<ServerResponse> {
        return Mono
            .fromCallable {
                // sleap added to simulate some work
                Thread.sleep(2000)
            }
            .subscribeOn(elastic())
            .flatMap {
                ServerResponse.ok()
                    .syncBody("pong-${env["HOSTNAME"]}-${env["server.port"]}")
            }
    }
}

在 application.yaml 中添加:

context.initializer.classes: com.lbpoc.server.BeansInitializer

Gradle中的依赖关系:

implementation('org.springframework.boot:spring-boot-starter-webflux')

其余客户端代码:

@SpringBootApplication
internal class RestClientApplication {
    @Bean
    @LoadBalanced
    fun webClientBuilder(): WebClient.Builder {
        return WebClient.builder()
    }

    @Bean
    @LoadBalanced
    fun restTemplate() = RestTemplateBuilder().build()
}

fun main(args: Array<String>) {
    runApplication<RestClientApplication>(*args)
}

class BeansInitializer : ApplicationContextInitializer<GenericApplicationContext> {
    override fun initialize(context: GenericApplicationContext) {
        clientBeans().initialize(context)
    }
}

fun clientBeans() = beans {
    bean("clientRoutes") {
        PingRoutes(ref()).router()
    }
    bean<PingHandlerWithWebClient>()
    bean<PingHandlerWithRestTemplate>()
}

internal class PingRoutes(private val pingHandlerWithWebClient: PingHandlerWithWebClient) {
    fun router() = org.springframework.web.reactive.function.server.router {
        GET("/api/ping", pingHandlerWithWebClient::ping)
    }
}

class PingHandlerWithWebClient(private val webClientBuilder: WebClient.Builder) {
    fun ping(serverRequest: ServerRequest) = webClientBuilder.build()
        .get()
        .uri("http://rest-server-poc/api/ping")
        .retrieve()
        .bodyToMono(String::class.java)
        .onErrorReturn(TimeoutException::class.java, "Read/write timeout")
        .flatMap {
            ServerResponse.ok().syncBody(it)
        }
}

class PingHandlerWithRestTemplate(private val restTemplate: RestTemplate) {
    fun ping(serverRequest: ServerRequest) = Mono.fromCallable {
        restTemplate.getForEntity("http://rest-server-poc/api/ping", String::class.java)
    }.flatMap {
        ServerResponse.ok().syncBody(it.body!!)
    }
}

在 application.yaml 中添加:

context.initializer.classes: com.lbpoc.client.BeansInitializer
spring:
  application:
    name: rest-client-poc-for-load-balancing
logging:
  level.org.springframework.cloud: DEBUG
  level.com.netflix.loadbalancer: DEBUG
rest-server-poc:
  listOfServers: localhost:8081,localhost:8082

Gradle中的依赖关系:

implementation('org.springframework.boot:spring-boot-starter-webflux')
implementation('org.springframework.cloud:spring-cloud-starter-netflix-ribbon')

您可以在服务器的两个或多个实例中尝试使用它，并且与Web客户端和rest模板完全相同。

功能区默认使用zoneAwareLoadBalancer，如果只有一个区域，则服务器的所有实例都将在“未知”区域中注册。

您可能对通过Web客户端保持连接有问题。 Web客户端在多个请求中重用同一连接，其余模板则不这样做。如果您的客户端和服务器之间有某种代理，则您可能会在通过Web客户端重用连接时遇到问题。要验证它，您可以像这样修改Web客户端Bean并运行测试:

@Bean
@LoadBalanced
fun webClientBuilder(): WebClient.Builder {
    return WebClient.builder()
        .clientConnector(ReactorClientHttpConnector { options ->
            options
                .compression(true)
                .afterNettyContextInit { ctx ->
                    ctx.markPersistent(false)
                }
        })
}

当然，这不是用于生产的好解决方案，但是这样做可以检查客户端应用程序内部的配置是否有问题，或者是外部问题(客户端和服务器之间存在问题)。例如。如果您正在使用kubernetes并使用服务器节点IP地址在服务发现中注册您的服务，则对该服务的每次调用都将通过kube-proxy负载均衡器进行，并且(默认情况下将使用轮询)路由到该服务的某个Pod 。