tanghc 5 years ago
parent b03e908173
commit 1532f3addc
  1. 75
      sop-common/sop-gateway-common/src/main/java/com/gitee/sop/gatewaycommon/route/EurekaRegistryListener.java

@ -6,6 +6,8 @@ import com.gitee.sop.gatewaycommon.manager.EnvironmentKeys;
import com.netflix.appinfo.InstanceInfo; import com.netflix.appinfo.InstanceInfo;
import com.netflix.discovery.shared.Application; import com.netflix.discovery.shared.Application;
import com.netflix.discovery.shared.Applications; import com.netflix.discovery.shared.Applications;
import lombok.AllArgsConstructor;
import lombok.Data;
import org.apache.commons.collections.CollectionUtils; import org.apache.commons.collections.CollectionUtils;
import org.springframework.cloud.netflix.eureka.CloudEurekaClient; import org.springframework.cloud.netflix.eureka.CloudEurekaClient;
import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEvent;
@ -27,50 +29,77 @@ public class EurekaRegistryListener extends BaseRegistryListener {
System.setProperty(EnvironmentKeys.ZUUL_CUSTOM_RULE_CLASSNAME.getKey(), EurekaEnvironmentServerChooser.class.getName()); 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 @Override
public void onEvent(ApplicationEvent applicationEvent) { public void onEvent(ApplicationEvent applicationEvent) {
Object source = applicationEvent.getSource(); Object source = applicationEvent.getSource();
CloudEurekaClient cloudEurekaClient = (CloudEurekaClient) source; CloudEurekaClient cloudEurekaClient = (CloudEurekaClient) source;
Applications applications = cloudEurekaClient.getApplications(); Applications applications = cloudEurekaClient.getApplications();
List<Application> registeredApplications = applications.getRegisteredApplications(); List<Application> registeredApplications = applications.getRegisteredApplications();
List<String> serviceList = registeredApplications
List<ServiceHolder> serviceList = registeredApplications
.stream() .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()); .collect(Collectors.toList());
final Set<String> currentServices = new HashSet<>(serviceList); final Set<ServiceHolder> currentServices = new HashSet<>(serviceList);
currentServices.removeAll(cacheServices); currentServices.removeAll(cacheServices);
// 如果有新的服务注册进来 // 如果有新的服务注册进来
if (currentServices.size() > 0) { if (currentServices.size() > 0) {
List<Application> newApplications = registeredApplications.stream() List<Application> newApplications = registeredApplications.stream()
.filter(application -> this.canOperator(application.getName()) .filter(application ->
&& currentServices.contains(application.getName())) this.canOperator(application.getName()) && containsService(currentServices, application.getName()))
.collect(Collectors.toList()); .collect(Collectors.toList());
this.doRegister(newApplications); this.doRegister(newApplications);
} }
cacheServices.removeAll(new HashSet<>(serviceList)); Set<String> removedServiceIdList = getRemovedServiceId(serviceList);
// 如果有服务删除 // 如果有服务删除
if (cacheServices.size() > 0) { if (removedServiceIdList.size() > 0) {
this.doRemove(cacheServices); this.doRemove(removedServiceIdList);
} }
// 缓存最新服务
cacheServices = new HashSet<>(serviceList); 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) { private void doRegister(List<Application> registeredApplications) {
registeredApplications.forEach(application -> { registeredApplications.forEach(application -> {
List<InstanceInfo> instances = application.getInstances(); List<InstanceInfo> instances = application.getInstances();
@ -92,4 +121,10 @@ public class EurekaRegistryListener extends BaseRegistryListener {
deletedServices.forEach(this::removeRoutes); deletedServices.forEach(this::removeRoutes);
} }
@Data
@AllArgsConstructor
private static class ServiceHolder {
private String serviceId;
private long lastUpdatedTimestamp;
}
} }

Loading…
Cancel
Save