springboot 网关实时刷新服务列表实现

springboot 网关实时刷新服务列表实现

2021-11-04 18:46:02
实现原理:每个微服务spring会启动单独的AnnotatioinApplicationContext,该context会自动初始ILoadBalancer等类其中ILoadBalancer包含1.ServerList: ServerList可用于获取实际的eureka中的服务列表2.ServerListUpdater:用于定时调用ServerList的刷新方法获取最新的服务列表更新ILoadBalancer中的Server这里通过 LoadBalancerConfiguration 配置类,在postCons@Configuration@Slf4jpublic class LoadBalancerConfiguration {  @Autowiredprivate SpringClientFactory clientFactory;@Value("${ribbon.eureka.approximateZoneFromHostname:false}")private boolean approximateZoneFromHostname = false;@PostConstructpublic void postCons() {this.clientFactory.setConfigurations(Arrays.asList(new RibbonClientSpecification("default.ILoadBalancer.visiter", new Class<?>[] {LoadBalancerConfiguration.class})));}private Optional<EurekaHttpClient> flectReadEurekaHttpClient(EurekaClient eurekaClient) {if (AopUtils.isAopProxy(eurekaClient)) {try {eurekaClient = (EurekaClient) Target(eurekaClient);} catch (Exception e) {("error get target from aop proxy:{}", e);}}EurekaHttpClient queryClient = null;if (eurekaClient instanceof DiscoveryClient) {DiscoveryClient discoveryClient = (DiscoveryClient) eurekaClient;try {Object eurekaTransport = adField(discoveryClient,"eurekaTransport",true);queryClient =(EurekaHttpClient) adField(eurekaTransport, "queryClient",true);log.info("success flect read EurekaHttpClient:{}", queryClient);} catch (Exception e) {("error flect read EurekaHttpClient:{}", e);}}return Optional.ofNullable(queryClient);}@Bean@Lazypublic ServerList<?> ribbonServerList(IClientConfig config,Provider<EurekaClient> eurekaClientProvider) {EurekaClient eurekaClient = ();Optional<EurekaHttpClient> flectReadEurekaHttpClient = this.flectReadEurekaHttpClient(eurekaClient);if (!flectReadEurekaHttpClient.isPresent()) {log.warn("can not read EurekaHttpClient from this kind eurekaClient:{}", eurekaClient);return null;}CustomerDiscoveryEnabledNIWSServerList discoveryServerList = new CustomerDiscoveryEnabledNIWSServerList(config, eurekaClientProvider);EurekaHttpClient eurekaHttpClient = ();CustomerDomainExtractingServerList serverList = new CustomerDomainExtractingServerList(discoveryServerList, eurekaHttpClient,config,this.approximateZoneFromHostname);return serverList;}}这个方法里，向clientFactory（SpringClientFactory）中设置了默认配置类，用于自定义初始化ServerList的ribbonServerList方法自定义ServerList的目的有:1.获取到EurekaHttpClient,用于将EurekaClient的InstanceInfo的instanceId转回注册EureKaServer成功的InstanceInfo2.通过DiscoveryEnabledNIWSServerList的createServer方法，把EurekaServer获取到的InstanceInfo类型数据转为ILoadBalancer使用的Server所以流程为:1.任一系统启动的时候，监听 spring 启动后的 WebServerInitializedEvent 事件，该事件发布后，当前实例已经注册到 eureka server2.在WebServerInitializedEvent事件的处理方法中，获取ApplicationInfoManager中标识当前eureka客户端信息的InstanceInfo@Autowiredprivate ApplicationInfoManager manager;@Overridepublic String serverId() {InstanceInfo info = Info();InstanceId();}获取到 instanceId3.通过事件广播途径，告知集群当前系统启动完成,实例id为 instanceId如:ServerOnlineEvent:private String serviceName;private String instanceId;4.网关服务监听集群实例变更事件ServerOnlineEvent获取变更服务(serviceName)当前仍然在线的服务列表(instanceId[]),发布ServiceServerListChangedEvent@Data@NoArgsConstructor@AllArgsConstructorpublic class ServiceServerListChangedEvent {private String serviceName;private List<ApplicationInstanceInfoServerId> instanceServerInfoIds;}5.ServiceInstanceInfoChangedListener在网关服务内部监听ServiceServerListChangedEvent通过SpringClientFactory获取对应 serviceName 对应的ILoadBalancer负载器构建LoadBalancerVisitorLoadBalancerVisitor 进行的操作有:1.停用ILoadBalancer默认的ServerListUpdater，完全切换到事件监听上2.调用CustomerDomainExtractingServerList的createServer方法，将 instanceId 转为 具体注册在Eureka Server中的InstanceInfo,并继续由CustomerDiscoveryEnabledNIWSServerList转为负载器内部使用的DiscoveryEnabledServer3.调用ILoadBalancer的setServersList更新服务列表完weike-gateway当前使用akka集群监控服务列表的变更事件附录:1.import javax.inject.Provider;import comflix.appinfo.InstanceInfo;import fig.CommonClientConfigKey;import fig.DefaultClientConfigImpl;import fig.IClientConfig;import comflix.discovery.EurekaClient;import comflix.niws.loadbalancer.DiscoveryEnabledNIWSServerList;import comflix.niws.loadbalancer.DiscoveryEnabledServer;/*** * @say little Boy, don't be sad.* @name Rezar* @time 2021-11-04 11:50:47* @Desc 些年若许,不负芳华.**/public class CustomerDiscoveryEnabledNIWSServerList extends DiscoveryEnabledNIWSServerList {private boolean isSecure;private boolean shouldUseIpAddr;@SuppressWarnings("deprecation")public CustomerDiscoveryEnabledNIWSServerList(IClientConfig clientConfig,Provider<EurekaClient> eurekaClientProvider) {super(clientConfig, eurekaClientProvider);this.isSecure = Boolean.parseBoolean("" + Property(CommonClientConfigKey.IsSecure, "false"));this.shouldUseIpAddr = PropertyAsBoolean(CommonClientConfigKey.UseIPAddrForServer,DefaultClientConfigImpl.DEFAULT_USEIPADDRESS_FOR_SERVER);}DiscoveryEnabledServer createServer(InstanceInfo instanceInfo) {ateServer(instanceInfo, isSecure, shouldUseIpAddr);}}2.import urrent.TimeUnit;import org.springframework.cloudflix.ribbon.eureka.DomainExtractingServerList;import comflix.appinfo.InstanceInfo;import fig.IClientConfig;import comflix.ansport.EurekaHttpClient;import comflix.niws.loadbalancer.DiscoveryEnabledServer;slf4j.Slf4j;/*** * @say little Boy, don't be sad.* @name Rezar* @time 2021-11-04 12:03:59* @Desc 些年若许,不负芳华.**/@Slf4jpublic class CustomerDomainExtractingServerList extends DomainExtractingServerList {private CustomerDiscoveryEnabledNIWSServerList serverList;private EurekaHttpClient queryClient;public CustomerDomainExtractingServerList(CustomerDiscoveryEnabledNIWSServerList list, EurekaHttpClient registrationClient,IClientConfig clientConfig,boolean approximateZoneFromHostname) {super(list, clientConfig, approximateZoneFromHostname);this.serverList = list;this.queryClient = registrationClient;}DiscoveryEnabledServer createServer(String instanceId) {if (queryClient == null) {return null;}try {log.info("try getInstance by instanceId:{}", instanceId);InstanceInfo instanceInfo = this.instanceInfo(instanceId, 0);log.info("for instanceId:{} and instanceInfo:{}", instanceId, instanceInfo);return ateServer(instanceInfo);} catch (Exception e) {("error createServer:{}", e);}return null;}private InstanceInfo instanceInfo(String instanceId,int curTimes) {if (curTimes >= 3) {return null;}try {return Instance(instanceId).getEntity();} catch (Exception e) {("error getInstance:{}", e);}try {TimeUnit.SECONDS.sleep(3);} catch (InterruptedException e) {log.warn("InterruptedException:{}", e);}return instanceInfo(instanceId, curTimes + 1);}}3.import java.util.ArrayList;import java.util.HashSet;import java.util.List;import java.util.Set;import java.util.function.Function;import java.util.stream.Collectors;import comflix.loadbalancer.ServerListFilter;import comflix.loadbalancer.ZoneAwareLoadBalancer;import comflix.niws.loadbalancer.DiscoveryEnabledServer;weike.gateway.infa.serverRealTimeUpdate.vo.ApplicationInstanceInfoServerId;import lombok.Getter;slf4j.Slf4j;/*** * @say little Boy, don't be sad.* @name Rezar* @time 2021-11-03 03:10:59* @Desc 些年若许,不负芳华.**/@Slf4jpublic class LoadBalancerVisitor {@Getterprivate String serviceId;private ZoneAwareLoadBalancer<DiscoveryEnabledServer> balancer;private Function<ApplicationInstanceInfoServerId, DiscoveryEnabledServer> serverApply;private Set<DiscoveryEnabledServer> allOnlineInstanceServer = new HashSet<>();public LoadBalancerVisitor(ZoneAwareLoadBalancer<DiscoveryEnabledServer> balancer) {balancer.setFilter(Name(),Filter()));// 停用刷新balancer.stopServerListRefreshing();this.serviceId = Name();this.balancer = balancer;CustomerDomainExtractingServerList serverList = (CustomerDomainExtractingServerList) ServerListImpl();this.serverApply = serverId -> InstanceInfoServerId());log.info("visitor configed for service name:{}",Name());}public synchronized void serverListChanged(List<ApplicationInstanceInfoServerId> onlineServerList) {this.allOnlineInstanceServer = onlineServerList.stream().map(this.serverApply::apply).Set());this.balancer.setServersList(new ArrayList<>(this.allOnlineInstanceServer));log.info("service:{} online server list:{}",this.serviceId, allOnlineInstanceServer);}private ServerListFilter<DiscoveryEnabledServer> buildCustomerVisitableFilter(String clientName,ServerListFilter<DiscoveryEnabledServer> filter) {return new ServerListFilter<DiscoveryEnabledServer>() {@Overridepublic List<DiscoveryEnabledServer> getFilteredListOfServers(List<DiscoveryEnabledServer> servers) {log.info("ServerListFilter clientName:{}-{}", clientName, allOnlineInstanceServer);FilteredListOfServers(new ArrayList<>(allOnlineInstanceServer));}};} }4.import java.util.List;import java.util.Map;import java.util.Optional;import urrent.ConcurrentHashMap;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.cloudflix.ribbon.SpringClientFactory;import t.event.EventListener;import org.springframework.stereotype.Component;import comflix.loadbalancer.ZoneAwareLoadBalancer;import comflix.niws.loadbalancer.DiscoveryEnabledServer;weike.gateway.infa.serverRealTimeUpdate.vo.ApplicationInstanceInfoServerId;weike.gateway.infa.serverRealTimeUpdate.vo.ServiceServerListChangedEvent;slf4j.Slf4j;/*** * @say little Boy, don't be sad.* @name Rezar* @time 2021-11-03 03:57:11* @Desc 些年若许,不负芳华.**/@Component@Slf4jpublic class ServiceInstanceInfoChangedListener {@Autowiredprivate SpringClientFactory clientFactory;private final Map<String, LoadBalancerVisitor> visitorMap = new ConcurrentHashMap<>();@SuppressWarnings("unchecked")private void configLoadBalancer(Object bean) {if (bean instanceof ZoneAwareLoadBalancer) {ZoneAwareLoadBalancer<DiscoveryEnabledServer> tmp = (ZoneAwareLoadBalancer<DiscoveryEnabledServer>) bean;if (Name())) {return;}LoadBalancerVisitor visitor = new LoadBalancerVisitor(tmp);this.visitorMap.ServiceId(), visitor);}}public synchronized Optional<LoadBalancerVisitor> getBalancer(String serverId) {if (!ainsKey(serverId)) {figLoadBalancer(LoadBalancer(serverId));}return Optional.ofNullable((serverId));}@EventListenerpublic void onChanged(ServiceServerListChangedEvent event) {log.info("ServiceServerListChangedEvent:{}", event);String serviceId = ServiceName();List<ApplicationInstanceInfoServerId> onlineServerList = InstanceServerInfoIds();Balancer(serviceId).ifPresent(visitor -> {visitor.serverListChanged(onlineServerList);});}}