元數(shù)據(jù)
除了普通的基礎設置之外，eureka支持自定義元數(shù)據(jù)。配置方式如下
eureka:
instance:
metadata-map:
cluster: cl1
name: zhaozhen
獲取元數(shù)據(jù)代碼
List<ServiceInstance> list = discoveryClient.getInstances("zhao-service-resume");
ServiceInstance serviceInstance = list.get(0);
list.stream().forEach(s->{
System.out.println(s.getMetadata());
});
在調(diào)用時通過斷點可以知道具體的元數(shù)據(jù)。在實際使用過程中，我們可以針對配置的不同元數(shù)據(jù)采取不同的執(zhí)行
file
可用性
從技術網(wǎng)站上搜到的一個面試題就有這樣的問題:eureka怎么保證可用性.
眾所周知，eureka采用的是AP模式，實現(xiàn)高可用最好的方式就是利用最少三臺eureke server實例，實現(xiàn)兩兩之間的服務注冊。從而達到同步數(shù)據(jù)的目的
那么這就涉及到如下的方面
- eureka client和eureka server之間如何進行通信
- eureka注冊在客戶端和服務端分別怎么操作實現(xiàn)可用性的
- eureka續(xù)約/心跳在客戶端和服務端分別怎么操作實現(xiàn)可用性的
- eureka下線是怎么操作的
eureka client和eureka server之間如何進行通信
通過查詢各種資料并追蹤自動配置類發(fā)現(xiàn)，eureka和eureka之間的通信是采用類似springmvc的Jersey框架暴露接口進行通信的。通信的形式基本類似于我們使用http進行請求的方式。在EurekaServerAutoConfiguration中通過注入FilterRegistrationBean實現(xiàn)了在filter中加入包含了指定包名下的所有的Jersey的外部接口
/**
* Register the Jersey filter
*/
@Bean
public FilterRegistrationBean jerseyFilterRegistration(
javax.ws.rs.core.Application eurekaJerseyApp) {
FilterRegistrationBean bean = new FilterRegistrationBean();
bean.setFilter(new ServletContainer(eurekaJerseyApp));
bean.setOrder(Ordered.LOWEST_PRECEDENCE);
bean.setUrlPatterns(
Collections.singletonList(EurekaConstants.DEFAULT_PREFIX + "/*"));
return bean;
}
/**
* Construct a Jersey {@link javax.ws.rs.core.Application} with all the resources
* required by the Eureka server.
*/
@Bean
public javax.ws.rs.core.Application jerseyApplication(Environment environment,
ResourceLoader resourceLoader) {
ClassPathScanningCandidateComponentProvider provider = new ClassPathScanningCandidateComponentProvider(
false, environment);
// Filter to include only classes that have a particular annotation.
//
provider.addIncludeFilter(new AnnotationTypeFilter(Path.class));
provider.addIncludeFilter(new AnnotationTypeFilter(Provider.class));
// Find classes in Eureka packages (or subpackages)
//
Set<Class<?>> classes = new HashSet<>();
for (String basePackage : EUREKA_PACKAGES) {
Set<BeanDefinition> beans = provider.findCandidateComponents(basePackage);
for (BeanDefinition bd : beans) {
Class<?> cls = ClassUtils.resolveClassName(bd.getBeanClassName(),
resourceLoader.getClassLoader());
classes.add(cls);
}
}
// Construct the Jersey ResourceConfig
//
Map<String, Object> propsAndFeatures = new HashMap<>();
propsAndFeatures.put(
// Skip static content used by the webapp
ServletContainer.PROPERTY_WEB_PAGE_CONTENT_REGEX,
EurekaConstants.DEFAULT_PREFIX + "/(fonts|images|css|js)/.*");
DefaultResourceConfig rc = new DefaultResourceConfig(classes);
rc.setPropertiesAndFeatures(propsAndFeatures);
return rc;
}
代碼中掃描的EUREKA_PACKAGES（private static final String[] EUREKA_PACKAGES = new String[] { "com.netflix.discovery",
"com.netflix.eureka" };）即是Jersey框架的具體的接口類
file
另外可以提一點的就是，eureka對外暴露的dashboard依然采用的是springmvc的controller形式。具體的可以看到在EurekaServerAutoConfiguration中注入的EurekaController
@Bean
@ConditionalOnProperty(prefix = "eureka.dashboard", name = "enabled", matchIfMissing = true)
public EurekaController eurekaController() {
return new EurekaController(this.applicationInfoManager);
}
感興趣的可以再研究下后續(xù)EurekaController的內(nèi)部實現(xiàn)
eureka注冊在客戶端和服務端分別怎么操作實現(xiàn)可用性的
服務每隔30秒會向注冊中?續(xù)約(?跳)?次（也稱為報活），如果沒有續(xù)約，租約在90秒后到期，然后服務會被失效。每隔30秒的續(xù)約操作我們稱之為?跳檢測
首先在服務端，通過上述的Jersey框架暴露的接口進行注冊，在ApplicationResource中通過addInstance進行注冊，在這個過程中另一個eureka server也相當于是一個eureka client，同樣會進行注冊
file
通過addInstance中的register方法，一直向下調(diào)試到PeerAwareInstanceRegistryImpl的replicateInstanceActionsToPeers相互注冊方法
/**
* Replicates all instance changes to peer eureka nodes except for
* replication traffic to this node.
*
*/
private void replicateInstanceActionsToPeers(Action action, String appName,
String id, InstanceInfo info, InstanceStatus newStatus,
PeerEurekaNode node) {
try {
InstanceInfo infoFromRegistry = null;
CurrentRequestVersion.set(Version.V2);
switch (action) {
case Cancel:
node.cancel(appName, id);
break;
case Heartbeat:
InstanceStatus overriddenStatus = overriddenInstanceStatusMap.get(id);
infoFromRegistry = getInstanceByAppAndId(appName, id, false);
node.heartbeat(appName, id, infoFromRegistry, overriddenStatus, false);
break;
case Register:
node.register(info);
break;
case StatusUpdate:
infoFromRegistry = getInstanceByAppAndId(appName, id, false);
node.statusUpdate(appName, id, newStatus, infoFromRegistry);
break;
case DeleteStatusOverride:
infoFromRegistry = getInstanceByAppAndId(appName, id, false);
node.deleteStatusOverride(appName, id, infoFromRegistry);
break;
}
} catch (Throwable t) {
logger.error("Cannot replicate information to {} for action {}", node.getServiceUrl(), action.name(), t);
}
}
此時，注冊時，進入的是Register
public void register(final InstanceInfo info) throws Exception {
long expiryTime = System.currentTimeMillis() + getLeaseRenewalOf(info);
batchingDispatcher.process(
taskId("register", info),
new InstanceReplicationTask(targetHost, Action.Register, info, null, true) {
public EurekaHttpResponse<Void> execute() {
return replicationClient.register(info);
}
},
expiryTime
);
}
查閱源碼可知此處的getLeaseRenewalOf(info)的默認值為90秒,這就印證了90秒到期的說法
private static int getLeaseRenewalOf(InstanceInfo info) {
return (info.getLeaseInfo() == null ? Lease.DEFAULT_DURATION_IN_SECS : info.getLeaseInfo().getRenewalIntervalInSecs()) * 1000;
}
發(fā)起請求
@Override
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();
}
}
}
發(fā)起請求的地址可以追蹤到的是ApplicationsResource中的
@Path("{appId}")
public ApplicationResource getApplicationResource(
@PathParam("version") String version,
@PathParam("appId") String appId) {
CurrentRequestVersion.set(Version.toEnum(version));
return new ApplicationResource(appId, serverConfig, registry);
}
此處重新構建了一個ApplicationResource對象。并將服務的信息配置等傳遞到application中，等待后續(xù)使用
分析完這一段之后,我對addInstance如何接收請求的還是有疑問，經(jīng)過斷點調(diào)試發(fā)現(xiàn)，這個過程實際上是通過EurekaServerAutoConfiguration引入的 EurekaServerInitializerConfiguration來完成的，
@Configuration
public class EurekaServerInitializerConfiguration
implements ServletContextAware, SmartLifecycle, Ordered {
}
EurekaServerInitializerConfiguration實現(xiàn)了SmartLifecycle方法，start方法會再容器初始化時執(zhí)行。而start方法的內(nèi)容
@Override
public void start() {
new Thread(new Runnable() {
@Override
public void run() {
try {
//TODO: is this class even needed now?
eurekaServerBootstrap.contextInitialized(EurekaServerInitializerConfiguration.this.servletContext);
log.info("Started Eureka Server");
publish(new EurekaRegistryAvailableEvent(getEurekaServerConfig()));
EurekaServerInitializerConfiguration.this.running = true;
publish(new EurekaServerStartedEvent(getEurekaServerConfig()));
}
catch (Exception ex) {
// Help!
log.error("Could not initialize Eureka servlet context", ex);
}
}
}).start();
}
具體的業(yè)務內(nèi)容在
public void contextInitialized(ServletContext context) {
try {
initEurekaEnvironment();
initEurekaServerContext();
context.setAttribute(EurekaServerContext.class.getName(), this.serverContext);
}
catch (Throwable e) {
log.error("Cannot bootstrap eureka server :", e);
throw new RuntimeException("Cannot bootstrap eureka server :", e);
}
}
第一步initEurekaEnvironment為初始化環(huán)境，第二步initEurekaServerContext為業(yè)務操作
而隨后的操作中最主要的是
int registryCount = this.registry.syncUp();
this.registry.openForTraffic(this.applicationInfoManager, registryCount);
// Register all monitoring statistics.
EurekaMonitors.registerAllStats();
openForTraffic中主要是為開啟服務通信做準備
@Override
public void openForTraffic(ApplicationInfoManager applicationInfoManager, int count) {
// Renewals happen every 30 seconds and for a minute it should be a factor of 2.
this.expectedNumberOfClientsSendingRenews = count;
updateRenewsPerMinThreshold();
logger.info("Got {} instances from neighboring DS node", count);
logger.info("Renew threshold is: {}", numberOfRenewsPerMinThreshold);
this.startupTime = System.currentTimeMillis();
if (count > 0) {
this.peerInstancesTransferEmptyOnStartup = false;
}
DataCenterInfo.Name selfName = applicationInfoManager.getInfo().getDataCenterInfo().getName();
boolean isAws = Name.Amazon == selfName;
if (isAws && serverConfig.shouldPrimeAwsReplicaConnections()) {
logger.info("Priming AWS connections for all replicas..");
primeAwsReplicas(applicationInfoManager);
}
logger.info("Changing status to UP");
applicationInfoManager.setInstanceStatus(InstanceStatus.UP);
super.postInit();
}
引發(fā)向addIntsance發(fā)起請求的就是 applicationInfoManager.setInstanceStatus(InstanceStatus.UP);這個方法內(nèi)部執(zhí)行一串事件
其中就有向addInstance發(fā)起請求的
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);
}
}
}
}
DiscoveryClient類內(nèi)部
statusChangeListener = new ApplicationInfoManager.StatusChangeListener() {
@Override
public String getId() {
return "statusChangeListener";
}
@Override
public void notify(StatusChangeEvent statusChangeEvent) {
if (InstanceStatus.DOWN == statusChangeEvent.getStatus() ||
InstanceStatus.DOWN == statusChangeEvent.getPreviousStatus()) {
// log at warn level if DOWN was involved
logger.warn("Saw local status change event {}", statusChangeEvent);
} else {
logger.info("Saw local status change event {}", statusChangeEvent);
}
instanceInfoReplicator.onDemandUpdate();
}
};
指向
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 {
Future next = scheduler.schedule(this, replicationIntervalSeconds, TimeUnit.SECONDS);
scheduledPeriodicRef.set(next);
}
}
最后即是向addInstance發(fā)起請求的地方
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();
}
發(fā)起請求即是向ApplicationResource的Instance方法發(fā)起。
eureka續(xù)約在客戶端和服務端分別怎么操作實現(xiàn)可用性的
從上面注冊中可推測出續(xù)約/心跳接口可能也是在DiscoveryClient中完成的。搜索HeatBeat之后發(fā)現(xiàn)注入 DiscoveryClient方法中有一個初始化定時任務的方法
private void initScheduledTasks() {
if (clientConfig.shouldFetchRegistry()) {
// registry cache refresh timer
int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();
int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();
scheduler.schedule(
new TimedSupervisorTask(
"cacheRefresh",
scheduler,
cacheRefreshExecutor,
registryFetchIntervalSeconds,
TimeUnit.SECONDS,
expBackOffBound,
new CacheRefreshThread()
),
registryFetchIntervalSeconds, TimeUnit.SECONDS);
}
if (clientConfig.shouldRegisterWithEureka()) {
int renewalIntervalInSecs = instanceInfo.getLeaseInfo().getRenewalIntervalInSecs();
int expBackOffBound = clientConfig.getHeartbeatExecutorExponentialBackOffBound();
logger.info("Starting heartbeat executor: " + "renew interval is: {}", renewalIntervalInSecs);
// Heartbeat timer
scheduler.schedule(
new TimedSupervisorTask(
"heartbeat",
scheduler,
heartbeatExecutor,
renewalIntervalInSecs,
TimeUnit.SECONDS,
expBackOffBound,
new HeartbeatThread()
),
renewalIntervalInSecs, TimeUnit.SECONDS);
其中就有心跳的定時任務。默認的心跳間隔時間renewalIntervalInSecs為30秒
/**
* The heartbeat task that renews the lease in the given intervals.
*/
private class HeartbeatThread implements Runnable {
public void run() {
if (renew()) {
lastSuccessfulHeartbeatTimestamp = System.currentTimeMillis();
}
}
}
renew方法中即是向服務端發(fā)起調(diào)用的過程，與上述注冊基本相同
Eureka下線服務
Eureka下線是在EurekaClientAutoConfiguration中注入EurekaClient時定義的shutDown方法。
我們可以看到
@PreDestroy
@Override
public synchronized void shutdown() {
if (isShutdown.compareAndSet(false, true)) {
logger.info("Shutting down DiscoveryClient ...");
if (statusChangeListener != null && applicationInfoManager != null) {
applicationInfoManager.unregisterStatusChangeListener(statusChangeListener.getId());
}
cancelScheduledTasks();
// If APPINFO was registered
if (applicationInfoManager != null
&& clientConfig.shouldRegisterWithEureka()
&& clientConfig.shouldUnregisterOnShutdown()) {
applicationInfoManager.setInstanceStatus(InstanceStatus.DOWN);
unregister();
}
if (eurekaTransport != null) {
eurekaTransport.shutdown();
}
heartbeatStalenessMonitor.shutdown();
registryStalenessMonitor.shutdown();
logger.info("Completed shut down of DiscoveryClient");
}
}
同樣的。執(zhí)行了一個取消定時任務的狀態(tài)。。另外利用上面說的applicationInfoManager.setInstanceStatus()方法進行了事件通知，另外unregister();進行了取消注冊操作。eurekaTransport.shutdown();關閉傳輸。
Eureka的功能特性總體上來說就是這樣。有些地方可能還是不夠清楚。歡迎大家一起溝通探討
歡迎關注和點贊，以及總結的分類面試題https://github.com/zhendiao/JavaInterview
