序
之前说过，要把一个SpringBoot应用注册到Eureka Server或者是从Eureka Server上获取服务列表，主要做了一下两件事：
1.在应用主类中配置@EnableDiscoveryClient注解。
2.在application.yml中配置eureka.client.service-url.defaultZone指定注册中心的位置。
关于@EnableDiscoveryClient注解在上篇文章中已经介绍过了，今天来看一看客户端是怎么获取到注册中心的位置的。
getDiscoveryServiceUrls方法
获取注册中心的方法是DiscoveryClient的getDiscoveryServiceUrls方法。先看一下这个方法的代码。
/**
* @deprecated see replacement in {@link com.netflix.discovery.endpoint.EndpointUtils}
*
* Get the list of all eureka service urls from properties file for the eureka client to talk to.
*
* @param instanceZone The zone in which the client resides
* @param preferSameZone true if we have to prefer the same zone as the client, false otherwise
* @return The list of all eureka service urls for the eureka client to talk to
*/
@Deprecated
@Override
public List<String> getServiceUrlsFromConfig(String instanceZone, boolean preferSameZone) {
return EndpointUtils.getServiceUrlsFromConfig(clientConfig, instanceZone, preferSameZone);
}
可以看到这个方法已经不再被使用的，而是link到了EndpointUtils的getServiceUrlsFromConfig方法。那么来看一下getServiceUrlsFromConfig的代码。
/**
* Get the list of all eureka service urls from properties file for the eureka client to talk to.
*
* @param clientConfig the clientConfig to use
* @param instanceZone The zone in which the client resides
* @param preferSameZone true if we have to prefer the same zone as the client, false otherwise
* @return The list of all eureka service urls for the eureka client to talk to
*/
public static List<String> getServiceUrlsFromConfig(EurekaClientConfig clientConfig, String instanceZone, boolean preferSameZone) {
List<String> orderedUrls = new ArrayList<String>();
String region = getRegion(clientConfig);
String[] availZones = clientConfig.getAvailabilityZones(clientConfig.getRegion());
if (availZones == null || availZones.length == 0) {
availZones = new String[1];
availZones[0] = DEFAULT_ZONE;
}
logger.debug("The availability zone for the given region {} are {}", region, availZones);
int myZoneOffset = getZoneOffset(instanceZone, preferSameZone, availZones);
List<String> serviceUrls = clientConfig.getEurekaServerServiceUrls(availZones[myZoneOffset]);
if (serviceUrls != null) {
orderedUrls.addAll(serviceUrls);
}
int currentOffset = myZoneOffset == (availZones.length - 1) ? 0 : (myZoneOffset + 1);
while (currentOffset != myZoneOffset) {
serviceUrls = clientConfig.getEurekaServerServiceUrls(availZones[currentOffset]);
if (serviceUrls != null) {
orderedUrls.addAll(serviceUrls);
}
if (currentOffset == (availZones.length - 1)) {
currentOffset = 0;
} else {
currentOffset++;
}
}
if (orderedUrls.size() < 1) {
throw new IllegalArgumentException("DiscoveryClient: invalid serviceUrl specified!");
}
return orderedUrls;
}
先看一下这个方法的入参。clientConfig包含了一些默认配置以及我们在application.yml里面添加或修改的配置。如果我们在application.yml里配置了region，region下面配置了多个zone的话，那么instanceZone就默认取region下配置的第一个zone。配置文件如下：
server:
port: 8882
spring:
application:
name: product-service
eureka:
client:
prefer-same-zone-eureka: true
region: beijing
availability-zones:
beijing: zone-2,zone-1
service-url:
zone-1: http://peer1:8880/eureka/
zone-2: http://peer2:8881/eureka/
如果我们没有自己配置region，那么instanceZone默认为“defaultZone”，这时我们在service-url后面需要配置为defaultZone，如果配置为其他名称的话会找不到注册中心的url。配置文件如下：
server:
port: 8882
spring:
application:
name: product-service
eureka:
client:
prefer-same-zone-eureka: true
# region: beijing
# availability-zones:
# beijing: zone-2,zone-1
service-url:
defaultZone: http://peer1:8880/eureka/,http://peer2:8881/eureka/
# zone-1: http://peer1:8880/eureka/
# zone-2: http://peer2:8881/eureka/
关于region和zone的说明可以参考https://www.cnblogs.com/junjiang3/p/9061867.html
最后来看preferSameZone这个入参。preferSameZone对应的是配置文件中的 prefer-same-zone-eureka，为true时优先选择client实例所在的zone，一般默认为true。
如果prefer-same-zone-eureka为false，按照service-url下的 list取第一个注册中心来注册，并和其维持心跳检测。不会再向list内的其它的注册中心注册和维持心跳。只有在第一个注册失败的情况下，才会依次向其它的注册中心注册，总共重试3次，如果3个service-url都没有注册成功，则注册失败。每隔一个心跳时间，会再次尝试。
如果prefer-same-zone-eureka为true，先通过region取availability-zones内的第一个zone，然后通过这个zone取service-url下的list，并向list内的第一个注册中心进行注册和维持心跳，不会再向list内的其它的注册中心注册和维持心跳。只有在第一个注册失败的情况下，才会依次向其它的注册中心注册，总共重试3次，如果3个service-url都没有注册成功，则注册失败。每隔一个心跳时间，会再次尝试。
继续看getServiceUrlsFromConfig方法。
List<String> orderedUrls = new ArrayList<String>();
String region = getRegion(clientConfig);
String[] availZones = clientConfig.getAvailabilityZones(clientConfig.getRegion());
if (availZones == null || availZones.length == 0) {
availZones = new String[1];
availZones[0] = DEFAULT_ZONE;
}
logger.debug("The availability zone for the given region {} are {}", region, availZones);
int myZoneOffset = getZoneOffset(instanceZone, preferSameZone, availZones);
List<String> serviceUrls = clientConfig.getEurekaServerServiceUrls(availZones[myZoneOffset]);
if (serviceUrls != null) {
orderedUrls.addAll(serviceUrls);
}
这部分的逻辑是先从clientConfig里获取region，然后或者这个region里面的zone，通过getAvailabilityZones方法返回一个数组availZones 。如果availZones 为空或null的话，就默认一个值“default”。但是availZones 其实并不会为空或者null。可以看一下getAvailabilityZones方法的代码（有两个实现类，这里看EurekaClientConfigBean里的实现方法）。
public String[] getAvailabilityZones(String region) {
String value = (String)this.availabilityZones.get(region);
if (value == null) {
value = "defaultZone";
}
return value.split(",");
}
可以看到，如果我们没有在region下面配置zone的话，那么返回的String数组里默认一个值“defaultZone”。
接下来看getZoneOffset方法。
private static int getZoneOffset(String myZone, boolean preferSameZone, String[] availZones) {
for(int i = 0; i < availZones.length; ++i) {
if (myZone != null && availZones[i].equalsIgnoreCase(myZone.trim()) == preferSameZone) {
return i;
}
}
logger.warn("DISCOVERY: Could not pick a zone based on preferred zone settings. My zone - {}, preferSameZone - {}. Defaulting to {}", new Object[]{myZone, preferSameZone, availZones[0]});
return 0;
}
当preferSameZone为true是，返回myZone，也就是region下面第一个zone在service-url里面的index。当preferSameZone为false时，返回region下面第一个非myZone元素在service-url里面的index。
然后就是根据这个index获取service-url了。
但是getServiceUrlsFromConfig方法返回的是所有service-url的集合，刚才我们只是获取了一个url，获取所有url的方法可以看以下代码。
int currentOffset = myZoneOffset == (availZones.length - 1) ? 0 : (myZoneOffset + 1);
while (currentOffset != myZoneOffset) {
serviceUrls = clientConfig.getEurekaServerServiceUrls(availZones[currentOffset]);
if (serviceUrls != null) {
orderedUrls.addAll(serviceUrls);
}
if (currentOffset == (availZones.length - 1)) {
currentOffset = 0;
} else {
currentOffset++;
}
}
这段代码的意思是，判断我们前面得到的那个zone是否是region下面所有zone集合的最后一个元素，如果是的话，给currentOffset赋值为0；如果不是的话，给currentOffset赋值为zone数组中下一个元素的index。后面进入while循环，跳出的条件是currentOffset==myZoneOffset。如果currentOffset到了zone数组的最后一位，那么给currentOffset赋值0，从头开始遍历zone数组，直到currentOffset==myZoneOffset。这样我们就获得了所有的url。
关于断点调试
要理清getServiceUrlsFromConfig方法的逻辑，最好的办法还是打断点进去一步一步跟代码。但是直接在getServiceUrlsFromConfig里面打断点，启动项目的时候不会走进去。但是在getServiceUrlsFromConfig下面有一个getServiceUrlsMapFromConfig方法，这个方法的逻辑和getServiceUrlsFromConfig完全一样，唯一不同的是返回值是Map而不是List。在项目启动初始化的时候会进入这个方法。为什么会走到getServiceUrlsMapFromConfig方法里呢？回到DiscoveryClient这个类里面。当DiscoveryClient初始化时，它的构造方法如下：
@Inject
DiscoveryClient(ApplicationInfoManager applicationInfoManager, EurekaClientConfig config, AbstractDiscoveryClientOptionalArgs args,
Provider<BackupRegistry> backupRegistryProvider, EndpointRandomizer endpointRandomizer) {
if (args != null) {
this.healthCheckHandlerProvider = args.healthCheckHandlerProvider;
this.healthCheckCallbackProvider = args.healthCheckCallbackProvider;
this.eventListeners.addAll(args.getEventListeners());
this.preRegistrationHandler = args.preRegistrationHandler;
} else {
this.healthCheckCallbackProvider = null;
this.healthCheckHandlerProvider = null;
this.preRegistrationHandler = null;
}
this.applicationInfoManager = applicationInfoManager;
InstanceInfo myInfo = applicationInfoManager.getInfo();
clientConfig = config;
staticClientConfig = clientConfig;
transportConfig = config.getTransportConfig();
instanceInfo = myInfo;
if (myInfo != null) {
appPathIdentifier = instanceInfo.getAppName() + "/" + instanceInfo.getId();
} else {
logger.warn("Setting instanceInfo to a passed in null value");
}
this.backupRegistryProvider = backupRegistryProvider;
this.endpointRandomizer = endpointRandomizer;
this.urlRandomizer = new EndpointUtils.InstanceInfoBasedUrlRandomizer(instanceInfo);
localRegionApps.set(new Applications());
fetchRegistryGeneration = new AtomicLong(0);
remoteRegionsToFetch = new AtomicReference<String>(clientConfig.fetchRegistryForRemoteRegions());
remoteRegionsRef = new AtomicReference<>(remoteRegionsToFetch.get() == null ? null : remoteRegionsToFetch.get().split(","));
if (config.shouldFetchRegistry()) {
this.registryStalenessMonitor = new ThresholdLevelsMetric(this, METRIC_REGISTRY_PREFIX + "lastUpdateSec_", new long[]{15L, 30L, 60L, 120L, 240L, 480L});
} else {
this.registryStalenessMonitor = ThresholdLevelsMetric.NO_OP_METRIC;
}
if (config.shouldRegisterWithEureka()) {
this.heartbeatStalenessMonitor = new ThresholdLevelsMetric(this, METRIC_REGISTRATION_PREFIX + "lastHeartbeatSec_", new long[]{15L, 30L, 60L, 120L, 240L, 480L});
} else {
this.heartbeatStalenessMonitor = ThresholdLevelsMetric.NO_OP_METRIC;
}
logger.info("Initializing Eureka in region {}", clientConfig.getRegion());
if (!config.shouldRegisterWithEureka() && !config.shouldFetchRegistry()) {
logger.info("Client configured to neither register nor query for data.");
scheduler = null;
heartbeatExecutor = null;
cacheRefreshExecutor = null;
eurekaTransport = null;
instanceRegionChecker = new InstanceRegionChecker(new PropertyBasedAzToRegionMapper(config), clientConfig.getRegion());
// This is a bit of hack to allow for existing code using DiscoveryManager.getInstance()
// to work with DI'd DiscoveryClient
DiscoveryManager.getInstance().setDiscoveryClient(this);
DiscoveryManager.getInstance().setEurekaClientConfig(config);
initTimestampMs = System.currentTimeMillis();
initRegistrySize = this.getApplications().size();
registrySize = initRegistrySize;
logger.info("Discovery Client initialized at timestamp {} with initial instances count: {}",
initTimestampMs, initRegistrySize);
return; // no need to setup up an network tasks and we are done
}
try {
// default size of 2 - 1 each for heartbeat and cacheRefresh
scheduler = Executors.newScheduledThreadPool(2,
new ThreadFactoryBuilder()
.setNameFormat("DiscoveryClient-%d")
.setDaemon(true)
.build());
heartbeatExecutor = new ThreadPoolExecutor(
1, clientConfig.getHeartbeatExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(),
new ThreadFactoryBuilder()
.setNameFormat("DiscoveryClient-HeartbeatExecutor-%d")
.setDaemon(true)
.build()
); // use direct handoff
cacheRefreshExecutor = new ThreadPoolExecutor(
1, clientConfig.getCacheRefreshExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(),
new ThreadFactoryBuilder()
.setNameFormat("DiscoveryClient-CacheRefreshExecutor-%d")
.setDaemon(true)
.build()
); // use direct handoff
eurekaTransport = new EurekaTransport();
scheduleServerEndpointTask(eurekaTransport, args);
AzToRegionMapper azToRegionMapper;
if (clientConfig.shouldUseDnsForFetchingServiceUrls()) {
azToRegionMapper = new DNSBasedAzToRegionMapper(clientConfig);
} else {
azToRegionMapper = new PropertyBasedAzToRegionMapper(clientConfig);
}
if (null != remoteRegionsToFetch.get()) {
azToRegionMapper.setRegionsToFetch(remoteRegionsToFetch.get().split(","));
}
instanceRegionChecker = new InstanceRegionChecker(azToRegionMapper, clientConfig.getRegion());
} catch (Throwable e) {
throw new RuntimeException("Failed to initialize DiscoveryClient!", e);
}
if (clientConfig.shouldFetchRegistry() && !fetchRegistry(false)) {
fetchRegistryFromBackup();
}
// call and execute the pre registration handler before all background tasks (inc registration) is started
if (this.preRegistrationHandler != null) {
this.preRegistrationHandler.beforeRegistration();
}
if (clientConfig.shouldRegisterWithEureka() && clientConfig.shouldEnforceRegistrationAtInit()) {
try {
if (!register() ) {
throw new IllegalStateException("Registration error at startup. Invalid server response.");
}
} catch (Throwable th) {
logger.error("Registration error at startup: {}", th.getMessage());
throw new IllegalStateException(th);
}
}
// finally, init the schedule tasks (e.g. cluster resolvers, heartbeat, instanceInfo replicator, fetch
initScheduledTasks();
try {
Monitors.registerObject(this);
} catch (Throwable e) {
logger.warn("Cannot register timers", e);
}
// This is a bit of hack to allow for existing code using DiscoveryManager.getInstance()
// to work with DI'd DiscoveryClient
DiscoveryManager.getInstance().setDiscoveryClient(this);
DiscoveryManager.getInstance().setEurekaClientConfig(config);
initTimestampMs = System.currentTimeMillis();
initRegistrySize = this.getApplications().size();
registrySize = initRegistrySize;
logger.info("Discovery Client initialized at timestamp {} with initial instances count: {}",
initTimestampMs, initRegistrySize);
}
里面有这样一段
eurekaTransport.bootstrapResolver = EurekaHttpClients.newBootstrapResolver(
clientConfig,
transportConfig,
eurekaTransport.transportClientFactory,
applicationInfoManager.getInfo(),
applicationsSource,
endpointRandomizer
);
再进入newBootstrapResolver方法里。
public static ClosableResolver<AwsEndpoint> newBootstrapResolver(
final EurekaClientConfig clientConfig,
final EurekaTransportConfig transportConfig,
final TransportClientFactory transportClientFactory,
final InstanceInfo myInstanceInfo,
final ApplicationsResolver.ApplicationsSource applicationsSource,
final EndpointRandomizer randomizer) {
if (COMPOSITE_BOOTSTRAP_STRATEGY.equals(transportConfig.getBootstrapResolverStrategy())) {
if (clientConfig.shouldFetchRegistry()) {
return compositeBootstrapResolver(
clientConfig,
transportConfig,
transportClientFactory,
myInstanceInfo,
applicationsSource,
randomizer
);
} else {
logger.warn("Cannot create a composite bootstrap resolver if registry fetch is disabled." +
" Falling back to using a default bootstrap resolver.");
}
}
// if all else fails, return the default
return defaultBootstrapResolver(clientConfig, myInstanceInfo, randomizer);
}
最后一句的引用了defaultBootstrapResolver方法，继续进入defaultBootstrapResolver里看一看。
/**
* @return a bootstrap resolver that resolves eureka server endpoints based on either DNS or static config,
* depending on configuration for one or the other. This resolver will warm up at the start.
*/
static ClosableResolver<AwsEndpoint> defaultBootstrapResolver(final EurekaClientConfig clientConfig,
final InstanceInfo myInstanceInfo,
final EndpointRandomizer randomizer) {
String[] availZones = clientConfig.getAvailabilityZones(clientConfig.getRegion());
String myZone = InstanceInfo.getZone(availZones, myInstanceInfo);
ClusterResolver<AwsEndpoint> delegateResolver = new ZoneAffinityClusterResolver(
new ConfigClusterResolver(clientConfig, myInstanceInfo),
myZone,
true,
randomizer
);
List<AwsEndpoint> initialValue = delegateResolver.getClusterEndpoints();
if (initialValue.isEmpty()) {
String msg = "Initial resolution of Eureka server endpoints failed. Check ConfigClusterResolver logs for more info";
logger.error(msg);
failFastOnInitCheck(clientConfig, msg);
}
return new AsyncResolver<>(
EurekaClientNames.BOOTSTRAP,
delegateResolver,
initialValue,
1,
clientConfig.getEurekaServiceUrlPollIntervalSeconds() * 1000
);
}
这个方法是用来解析eureka的端点，两种途径DNS和配置文件。
里面有List<AwsEndpoint> initialValue = delegateResolver.getClusterEndpoints();这么一句。getClusterEndpoints这个方法有多个实现，我们选择ConfigClusterResolver类中的实现方法。
public List<AwsEndpoint> getClusterEndpoints() {
if (clientConfig.shouldUseDnsForFetchingServiceUrls()) {
if (logger.isInfoEnabled()) {
logger.info("Resolving eureka endpoints via DNS: {}", getDNSName());
}
return getClusterEndpointsFromDns();
} else {
logger.info("Resolving eureka endpoints via configuration");
return getClusterEndpointsFromConfig();
}
}
可以看到这个方法中飞为了从dns中获取url或者从config中获取。直接进入getClusterEndpointsFromConfig方法。
private List<AwsEndpoint> getClusterEndpointsFromConfig() {
String[] availZones = clientConfig.getAvailabilityZones(clientConfig.getRegion());
String myZone = InstanceInfo.getZone(availZones, myInstanceInfo);
Map<String, List<String>> serviceUrls = EndpointUtils
.getServiceUrlsMapFromConfig(clientConfig, myZone, clientConfig.shouldPreferSameZoneEureka());
List<AwsEndpoint> endpoints = new ArrayList<>();
for (String zone : serviceUrls.keySet()) {
for (String url : serviceUrls.get(zone)) {
try {
endpoints.add(new AwsEndpoint(url, getRegion(), zone));
} catch (Exception ignore) {
logger.warn("Invalid eureka server URI: {}; removing from the server pool", url);
}
}
}
logger.debug("Config resolved to {}", endpoints);
if (endpoints.isEmpty()) {
logger.error("Cannot resolve to any endpoints from provided configuration: {}", serviceUrls);
}
return endpoints;
}
不容易，终于看到getServiceUrlsMapFromConfig的调用了。所以为了理解client是如何获取eureka server的url的，我们可以在启动client的时候在getServiceUrlsMapFromConfig里面打上端点调试。
网友评论