Spring Cloud是一個生態，它提供了一套標準，這套標準可以通過不同的組件來實作，其中就包含服務注冊/發現、熔斷、負載均衡等，在spring-cloud-common這個包中，org.springframework.cloud.client.serviceregistry路徑下，可以看到一個服務注冊的介面定義ServiceRegistry，它就是定義了spring cloud中服務注冊的一個介面，
public interface ServiceRegistry<R extends Registration> {
void register(R registration);
void deregister(R registration);
void close();
void setStatus(R registration, String status);
<T> T getStatus(R registration);
}
我們看一下它的類關系圖，這個介面有一個唯一的實作EurekaServiceRegistry，表示采用的是Eureka Server作為服務注冊中心，
自動注冊的觸發機制
Eureka自動注冊，是通過EurekaAutoServiceRegistration這個物件來觸發的，
在Spring Boot專案啟動時，會基于自動裝配機制，在EurekaClientAutoConfiguration這個配置類中，初始化一個EurekaAutoServiceRegistration這個Bean物件，代碼如下，
public class EurekaClientAutoConfiguration {
@Bean
@ConditionalOnBean(AutoServiceRegistrationProperties.class)
@ConditionalOnProperty(
value = "https://www.cnblogs.com/mic112/p/spring.cloud.service-registry.auto-registration.enabled",
matchIfMissing = true)
public EurekaAutoServiceRegistration eurekaAutoServiceRegistration(
ApplicationContext context, EurekaServiceRegistry registry,
EurekaRegistration registration) {
return new EurekaAutoServiceRegistration(context, registry, registration);
}
}
EurekaAutoServiceRegistration這個類的定義如下，
public class EurekaAutoServiceRegistration implements AutoServiceRegistration,
SmartLifecycle, Ordered, SmartApplicationListener {
//省略
@Override
public void start() {
// only set the port if the nonSecurePort or securePort is 0 and this.port != 0
if (this.port.get() != 0) {
if (this.registration.getNonSecurePort() == 0) {
this.registration.setNonSecurePort(this.port.get());
}
if (this.registration.getSecurePort() == 0 && this.registration.isSecure()) {
this.registration.setSecurePort(this.port.get());
}
}
// only initialize if nonSecurePort is greater than 0 and it isn't already running
// because of containerPortInitializer below
if (!this.running.get() && this.registration.getNonSecurePort() > 0) {
this.serviceRegistry.register(this.registration);
this.context.publishEvent(new InstanceRegisteredEvent<>(this,
this.registration.getInstanceConfig()));
this.running.set(true);
}
}
//省略...
}
我們發現,EurekaAutoServiceRegistration實作了SmartLifecycle介面，當Spring容器加載完所有的Bean并且初始化之后，會繼續回呼實作了SmartLifeCycle介面的類中對應的方法，比如（start），
SmartLifeCycle知識拓展
我拓展一下SmartLifeCycle這塊的知識， SmartLifeCycle是一個介面，當Spring容器加載完所有的Bean并且初始化之后，會繼續回呼實作了SmartLifeCycle介面的類中對應的方法，比如（start），
實際上我們自己也可以拓展，比如在springboot工程的main方法同級目錄下，寫一個測驗類，實作SmartLifeCycle介面，并且通過@Service宣告為一個bean，因為要被spring去加載，首先得是bean，
@Service
public class TestSmartLifeCycle implements SmartLifecycle {
@Override
public void start() {
System.out.println("start");
}
@Override
public void stop() {
System.out.println("stop");
}
@Override
public boolean isRunning() {
return false;
}
}
接著，我們啟動spring boot應用后，可以看到控制臺輸出了start字串，
我們在DefaultLifecycleProcessor.startBeans方法上加一個debug，可以很明顯的看到我們自己定義的TestSmartLifeCycle被掃描到了，并且最后會呼叫該bean的start方法，
在startBeans方法中，我們可以看到它首先會獲得所有實作了SmartLifeCycle的Bean，然后會回圈呼叫實作了SmartLifeCycle的bean的start方法，代碼如下，
private void startBeans(boolean autoStartupOnly) {
Map<String, Lifecycle> lifecycleBeans = this.getLifecycleBeans();
Map<Integer, DefaultLifecycleProcessor.LifecycleGroup> phases = new HashMap();
lifecycleBeans.forEach((beanName, bean) -> {
if (!autoStartupOnly || bean instanceof SmartLifecycle && ((SmartLifecycle)bean).isAutoStartup()) {
int phase = this.getPhase(bean);
DefaultLifecycleProcessor.LifecycleGroup group = (DefaultLifecycleProcessor.LifecycleGroup)phases.get(phase);
if (group == null) {
group = new DefaultLifecycleProcessor.LifecycleGroup(phase, this.timeoutPerShutdownPhase, lifecycleBeans, autoStartupOnly);
phases.put(phase, group);
}
group.add(beanName, bean);
}
});
if (!phases.isEmpty()) {
List<Integer> keys = new ArrayList(phases.keySet());
Collections.sort(keys);
Iterator var5 = keys.iterator();
while(var5.hasNext()) {
Integer key = (Integer)var5.next();
((DefaultLifecycleProcessor.LifecycleGroup)phases.get(key)).start(); //回圈呼叫實作了SmartLifeCycle介面的start方法，
}
}
}
SmartLifeCycle介面的回呼，是在SpringBoot啟動時觸發，具體的執行路徑如下！
SpringApplication.run() -> this.refreshContext(context);->this.refresh(context);->ServletWebServerApplicationContext.refresh()->this.finishRefresh();->AbstractApplicationContext.finishRefresh->DefaultLifecycleProcessor.onRefresh()-> this.startBeans->this.start()->this.doStart()->
服務注冊
因此，當SpringBoot啟動時，會觸發在EurekaAutoServiceRegistration中的start方法，代碼如下，
public class EurekaAutoServiceRegistration implements AutoServiceRegistration,
SmartLifecycle, Ordered, SmartApplicationListener {
//省略
@Override
public void start() {
// only set the port if the nonSecurePort or securePort is 0 and this.port != 0
if (this.port.get() != 0) {
if (this.registration.getNonSecurePort() == 0) {
this.registration.setNonSecurePort(this.port.get());
}
if (this.registration.getSecurePort() == 0 && this.registration.isSecure()) {
this.registration.setSecurePort(this.port.get());
}
}
// only initialize if nonSecurePort is greater than 0 and it isn't already running
// because of containerPortInitializer below
if (!this.running.get() && this.registration.getNonSecurePort() > 0) {
//實作服務注冊，
this.serviceRegistry.register(this.registration);
//發布一個事件
this.context.publishEvent(new InstanceRegisteredEvent<>(this,
this.registration.getInstanceConfig()));
this.running.set(true);
}
}
//省略...
}
EurekaServiceRegistry.register
this.serviceRegistry.register(this.registration);，實際呼叫的是EurekaServiceRegistry這個物件中的register方法，代碼如下，
public class EurekaServiceRegistry implements ServiceRegistry<EurekaRegistration> {
private static final Log log = LogFactory.getLog(EurekaServiceRegistry.class);
@Override
public void register(EurekaRegistration reg) {
maybeInitializeClient(reg);
if (log.isInfoEnabled()) {
log.info("Registering application "
+ reg.getApplicationInfoManager().getInfo().getAppName()
+ " with eureka with status "
+ reg.getInstanceConfig().getInitialStatus());
}
//設定當前實體的狀態，一旦這個實體的狀態發生變化，只要狀態不是DOWN，那么就會被監聽器監聽并且執行服務注冊，
reg.getApplicationInfoManager()
.setInstanceStatus(reg.getInstanceConfig().getInitialStatus());
//設定健康檢查的處理
reg.getHealthCheckHandler().ifAvailable(healthCheckHandler -> reg
.getEurekaClient().registerHealthCheck(healthCheckHandler));
}
}
從上述代碼來看，注冊方法中并沒有真正呼叫Eureka的方法去執行注冊，而是僅僅設定了一個狀態以及設定健康檢查處理器，我們繼續看一下reg.getApplicationInfoManager().setInstanceStatus方法，
public synchronized void setInstanceStatus(InstanceStatus status) {
InstanceStatus next = instanceStatusMapper.map(status);
if (next == null) {
return;
}
InstanceStatus prev = instanceInfo.setStatus(next);
if (prev != null) {
for (StatusChangeListener listener : listeners.values()) {
try {
listener.notify(new StatusChangeEvent(prev, next));
} catch (Exception e) {
logger.warn("failed to notify listener: {}", listener.getId(), e);
}
}
}
}
在這個方法中，它會通過監聽器來發布一個狀態變更事件，ok，此時listener的實體是StatusChangeListener，也就是呼叫StatusChangeListener的notify()方法，這個事件是觸發一個服務狀態變更，應該是有地方會監聽這個事件，然后基于這個事件，
這個時候我們以為找到了方向，然后點擊進去一看，卞擊，發現它是一個介面，而且我們發現它是靜態的內部介面，還無法直接看到它的實作類，
依我多年原始碼閱讀經驗，于是又往回找，因為我基本上能猜測到一定是在某個地方做了初始化的作業，于是，我想找到EurekaServiceRegistry.register方法中的reg.getApplicationInfoManager這個實體是什么，而且我們發現ApplicationInfoManager是來自于EurekaRegistration這個類中的屬性，
public class EurekaRegistration implements Registration {
private final ApplicationInfoManager applicationInfoManager;
private ObjectProvider<HealthCheckHandler> healthCheckHandler;
private EurekaRegistration(CloudEurekaInstanceConfig instanceConfig,
EurekaClient eurekaClient, ApplicationInfoManager applicationInfoManager,
ObjectProvider<HealthCheckHandler> healthCheckHandler) {
this.eurekaClient = eurekaClient;
this.instanceConfig = instanceConfig;
this.applicationInfoManager = applicationInfoManager;
this.healthCheckHandler = healthCheckHandler;
}
}
而EurekaRegistration又是在EurekaAutoServiceRegistration這個類中實體化的，
那我們去EurekaAutoServiceRegistration這個配置類中，找一下applicationInfoManager的實體化程序，代碼如下：
@Configuration(proxyBeanMethods = false)
@ConditionalOnMissingRefreshScope
protected static class EurekaClientConfiguration {
@Bean
@ConditionalOnMissingBean(value = https://www.cnblogs.com/mic112/p/ApplicationInfoManager.class,
search = SearchStrategy.CURRENT)
@org.springframework.cloud.context.config.annotation.RefreshScope
@Lazy
public ApplicationInfoManager eurekaApplicationInfoManager(
EurekaInstanceConfig config) {
InstanceInfo instanceInfo = new InstanceInfoFactory().create(config);
return new ApplicationInfoManager(config, instanceInfo); //構建了一個ApplicationInfoManager實體，
}
}
在ApplicationInfoManager的構造方法中，初始化了一個listeners物件，它是一個ConcurrentHashMap集合，但是初始化的時候，這個集合并沒有賦值，
@Inject
public ApplicationInfoManager(EurekaInstanceConfig config, InstanceInfo instanceInfo, OptionalArgs optionalArgs) {
this.config = config;
this.instanceInfo = instanceInfo;
this.listeners = new ConcurrentHashMap<String, StatusChangeListener>();
if (optionalArgs != null) {
this.instanceStatusMapper = optionalArgs.getInstanceStatusMapper();
} else {
this.instanceStatusMapper = NO_OP_MAPPER;
}
// Hack to allow for getInstance() to use the DI'd ApplicationInfoManager
instance = this;
}
遇到這個問題，我們先別慌，先來看一下ApplicationInfoManager這個類中對listeners賦值的方法如下，
public void registerStatusChangeListener(StatusChangeListener listener) {
listeners.put(listener.getId(), listener);
}
這個方法唯一的呼叫方是：DiscoveryClient.initScheduledTasks方法，
這個方法又是在哪里呼叫的呢？
DiscoveryClient
在EurekaClientAutoConfiguration這個自動配置類的靜態內部類EurekaClientConfiguration中，通過@Bean注入了一個CloudEurekaClient實體，代碼如下，
@Configuration(proxyBeanMethods = false)
@ConditionalOnMissingRefreshScope
protected static class EurekaClientConfiguration {
@Autowired
private ApplicationContext context;
@Autowired
private AbstractDiscoveryClientOptionalArgs<?> optionalArgs;
@Bean(destroyMethod = "shutdown")
@ConditionalOnMissingBean(value = https://www.cnblogs.com/mic112/p/EurekaClient.class,
search = SearchStrategy.CURRENT)
public EurekaClient eurekaClient(ApplicationInfoManager manager,
EurekaClientConfig config) {
return new CloudEurekaClient(manager, config, this.optionalArgs,
this.context);
}
}
從名字不難猜測出，EurekaClient應該是專門負責和EurekaServer進行互動的客戶端實作類，而這里回傳的實體物件是CloudEurekaClient，構造代碼如下，
public CloudEurekaClient(ApplicationInfoManager applicationInfoManager,
EurekaClientConfig config, AbstractDiscoveryClientOptionalArgs<?> args,
ApplicationEventPublisher publisher) {
super(applicationInfoManager, config, args);
this.applicationInfoManager = applicationInfoManager;
this.publisher = publisher;
this.eurekaTransportField = ReflectionUtils.findField(DiscoveryClient.class,
"eurekaTransport");
ReflectionUtils.makeAccessible(this.eurekaTransportField);
}
注意，在CloudEurekaClient這個類的構造方法中，傳遞了ApplicationInfoManager這個實體，后續會用到，
同時，該構造方法中會同步呼叫 super(applicationInfoManager, config, args); ，也就是呼叫父類DiscoveryClient的構造方法，代碼如下，
public DiscoveryClient(ApplicationInfoManager applicationInfoManager, final EurekaClientConfig config, AbstractDiscoveryClientOptionalArgs args) {
this(applicationInfoManager, config, args, ResolverUtils::randomize);
}
最侄訓呼叫DiscoveryClient中多載的如下方法，代碼比較長，把非關鍵代碼省略，
DiscoveryClient(ApplicationInfoManager applicationInfoManager, EurekaClientConfig config, AbstractDiscoveryClientOptionalArgs args,
Provider<BackupRegistry> backupRegistryProvider, EndpointRandomizer endpointRandomizer) {
//省略....
if (config.shouldFetchRegistry()) { //是否要從eureka server上獲取服務地址資訊
this.registryStalenessMonitor = new ThresholdLevelsMetric(this, METRIC_REGISTRY_PREFIX + "lastUpdateSec_", new long[]{15L, 30L, 60L, 120L, 240L, 480L});
} else {
this.registryStalenessMonitor = ThresholdLevelsMetric.NO_OP_METRIC;
}
//是否要注冊到eureka server上
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);
}
//如果需要注冊到Eureka server并且是開啟了初始化的時候強制注冊，則呼叫register()發起服務注冊
if (clientConfig.shouldFetchRegistry()) {
try {
//從Eureka-Server中拉去注冊地址資訊
boolean primaryFetchRegistryResult = fetchRegistry(false);
if (!primaryFetchRegistryResult) {
logger.info("Initial registry fetch from primary servers failed");
}
//從備用地址拉去服務注冊資訊
boolean backupFetchRegistryResult = true;
if (!primaryFetchRegistryResult && !fetchRegistryFromBackup()) {
backupFetchRegistryResult = false;
logger.info("Initial registry fetch from backup servers failed");
}
//如果還是沒有拉取到，并且配置了強制拉取注冊表的話，就會拋例外
if (!primaryFetchRegistryResult && !backupFetchRegistryResult && clientConfig.shouldEnforceFetchRegistryAtInit()) {
throw new IllegalStateException("Fetch registry error at startup. Initial fetch failed.");
}
} catch (Throwable th) {
logger.error("Fetch registry error at startup: {}", th.getMessage());
throw new IllegalStateException(th);
}
}
// call and execute the pre registration handler before all background tasks (inc registration) is started
//這里是判斷一下有沒有預注冊處理器，有的話就執行一下
if (this.preRegistrationHandler != null) {
this.preRegistrationHandler.beforeRegistration();
}
//如果需要注冊到Eureka server并且是開啟了初始化的時候強制注冊，則呼叫register()發起服務注冊(默認情況下，shouldEnforceRegistrationAtInit為false)
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);
}
DiscoveryClient.initScheduledTasks
initScheduledTasks去啟動一個定時任務，
- 如果配置了開啟從注冊中心重繪服務串列，則會開啟cacheRefreshExecutor這個定時任務
- 如果開啟了服務注冊到Eureka，則通過需要做幾個事情.
- 建立心跳檢測機制
- 通過內部類來實體化StatusChangeListener 實體狀態監控介面，這個就是前面我們在分析啟動程序中所看到的，呼叫notify的方法，實際上會在這里體現，
private void initScheduledTasks() {
//如果配置了開啟從注冊中心重繪服務串列，則會開啟cacheRefreshExecutor這個定時任務
if (clientConfig.shouldFetchRegistry()) {
// registry cache refresh timer
//registryFetchIntervalSeconds:30s
int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();
//expBackOffBound:10
int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();
cacheRefreshTask = new TimedSupervisorTask(
"cacheRefresh",
scheduler,
cacheRefreshExecutor,
registryFetchIntervalSeconds,
TimeUnit.SECONDS,
expBackOffBound,
new CacheRefreshThread()
);
scheduler.schedule(
cacheRefreshTask,
registryFetchIntervalSeconds, TimeUnit.SECONDS);
}
//如果開啟了服務注冊到Eureka，則通過需要做幾個事情
if (clientConfig.shouldRegisterWithEureka()) {
int renewalIntervalInSecs = instanceInfo.getLeaseInfo().getRenewalIntervalInSecs();
int expBackOffBound = clientConfig.getHeartbeatExecutorExponentialBackOffBound();
logger.info("Starting heartbeat executor: " + "renew interval is: {}", renewalIntervalInSecs);
// 開啟一個心跳任務
heartbeatTask = new TimedSupervisorTask(
"heartbeat",
scheduler,
heartbeatExecutor,
renewalIntervalInSecs,
TimeUnit.SECONDS,
expBackOffBound,
new HeartbeatThread()
);
scheduler.schedule(
heartbeatTask,
renewalIntervalInSecs, TimeUnit.SECONDS);
//創建一個instanceInfoReplicator實體資訊復制器
instanceInfoReplicator = new InstanceInfoReplicator(
this,
instanceInfo,
clientConfig.getInstanceInfoReplicationIntervalSeconds(),
2); // burstSize
//初始化一個狀態變更監聽器
statusChangeListener = new ApplicationInfoManager.StatusChangeListener() {
@Override
public String getId() {
return "statusChangeListener";
}
@Override
public void notify(StatusChangeEvent statusChangeEvent) {
logger.info("Saw local status change event {}", statusChangeEvent);
instanceInfoReplicator.onDemandUpdate();
}
};
//注冊實體狀態變化的監聽
if (clientConfig.shouldOnDemandUpdateStatusChange()) {
applicationInfoManager.registerStatusChangeListener(statusChangeListener); //注意（case）
}
//啟動一個實體資訊復制器，主要就是為了開啟一個定時執行緒，每40秒判斷實體資訊是否變更，如果變更了則重新注冊
instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());
} else {
logger.info("Not registering with Eureka server per configuration");
}
}
在上述代碼中，我們發現了一個很重要的邏輯:applicationInfoManager.registerStatusChangeListener(statusChangeListener);
這個代碼是注冊一個StatusChangeListener，保存到ApplicationInfoManager中的listener集合中， （還記得前面原始碼分析中的服務注冊邏輯嗎？當服務器啟動或者停止時，會呼叫ApplicationInfoManager.listener，逐個遍歷呼叫listener.notify方法），而這個listener集合中的物件是在DiscoveryClient初始化的時候完成的，
instanceInfoReplicator.onDemandUpdate()
這個方法的主要作用是根據實體資料是否發生變化，來觸發服務注冊中心的資料，
public boolean onDemandUpdate() {
if (rateLimiter.acquire(burstSize, allowedRatePerMinute)) { //限流判斷
if (!scheduler.isShutdown()) { //提交一個任務
scheduler.submit(new Runnable() {
@Override
public void run() {
logger.debug("Executing on-demand update of local InstanceInfo");
//取出之前已經提交的任務，也就是在start方法中提交的更新任務，如果任務還沒有執行完成，則取消之前的任務，
Future latestPeriodic = scheduledPeriodicRef.get();
if (latestPeriodic != null && !latestPeriodic.isDone()) {
logger.debug("Canceling the latest scheduled update, it will be rescheduled at the end of on demand update");
latestPeriodic.cancel(false);//如果此任務未完成，就立即取消
}
//通過呼叫run方法，令任務在延時后執行，相當于周期性任務中的一次
InstanceInfoReplicator.this.run();
}
});
return true;
} else {
logger.warn("Ignoring onDemand update due to stopped scheduler");
return false;
}
} else {
logger.warn("Ignoring onDemand update due to rate limiter");
return false;
}
}
InstanceInfoReplicator.this.run();
run方法呼叫register方法進行服務注冊，并且在finally中，每30s會定時執行一下當前的run 方法進行檢查，
public void run() {
try {
//重繪實體資訊
discoveryClient.refreshInstanceInfo();
//是否有狀態更新過了，有的話獲取更新的時間
Long dirtyTimestamp = instanceInfo.isDirtyWithTime();
if (dirtyTimestamp != null) {//有臟資料，要重新注冊
discoveryClient.register();
instanceInfo.unsetIsDirty(dirtyTimestamp);
}
} catch (Throwable t) {
logger.warn("There was a problem with the instance info replicator", t);
} finally {
//每隔30s，執行一次當前的`run`方法
Future next = scheduler.schedule(this, replicationIntervalSeconds, TimeUnit.SECONDS);
scheduledPeriodicRef.set(next);
}
}
DiscoveryClient.register
記過上述分析后，最終我們找到了Eureka的服務注冊方法:eurekaTransport.registrationClient.register，最終呼叫的是 AbstractJerseyEurekaHttpClient#register(...)，
boolean register() throws Throwable {
logger.info(PREFIX + "{}: registering service...", appPathIdentifier);
EurekaHttpResponse<Void> httpResponse;
try {
httpResponse = eurekaTransport.registrationClient.register(instanceInfo);
} catch (Exception e) {
logger.warn(PREFIX + "{} - registration failed {}", appPathIdentifier, e.getMessage(), e);
throw e;
}
if (logger.isInfoEnabled()) {
logger.info(PREFIX + "{} - registration status: {}", appPathIdentifier, httpResponse.getStatusCode());
}
return httpResponse.getStatusCode() == Status.NO_CONTENT.getStatusCode();
}
AbstractJerseyEurekaHttpClient#register
很顯然，這里是發起了一次http請求，訪問Eureka-Server的apps/${APP_NAME}介面，將當前服務實體的資訊發送到Eureka Server進行保存，
至此，我們基本上已經知道Spring Cloud Eureka 是如何在啟動的時候把服務資訊注冊到Eureka Server上的了
public EurekaHttpResponse<Void> register(InstanceInfo info) {
String urlPath = "apps/" + info.getAppName();
ClientResponse response = null;
try {
Builder resourceBuilder = jerseyClient.resource(serviceUrl).path(urlPath).getRequestBuilder();
addExtraHeaders(resourceBuilder);
response = resourceBuilder
.header("Accept-Encoding", "gzip")
.type(MediaType.APPLICATION_JSON_TYPE)
.accept(MediaType.APPLICATION_JSON)
.post(ClientResponse.class, info);
return anEurekaHttpResponse(response.getStatus()).headers(headersOf(response)).build();
} finally {
if (logger.isDebugEnabled()) {
logger.debug("Jersey HTTP POST {}/{} with instance {}; statusCode={}", serviceUrl, urlPath, info.getId(),
response == null ? "N/A" : response.getStatus());
}
if (response != null) {
response.close();
}
}
}
服務注冊總結
服務注冊的程序分兩個步驟
- DiscoveryClient這個物件，在初始化時，呼叫
initScheduledTask()方法，構建一個StatusChangeListener監聽， - Spring Cloud應用在啟動時，基于SmartLifeCycle介面回呼，觸發StatusChangeListener事件通知
- 在StatusChangeListener的回呼方法中，通過呼叫
onDemandUpdate方法，去更新客戶端的地址資訊，從而完成服務注冊，
Eureka注冊資訊如何存盤？
Eureka Server收到客戶端的服務注冊請求后，需要把資訊存盤到Eureka Server中，它的存盤結構如下圖所示，
EurekaServer采用了ConcurrentHashMap集合的方式，來存盤服務提供者的地址資訊，其中，每個節點的實體資訊的最終存盤物件是InstanceInfo，>
Eureka Server接收請求處理
請求入口在： com.netflix.eureka.resources.ApplicationResource.addInstance()，
大家可以發現，這里所提供的REST服務，采用的是jersey來實作的，Jersey是基于JAX-RS標準，提供REST的實作的支持，這里就不展開分析了，
Eureka Server端定義的服務注冊介面實作如下：
@POST
@Consumes({"application/json", "application/xml"})
public Response addInstance(InstanceInfo info,
@HeaderParam(PeerEurekaNode.HEADER_REPLICATION) String isReplication) {
logger.debug("Registering instance {} (replication={})", info.getId(), isReplication);
// handle cases where clients may be registering with bad DataCenterInfo with missing data
//實體部署的資料中心， 這里是AWS實作的資料相關的邏輯，這里不涉及到，所以不需要去關心
DataCenterInfo dataCenterInfo = info.getDataCenterInfo();
if (dataCenterInfo instanceof UniqueIdentifier) {
String dataCenterInfoId = ((UniqueIdentifier) dataCenterInfo).getId();
if (isBlank(dataCenterInfoId)) {
boolean experimental = "true".equalsIgnoreCase(serverConfig.getExperimental("registration.validation.dataCenterInfoId"));
if (experimental) {
String entity = "DataCenterInfo of type " + dataCenterInfo.getClass() + " must contain a valid id";
return Response.status(400).entity(entity).build();
} else if (dataCenterInfo instanceof AmazonInfo) {
AmazonInfo amazonInfo = (AmazonInfo) dataCenterInfo;
String effectiveId = amazonInfo.get(AmazonInfo.MetaDataKey.instanceId);
if (effectiveId == null) {
amazonInfo.getMetadata().put(AmazonInfo.MetaDataKey.instanceId.getName(), info.getId());
}
} else {
logger.warn("Registering DataCenterInfo of type {} without an appropriate id", dataCenterInfo.getClass());
}
}
}
//在這里會呼叫服務注冊方法，傳遞`info`，表示客戶端的服務實體資訊，
registry.register(info, "true".equals(isReplication));
return Response.status(204).build(); // 204 to be backwards compatible
}
PeerAwareInstanceRegistryImpl.register
我們先來看PeerAwareInstanceRegistryImpl的類關系圖，從類關系圖可以看出，PeerAwareInstanceRegistry的最頂層介面為LeaseManager與LookupService,
- 其中LookupService定義了最基本的發現實體的行為，
- LeaseManager定義了處理客戶端注冊，續約，注銷等操作，
InstanceRegistry.register
接著進入到InstanceRegistry的register方法，在這個方法中，增加了一個handleRegistration方法的呼叫，這個方法用來發布一個EurekaInstanceRegisteredEvent事件，
@Override
public void register(final InstanceInfo info, final boolean isReplication) {
handleRegistration(info, resolveInstanceLeaseDuration(info), isReplication);
super.register(info, isReplication);
}
父類的register方法
接著呼叫父類PeerAwareInstanceRegistryImpl的register方法，代碼如下，
@Override
public void register(final InstanceInfo info, final boolean isReplication) {
int leaseDuration = Lease.DEFAULT_DURATION_IN_SECS; //租約過期時間
if (info.getLeaseInfo() != null && info.getLeaseInfo().getDurationInSecs() > 0) { //如果客戶端有自己定義心跳超時時間，則采用客戶端的
leaseDuration = info.getLeaseInfo().getDurationInSecs();
}
super.register(info, leaseDuration, isReplication); //節點注冊
replicateToPeers(Action.Register, info.getAppName(), info.getId(), info, null, isReplication); //把注冊資訊同步到其他集群節點，
}
其中：
- leaseDuration 表示租約過期時間，默認是90s，也就是當服務端超過90s沒有收到客戶端的心跳，則主動剔除該節點
- 呼叫super.register發起節點注冊
- 將資訊復制到Eureka Server集群中的其他機器上，同步的實作也很簡單，就是獲得集群中的所有節點，然后逐個發起注冊
AbstractInstanceRegistry.register
最終在這個抽象類的實體注冊類中完成服務注冊的實作，代碼如下，
public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) {
read.lock();
try {
//從registry中獲得當前實體資訊，根據appName， registry中保存了所有客戶端的實體資料
Map<String, Lease<InstanceInfo>> gMap = registry.get(registrant.getAppName());
REGISTER.increment(isReplication); //原子遞增，做資料統計
if (gMap == null) { //如果gMap為空，說明當前服務端沒有保存該實體資料，則通過下面代碼進行初始化
final ConcurrentHashMap<String, Lease<InstanceInfo>> gNewMap = new ConcurrentHashMap<String, Lease<InstanceInfo>>();
gMap = registry.putIfAbsent(registrant.getAppName(), gNewMap);
if (gMap == null) {
gMap = gNewMap;
}
}
//從gMap中查詢已經存在的Lease資訊，Lease中文翻譯為租約，實際上它把服務提供者的實體資訊包裝成了一個lease，里面提供了對于改服務實體的租約管理
Lease<InstanceInfo> existingLease = gMap.get(registrant.getId());
// 當instance已經存在時，和客戶端的instance的資訊做比較，時間最新的那個，為有效instance資訊
if (existingLease != null && (existingLease.getHolder() != null)) {
Long existingLastDirtyTimestamp = existingLease.getHolder().getLastDirtyTimestamp();
Long registrationLastDirtyTimestamp = registrant.getLastDirtyTimestamp();
logger.debug("Existing lease found (existing={}, provided={}", existingLastDirtyTimestamp, registrationLastDirtyTimestamp);
// this is a > instead of a >= because if the timestamps are equal, we still take the remote transmitted
// InstanceInfo instead of the server local copy.
// 比較lastDirtyTimestamp ， 以lastDirtyTimestamp大的為準
if (existingLastDirtyTimestamp > registrationLastDirtyTimestamp) {
logger.warn("There is an existing lease and the existing lease's dirty timestamp {} is greater" +
" than the one that is being registered {}", existingLastDirtyTimestamp, registrationLastDirtyTimestamp);
logger.warn("Using the existing instanceInfo instead of the new instanceInfo as the registrant");
registrant = existingLease.getHolder(); //重新賦值registrant為服務端最新的實體資訊
}
} else {
// 如果lease不存在，則認為是一個新的實體資訊，執行下面這段代碼（后續單獨分析它的作用）
synchronized (lock) {
if (this.expectedNumberOfClientsSendingRenews > 0) {
// Since the client wants to register it, increase the number of clients sending renews
this.expectedNumberOfClientsSendingRenews = this.expectedNumberOfClientsSendingRenews + 1;
updateRenewsPerMinThreshold();
}
}
logger.debug("No previous lease information found; it is new registration");
}
//創建一個Lease租約資訊
Lease<InstanceInfo> lease = new Lease<InstanceInfo>(registrant, leaseDuration);
if (existingLease != null) { // 當原來存在Lease的資訊時，設定serviceUpTimestamp, 保證服務啟動的時間一直是第一次注冊的那個（避免狀態變更影響到服務啟動時間）
lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp());
}
gMap.put(registrant.getId(), lease); //把當前服務實體保存到gMap中，
recentRegisteredQueue.add(new Pair<Long, String>(
System.currentTimeMillis(),
registrant.getAppName() + "(" + registrant.getId() + ")"));
// This is where the initial state transfer of overridden status happens
//如果實體狀態不等于UNKNOWN，則把當前實體狀態添加到overriddenInstanceStatusMap中
if (!InstanceStatus.UNKNOWN.equals(registrant.getOverriddenStatus())) {
logger.debug("Found overridden status {} for instance {}. Checking to see if needs to be add to the "
+ "overrides", registrant.getOverriddenStatus(), registrant.getId());
if (!overriddenInstanceStatusMap.containsKey(registrant.getId())) {
logger.info("Not found overridden id {} and hence adding it", registrant.getId());
overriddenInstanceStatusMap.put(registrant.getId(), registrant.getOverriddenStatus());
}
}
//重寫實體狀態
InstanceStatus overriddenStatusFromMap = overriddenInstanceStatusMap.get(registrant.getId());
if (overriddenStatusFromMap != null) {
logger.info("Storing overridden status {} from map", overriddenStatusFromMap);
registrant.setOverriddenStatus(overriddenStatusFromMap);
}
// Set the status based on the overridden status rules
InstanceStatus overriddenInstanceStatus = getOverriddenInstanceStatus(registrant, existingLease, isReplication);
registrant.setStatusWithoutDirty(overriddenInstanceStatus);// 設定實體資訊的狀態，但不標記 dirty
// If the lease is registered with UP status, set lease service up timestamp
if (InstanceStatus.UP.equals(registrant.getStatus())) { //如果服務實體資訊為UP狀態，則更新該實體的啟動時間，
lease.serviceUp();
}
registrant.setActionType(ActionType.ADDED); // 設定注冊型別為添加
recentlyChangedQueue.add(new RecentlyChangedItem(lease)); // 租約變更記錄佇列，記錄了實體的每次變化， 用于注冊資訊的增量獲取
registrant.setLastUpdatedTimestamp(); //修改最后一次更新時間
//讓快取失效
invalidateCache(registrant.getAppName(), registrant.getVIPAddress(), registrant.getSecureVipAddress());
logger.info("Registered instance {}/{} with status {} (replication={})",
registrant.getAppName(), registrant.getId(), registrant.getStatus(), isReplication);
} finally {
read.unlock();
}
}
EurekaServer注冊資訊存盤總結
至此，我們就把服務注冊在客戶端和服務端的處理程序做了一個詳細的分析，實際上在Eureka Server端，會把客戶端的地址資訊保存到ConcurrentHashMap中存盤，并且服務提供者和注冊中心之間，會建立一個心跳檢測機制，
用于監控服務提供者的健康狀態，
著作權宣告：本博客所有文章除特別宣告外，均采用 CC BY-NC-SA 4.0 許可協議，轉載請注明來自
Mic帶你學架構！
如果本篇文章對您有幫助，還請幫忙點個關注和贊，您的堅持是我不斷創作的動力，歡迎關注「跟著Mic學架構」公眾號公眾號獲取更多技術干貨！
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/380810.html
標籤：Java