|
|
|
|
@ -2,10 +2,13 @@ package com.gitee.sop.gatewaycommon.route; |
|
|
|
|
|
|
|
|
|
import com.gitee.sop.gatewaycommon.bean.InstanceDefinition; |
|
|
|
|
import com.gitee.sop.gatewaycommon.loadbalancer.EurekaEnvironmentServerChooser; |
|
|
|
|
import com.gitee.sop.gatewaycommon.loadbalancer.SopPropertiesFactory; |
|
|
|
|
import com.gitee.sop.gatewaycommon.manager.EnvironmentKeys; |
|
|
|
|
import com.netflix.appinfo.InstanceInfo; |
|
|
|
|
import com.netflix.discovery.shared.Application; |
|
|
|
|
import com.netflix.discovery.shared.Applications; |
|
|
|
|
import lombok.AllArgsConstructor; |
|
|
|
|
import lombok.Data; |
|
|
|
|
import org.apache.commons.collections.CollectionUtils; |
|
|
|
|
import org.springframework.cloud.netflix.eureka.CloudEurekaClient; |
|
|
|
|
import org.springframework.context.ApplicationEvent; |
|
|
|
|
@ -27,50 +30,77 @@ public class EurekaRegistryListener extends BaseRegistryListener { |
|
|
|
|
System.setProperty(EnvironmentKeys.ZUUL_CUSTOM_RULE_CLASSNAME.getKey(), EurekaEnvironmentServerChooser.class.getName()); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
private Set<String> cacheServices = new HashSet<>(); |
|
|
|
|
private Set<ServiceHolder> cacheServices = new HashSet<>(); |
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* 注册中心触发事件，可以从中获取服务<br> |
|
|
|
|
* |
|
|
|
|
* 这个方法做的事情有2个：<br> |
|
|
|
|
* |
|
|
|
|
* 1. 找出新注册的服务，调用pullRoutes方法<br> |
|
|
|
|
* 2. 找出删除的服务，调用removeRoutes方法<br> |
|
|
|
|
* |
|
|
|
|
* @param applicationEvent 事件体 |
|
|
|
|
*/ |
|
|
|
|
@Override |
|
|
|
|
public void onEvent(ApplicationEvent applicationEvent) { |
|
|
|
|
Object source = applicationEvent.getSource(); |
|
|
|
|
CloudEurekaClient cloudEurekaClient = (CloudEurekaClient) source; |
|
|
|
|
Applications applications = cloudEurekaClient.getApplications(); |
|
|
|
|
List<Application> registeredApplications = applications.getRegisteredApplications(); |
|
|
|
|
List<String> serviceList = registeredApplications |
|
|
|
|
|
|
|
|
|
List<ServiceHolder> serviceList = registeredApplications |
|
|
|
|
.stream() |
|
|
|
|
.map(Application::getName) |
|
|
|
|
.filter(application -> CollectionUtils.isNotEmpty(application.getInstances())) |
|
|
|
|
.map(Application::getInstances) |
|
|
|
|
.map(instanceInfos -> { |
|
|
|
|
// 根据更新时间倒叙
|
|
|
|
|
instanceInfos.sort(Comparator.comparing(InstanceInfo::getLastUpdatedTimestamp).reversed()); |
|
|
|
|
// 获取最新的个服务实例，说明这个服务实例刚刚重启过
|
|
|
|
|
return instanceInfos.get(0); |
|
|
|
|
}) |
|
|
|
|
.map(instanceInfo -> new ServiceHolder(instanceInfo.getAppName(), instanceInfo.getLastUpdatedTimestamp())) |
|
|
|
|
.collect(Collectors.toList()); |
|
|
|
|
|
|
|
|
|
final Set<String> currentServices = new HashSet<>(serviceList); |
|
|
|
|
final Set<ServiceHolder> currentServices = new HashSet<>(serviceList); |
|
|
|
|
currentServices.removeAll(cacheServices); |
|
|
|
|
// 如果有新的服务注册进来
|
|
|
|
|
if (currentServices.size() > 0) { |
|
|
|
|
List<Application> newApplications = registeredApplications.stream() |
|
|
|
|
.filter(application -> this.canOperator(application.getName()) |
|
|
|
|
&& currentServices.contains(application.getName())) |
|
|
|
|
.filter(application -> |
|
|
|
|
this.canOperator(application.getName()) && containsService(currentServices, application.getName())) |
|
|
|
|
.collect(Collectors.toList()); |
|
|
|
|
|
|
|
|
|
this.doRegister(newApplications); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
cacheServices.removeAll(new HashSet<>(serviceList)); |
|
|
|
|
Set<String> removedServiceIdList = getRemovedServiceId(serviceList); |
|
|
|
|
// 如果有服务删除
|
|
|
|
|
if (cacheServices.size() > 0) { |
|
|
|
|
this.doRemove(cacheServices); |
|
|
|
|
if (removedServiceIdList.size() > 0) { |
|
|
|
|
this.doRemove(removedServiceIdList); |
|
|
|
|
} |
|
|
|
|
// 缓存最新服务
|
|
|
|
|
|
|
|
|
|
cacheServices = new HashSet<>(serviceList); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* 获取已经下线的serviceId |
|
|
|
|
* |
|
|
|
|
* @param serviceList 最新的serviceId集合 |
|
|
|
|
* @return 返回已下线的serviceId |
|
|
|
|
*/ |
|
|
|
|
private Set<String> getRemovedServiceId(List<ServiceHolder> serviceList) { |
|
|
|
|
Set<String> cache = cacheServices.stream() |
|
|
|
|
.map(ServiceHolder::getServiceId) |
|
|
|
|
.collect(Collectors.toSet()); |
|
|
|
|
|
|
|
|
|
Set<String> newList = serviceList.stream() |
|
|
|
|
.map(ServiceHolder::getServiceId) |
|
|
|
|
.collect(Collectors.toSet()); |
|
|
|
|
|
|
|
|
|
cache.removeAll(newList); |
|
|
|
|
return cache; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
private static boolean containsService(Set<ServiceHolder> currentServices, String serviceId) { |
|
|
|
|
for (ServiceHolder currentService : currentServices) { |
|
|
|
|
if (currentService.getServiceId().equalsIgnoreCase(serviceId)) { |
|
|
|
|
return true; |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
return false; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
private void doRegister(List<Application> registeredApplications) { |
|
|
|
|
registeredApplications.forEach(application -> { |
|
|
|
|
List<InstanceInfo> instances = application.getInstances(); |
|
|
|
|
@ -92,4 +122,10 @@ public class EurekaRegistryListener extends BaseRegistryListener { |
|
|
|
|
deletedServices.forEach(this::removeRoutes); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@Data |
|
|
|
|
@AllArgsConstructor |
|
|
|
|
private static class ServiceHolder { |
|
|
|
|
private String serviceId; |
|
|
|
|
private long lastUpdatedTimestamp; |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
tanghc
5 years ago