项目用spring-boot-starter-webflux 实现了es6和es7的兼容层网关，但只是做可行性验证
真正生产使用的网关还需要更多功能，最基本的，限流，熔断，负载均衡，这些都独立往项目里加，每个点都需要花精力整合
实际这方面应用已经有了很成熟的方案，服务治理相关，k8s,ingress,istio,kong,nginx...
但对es这类bare服务，套用云服务的方案并不合适，kong/nginx因为是c+lua的技术栈，定制的成本较高
nodejs已经过气了，go和java，考虑生态，选择java
java方面的网关，很少单独提及，更多是做为java服务治理的一个组件来使用
java类网关，早些年大家都使用netty/mina原生实现
但后期有基于netty的各种封装好的http框架,再完全用netty开发http类网关就比较少见了，当然tcp/udp类，自定义rpc的还是免不了直接和netty打交道
整合netty的 http高性能网关类服务早些年个人用过 jersey，后来用vert.x,再后来就直接上spring-boot-starter-webflux了，抽空把历史代码扒出来
现在为了省去在webflux自行添加限流，熔断，降级等功能，直接使用spring-boot-gateway
实际spring-boot-gateway，本身集成mvn/webflux的两种方案，webflux的底层就是netty
本地代理其实可以理解为替代nginx,并实现一些和业务产品深度结合的功能
因为nginx c+lua的技术栈和开发成本较高
首先spring-cloud-gateway 原生是 spring-cloud的组件，应用场景和spring-cloud深耦合
例如，loadbanlance，依赖spring-cloud的服务发现组件，consule,nacos等
https://docs.spring.io/spring-cloud-commons/docs/current/reference/html/#spring-cloud-loadbalancer
Spring Cloud Commons provides the @EnableDiscoveryClient annotation. This looks for implementations of the DiscoveryClient and ReactiveDiscoveryClient interfaces with META-INF/spring.factories. Implementations of the discovery client add a configuration class to spring.factories under the org.springframework.cloud.client.discovery.EnableDiscoveryClient key. Examples of DiscoveryClient implementations include Spring Cloud Netflix Eureka, Spring Cloud Consul Discovery, and Spring Cloud Zookeeper Discovery.
spring-cloud-gateway 基础支持
spring:
cloud:
gateway:
routes:
- id: before_route
uri: https://example.org
predicates:
- Before=2017-01-20T17:42:47.789-07:00[America/Denver]
以该项为例，关键是uri 这个参数
网关单点1:1 uri 可以写一个http/https的访问地址
如果需要实现负载均衡gateway: server 1:n 则需要实现 uri("lb://backing-service:8088")
@Bean
public RouteLocator routes(RouteLocatorBuilder builder) {
return builder.routes()
.route("circuitbreaker_route", r -> r.path("/consumingServiceEndpoint")
.filters(f -> f.circuitBreaker(c -> c.name("myCircuitBreaker").fallbackUri("forward:/inCaseOfFailureUseThis").addStatusCode("INTERNAL_SERVER_ERROR"))
.rewritePath("/consumingServiceEndpoint", "/backingServiceEndpoint")).uri("lb://backing-service:8088")
.build();
}
es3
这里实际依赖了spring-cloud生态的服务发现组件，注册服务 backing-service 至注册中心，spring-cloud从注册中心获取真实的服务地址host:port,再通过客户端负载均衡lb 路由至真实服务
这是服务治理的基本原理流程
但是对目前的场景不适用，目前的场景，基本可以理解为把spring-cloud-gateway 当nginx用
路由到后端静态的几个地址即可
以es为例 3个client节点
es-client-01 192.168.10.11:9200
es-client-02 192.168.10.12:9200
es-client-03 192.168.10.13:9200
并不存在一个有效的注册中心，实际多一个注册中心组件，项目的复杂度就更高了，对es 这类服务组件，并不需要和spring的生态完全结合
我们需要定制lb://的解析，即，使lb:// 不通过注册中心，而是完全静态配置在本地（先搞静态吧，以后再考虑搞动态，此动态非彼动态）
需要注意withHealthChecks() 顾名思义 添加了withHealthChecks() 则会对 后端server进行健康检查，检查方式为验证对应的后端server /actuator/health 是否可达。这个路径是spring-boot/spring-cloud 组件的默认路径，但后端的es服务，这个路径并不可达。.withHealthChecks() 会返回Service Unavailable
{"timestamp":"2021-06-02T09:04:00.596+00:00","path":"/","status":503,"error":"Service Unavailable","requestId":"d3d01e2e-1"}
@Bean
public ServiceInstanceListSupplier discoveryClientServiceInstanceListSupplier(
ConfigurableApplicationContext context) {
return ServiceInstanceListSupplier.builder()
.withBase(new CustomServiceInstanceListSupplier())
// .withHealthChecks()
.build(context);
}
健康检查部分代码,关键注意/actuator/health
package org.springframework.cloud.loadbalancer.core;
public class HealthCheckServiceInstanceListSupplier extends DelegatingServiceInstanceListSupplier implements InitializingBean, DisposableBean {
private static final Log LOG = LogFactory.getLog(HealthCheckServiceInstanceListSupplier.class);
private final HealthCheck healthCheck;
private final String defaultHealthCheckPath;
private final Flux<List<ServiceInstance>> aliveInstancesReplay;
private Disposable healthCheckDisposable;
private final BiFunction<ServiceInstance, String, Mono<Boolean>> aliveFunction;
public HealthCheckServiceInstanceListSupplier(ServiceInstanceListSupplier delegate, HealthCheck healthCheck, BiFunction<ServiceInstance, String, Mono<Boolean>> aliveFunction) {
super(delegate);
this.defaultHealthCheckPath = (String)healthCheck.getPath().getOrDefault("default", "/actuator/health");
this.aliveFunction = aliveFunction;
this.healthCheck = healthCheck;
Repeat<Object> aliveInstancesReplayRepeat = Repeat.onlyIf((repeatContext) -> {
return this.healthCheck.getRefetchInstances();
}).fixedBackoff(healthCheck.getRefetchInstancesInterval());
Flux<List<ServiceInstance>> aliveInstancesFlux = Flux.defer(delegate).repeatWhen(aliveInstancesReplayRepeat).switchMap((serviceInstances) -> {
return this.healthCheckFlux(serviceInstances).map((alive) -> {
return Collections.unmodifiableList(new ArrayList(alive));
});
});
this.aliveInstancesReplay = aliveInstancesFlux.delaySubscription(healthCheck.getInitialDelay()).replay(1).refCount(1);
}
}
我们先不采用withHealthChecks，等后期有时间再自定义或配置withHealthChecks实现
package org.springframework.cloud.loadbalancer.core;
public class DiscoveryClientServiceInstanceListSupplier implements ServiceInstanceListSupplier {
public static final String SERVICE_DISCOVERY_TIMEOUT = "spring.cloud.loadbalancer.service-discovery.timeout";
private static final Log LOG = LogFactory.getLog(DiscoveryClientServiceInstanceListSupplier.class);
private Duration timeout = Duration.ofSeconds(30L);
private final String serviceId;
private final Flux<List<ServiceInstance>> serviceInstances;
public DiscoveryClientServiceInstanceListSupplier(DiscoveryClient delegate, Environment environment) {
this.serviceId = environment.getProperty("loadbalancer.client.name");
this.resolveTimeout(environment);
this.serviceInstances = Flux.defer(() -> {
return Flux.just(delegate.getInstances(this.serviceId));
}).subscribeOn(Schedulers.boundedElastic()).timeout(this.timeout, Flux.defer(() -> {
this.logTimeout();
return Flux.just(new ArrayList());
})).onErrorResume((error) -> {
this.logException(error);
return Flux.just(new ArrayList());
});
}
public DiscoveryClientServiceInstanceListSupplier(ReactiveDiscoveryClient delegate, Environment environment) {
this.serviceId = environment.getProperty("loadbalancer.client.name");
this.resolveTimeout(environment);
this.serviceInstances = Flux.defer(() -> {
return delegate.getInstances(this.serviceId).collectList().flux().timeout(this.timeout, Flux.defer(() -> {
this.logTimeout();
return Flux.just(new ArrayList());
})).onErrorResume((error) -> {
this.logException(error);
return Flux.just(new ArrayList());
});
});
}
public String getServiceId() {
return this.serviceId;
}
public Flux<List<ServiceInstance>> get() {
return this.serviceInstances;
}
private void resolveTimeout(Environment environment) {
String providedTimeout = environment.getProperty("spring.cloud.loadbalancer.service-discovery.timeout");
if (providedTimeout != null) {
this.timeout = DurationStyle.detectAndParse(providedTimeout);
}
}
private void logTimeout() {
if (LOG.isDebugEnabled()) {
LOG.debug(String.format("Timeout occurred while retrieving instances for service %s.The instances could not be retrieved during %s", this.serviceId, this.timeout));
}
}
private void logException(Throwable error) {
LOG.error(String.format("Exception occurred while retrieving instances for service %s", this.serviceId), error);
}
}