聊聊spring cloud的DefaultEurekaServerContext
序
本文主要研究一下spring cloud的DefaultEurekaServerContext
EurekaServerAutoConfiguration
@Configuration
@Import(EurekaServerInitializerConfiguration.class)
@ConditionalOnBean(EurekaServerMarkerConfiguration.Marker.class)
@EnableConfigurationProperties({ EurekaDashboardProperties.class,
InstanceRegistryProperties.class })
@PropertySource("classpath:/eureka/server.properties")
public class EurekaServerAutoConfiguration extends WebMvcConfigurerAdapter {
//......
@Bean
public EurekaServerContext eurekaServerContext(ServerCodecs serverCodecs,
PeerAwareInstanceRegistry registry, PeerEurekaNodes peerEurekaNodes) {
return new DefaultEurekaServerContext(this.eurekaServerConfig, serverCodecs,
registry, peerEurekaNodes, this.applicationInfoManager);
}
//......
}
DefaultEurekaServerContext
eureka-core-1.8.8-sources.jar!/com/netflix/eureka/DefaultEurekaServerContext.java
@PostConstruct
@Override
public void initialize() throws Exception {
logger.info("Initializing ...");
peerEurekaNodes.start();
registry.init(peerEurekaNodes);
logger.info("Initialized");
}
@PreDestroy
@Override
public void shutdown() throws Exception {
logger.info("Shutting down ...");
registry.shutdown();
peerEurekaNodes.shutdown();
logger.info("Shut down");
}
实例化后的时候执行peerEurekaNodes.start();以及registry.init(peerEurekaNodes); 销毁之前执行registry.shutdown();以及peerEurekaNodes.shutdown();
start
PeerEurekaNodes.start
eureka-core-1.8.8-sources.jar!/com/netflix/eureka/cluster/PeerEurekaNodes.java
public void start() {
taskExecutor = Executors.newSingleThreadScheduledExecutor(
new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r, "Eureka-PeerNodesUpdater");
thread.setDaemon(true);
return thread;
}
}
);
try {
updatePeerEurekaNodes(resolvePeerUrls());
Runnable peersUpdateTask = new Runnable() {
@Override
public void run() {
try {
updatePeerEurekaNodes(resolvePeerUrls());
} catch (Throwable e) {
logger.error("Cannot update the replica Nodes", e);
}
}
};
taskExecutor.scheduleWithFixedDelay(
peersUpdateTask,
serverConfig.getPeerEurekaNodesUpdateIntervalMs(),
serverConfig.getPeerEurekaNodesUpdateIntervalMs(),
TimeUnit.MILLISECONDS
);
} catch (Exception e) {
throw new IllegalStateException(e);
}
for (PeerEurekaNode node : peerEurekaNodes) {
logger.info("Replica node URL: {}", node.getServiceUrl());
}
}
这里首先执行updatePeerEurekaNodes,之后注册定时任务去定时触发updatePeerEurekaNodes,时间间隔为erverConfig.getPeerEurekaNodesUpdateIntervalMs()
resolvePeerUrls
/**
* Resolve peer URLs.
*
* @return peer URLs with node's own URL filtered out
*/
protected List<String> resolvePeerUrls() {
InstanceInfo myInfo = applicationInfoManager.getInfo();
String zone = InstanceInfo.getZone(clientConfig.getAvailabilityZones(clientConfig.getRegion()), myInfo);
List<String> replicaUrls = EndpointUtils
.getDiscoveryServiceUrls(clientConfig, zone, new EndpointUtils.InstanceInfoBasedUrlRandomizer(myInfo));
int idx = 0;
while (idx < replicaUrls.size()) {
if (isThisMyUrl(replicaUrls.get(idx))) {
replicaUrls.remove(idx);
} else {
idx++;
}
}
return replicaUrls;
}
首先通过resolvePeerUrls来获取replicaUrls,这里获取的是健康的url,然后剔除自己的url
updatePeerEurekaNodes
/**
* Given new set of replica URLs, destroy {@link PeerEurekaNode}s no longer available, and
* create new ones.
*
* @param newPeerUrls peer node URLs; this collection should have local node's URL filtered out
*/
protected void updatePeerEurekaNodes(List<String> newPeerUrls) {
if (newPeerUrls.isEmpty()) {
logger.warn("The replica size seems to be empty. Check the route 53 DNS Registry");
return;
}
Set<String> toShutdown = new HashSet<>(peerEurekaNodeUrls);
toShutdown.removeAll(newPeerUrls);
Set<String> toAdd = new HashSet<>(newPeerUrls);
toAdd.removeAll(peerEurekaNodeUrls);
if (toShutdown.isEmpty() && toAdd.isEmpty()) { // No change
return;
}
// Remove peers no long available
List<PeerEurekaNode> newNodeList = new ArrayList<>(peerEurekaNodes);
if (!toShutdown.isEmpty()) {
logger.info("Removing no longer available peer nodes {}", toShutdown);
int i = 0;
while (i < newNodeList.size()) {
PeerEurekaNode eurekaNode = newNodeList.get(i);
if (toShutdown.contains(eurekaNode.getServiceUrl())) {
newNodeList.remove(i);
eurekaNode.shutDown();
} else {
i++;
}
}
}
// Add new peers
if (!toAdd.isEmpty()) {
logger.info("Adding new peer nodes {}", toAdd);
for (String peerUrl : toAdd) {
newNodeList.add(createPeerEurekaNode(peerUrl));
}
}
this.peerEurekaNodes = newNodeList;
this.peerEurekaNodeUrls = new HashSet<>(newPeerUrls);
}
主要是跟原始或上次更新的peerEurekaNodeUrls对比,移除掉不健康的节点,移除的时候会调用PeerEurekaNode的shutdown方法,添加的时候通过createPeerEurekaNode创建
protected PeerEurekaNode createPeerEurekaNode(String peerEurekaNodeUrl) {
HttpReplicationClient replicationClient = JerseyReplicationClient.createReplicationClient(serverConfig, serverCodecs, peerEurekaNodeUrl);
String targetHost = hostFromUrl(peerEurekaNodeUrl);
if (targetHost == null) {
targetHost = "host";
}
return new PeerEurekaNode(registry, targetHost, peerEurekaNodeUrl, replicationClient, serverConfig);
}
添加的时候会创建新的PeerEurekaNode
registry.init(peerEurekaNodes)
eureka-core-1.8.8-sources.jar!/com/netflix/eureka/registry/PeerAwareInstanceRegistryImpl.java
public void init(PeerEurekaNodes peerEurekaNodes) throws Exception {
this.numberOfReplicationsLastMin.start();
this.peerEurekaNodes = peerEurekaNodes;
initializedResponseCache();
scheduleRenewalThresholdUpdateTask();
initRemoteRegionRegistry();
try {
Monitors.registerObject(this);
} catch (Throwable e) {
logger.warn("Cannot register the JMX monitor for the InstanceRegistry :", e);
}
}
这里初始化ResponseCache、调度RenewalThresholdUpdateTask、还有初始化RemoteRegionRegistry
shutdown
registry.shutdown()
eureka-core-1.8.8-sources.jar!/com/netflix/eureka/registry/PeerAwareInstanceRegistryImpl.java
/**
* Perform all cleanup and shutdown operations.
*/
@Override
public void shutdown() {
try {
DefaultMonitorRegistry.getInstance().unregister(Monitors.newObjectMonitor(this));
} catch (Throwable t) {
logger.error("Cannot shutdown monitor registry", t);
}
try {
peerEurekaNodes.shutdown();
} catch (Throwable t) {
logger.error("Cannot shutdown ReplicaAwareInstanceRegistry", t);
}
numberOfReplicationsLastMin.stop();
super.shutdown();
}
这里主要是调用peerEurekaNodes.shutdown(),还有super的shutdown
super.shutdown
eureka-core-1.8.8-sources.jar!/com/netflix/eureka/registry/AbstractInstanceRegistry.java
/**
* Perform all cleanup and shutdown operations.
*/
@Override
public void shutdown() {
deltaRetentionTimer.cancel();
evictionTimer.cancel();
renewsLastMin.stop();
}
主要是关闭一些计时器
peerEurekaNodes.shutdown()
eureka-core-1.8.8-sources.jar!/com/netflix/eureka/cluster/PeerEurekaNodes.java
public void shutdown() {
taskExecutor.shutdown();
List<PeerEurekaNode> toRemove = this.peerEurekaNodes;
this.peerEurekaNodes = Collections.emptyList();
this.peerEurekaNodeUrls = Collections.emptySet();
for (PeerEurekaNode node : toRemove) {
node.shutDown();
}
}
除了系统关闭会调用外,该eureka node被认为不健康的时候,被剔除时,也会调用shutdown方法
PeerEurekaNode.shutdown()
eureka-core-1.8.8-sources.jar!/com/netflix/eureka/cluster/PeerEurekaNode.java
/**
* Shuts down all resources used for peer replication.
*/
public void shutDown() {
batchingDispatcher.shutdown();
nonBatchingDispatcher.shutdown();
}
主要是关闭dispatcher
小结
EurekaServerContext主要注册了bean初始化及销毁时执行的操作。初始化时,启动peerNodes,然后初始化registry;销毁时,关闭registry,然后关闭peerNodes。关于peerNodes,主要的是定时任务以erverConfig.getPeerEurekaNodesUpdateIntervalMs()时间间隔去定时触发updatePeerEurekaNodes,而这个操作是跟原始或上次更新的peerEurekaNodeUrls对比,移除掉不健康的节点,添加新的节点如果有的话。移除的时候会调用PeerEurekaNode的shutdown方法,添加新的peerNode的时候通过createPeerEurekaNode创建。
doc
- Understanding Eureka Peer to Peer Communication
- java教程
- Java快速入门
- Java 开发环境配置
- Java基本语法
- Java 对象和类
- Java 基本数据类型
- Java 变量类型
- Java 修饰符
- Java 运算符
- Java 循环结构
- Java 分支结构
- Java Number类
- Java Character类
- Java String类
- Java StringBuffer和StringBuilder类
- Java 数组
- Java 日期时间
- Java 正则表达式
- Java 方法
- Java 流(Stream)、文件(File)和IO
- Java 异常处理
- Java 继承
- Java 重写(Override)与重载(Overload)
- Java 多态
- Java 抽象类
- Java 封装
- Java 接口
- Java 包(package)
- Java 数据结构
- Java 集合框架
- Java 泛型
- Java 序列化
- Java 网络编程
- Java 发送邮件
- Java 多线程编程
- Java Applet基础
- Java 文档注释
- Python实现SMTP邮件发送
- 浅谈多卡服务器下隐藏部分 GPU 和 TensorFlow 的显存使用设置
- 通过PHP设置BugFree获取邮箱通知
- Django封装交互接口代码
- 使用K.function()调试keras操作
- tensorflow图像裁剪进行数据增强操作
- ThinkPHP3.2.3框架Memcache缓存使用方法实例总结
- Python+PyQt5+MySQL实现天气管理系统
- 浅谈Python协程
- Python logging模块异步线程写日志实现过程解析
- php-fpm重启导致的程序执行中断问题详解
- Python Socket TCP双端聊天功能实现过程详解
- django 将自带的数据库sqlite3改成mysql实例
- 利用python对mysql表做全局模糊搜索并分页实例
- PHP chop()函数讲解