聊聊nacos的ServerListManager
阿新 • • 發佈:2019-12-31
序
本文主要研究一下nacos的ServerListManager
ServerListManager
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/cluster/ServerListManager.java
@Component("serverListManager")
public class ServerListManager {
private static final int STABLE_PERIOD = 60 * 1000;
@Autowired
private SwitchDomain switchDomain;
private List<ServerChangeListener> listeners = new ArrayList<>();
private List<Server> servers = new ArrayList<>();
private List<Server> healthyServers = new ArrayList<>();
private Map<String,List<Server>> distroConfig = new ConcurrentHashMap<>();
private Map<String,Long> distroBeats = new ConcurrentHashMap<>(16);
private Set<String> liveSites = new HashSet<>();
private final static String LOCALHOST_SITE = UtilsAndCommons.UNKNOWN_SITE;
private long lastHealthServerMillis = 0L;
private boolean autoDisabledHealthCheck = false ;
private Synchronizer synchronizer = new ServerStatusSynchronizer();
public void listen(ServerChangeListener listener) {
listeners.add(listener);
}
@PostConstruct
public void init() {
GlobalExecutor.registerServerListUpdater(new ServerListUpdater());
GlobalExecutor.registerServerStatusReporter(new ServerStatusReporter(),5000);
}
//......
}
複製程式碼
- ServerListManager的init方法註冊了ServerListUpdater、ServerStatusReporter兩個定時任務
GlobalExecutor
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/misc/GlobalExecutor.java
public class GlobalExecutor {
public static final long HEARTBEAT_INTERVAL_MS = TimeUnit.SECONDS.toMillis(5L);
public static final long LEADER_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(15L);
public static final long RANDOM_MS = TimeUnit.SECONDS.toMillis(5L);
public static final long TICK_PERIOD_MS = TimeUnit.MILLISECONDS.toMillis(500L);
private static final long NACOS_SERVER_LIST_REFRESH_INTERVAL = TimeUnit.SECONDS.toMillis(5);
private static final long PARTITION_DATA_TIMED_SYNC_INTERVAL = TimeUnit.SECONDS.toMillis(5);
private static final long SERVER_STATUS_UPDATE_PERIOD = TimeUnit.SECONDS.toMillis(5);
//......
public static void registerServerListUpdater(Runnable runnable) {
executorService.scheduleAtFixedRate(runnable,NACOS_SERVER_LIST_REFRESH_INTERVAL,TimeUnit.MILLISECONDS);
}
public static void registerServerStatusReporter(Runnable runnable,long delay) {
SERVER_STATUS_EXECUTOR.schedule(runnable,delay,TimeUnit.MILLISECONDS);
}
//......
}
複製程式碼
- registerServerListUpdater是每隔NACOS_SERVER_LIST_REFRESH_INTERVAL排程一次,預設是5秒;registerServerStatusReporter則是排程一個延時任務
ServerListUpdater
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/cluster/ServerListManager.java
public class ServerListUpdater implements Runnable {
@Override
public void run() {
try {
List<Server> refreshedServers = refreshServerList();
List<Server> oldServers = servers;
if (CollectionUtils.isEmpty(refreshedServers)) {
Loggers.RAFT.warn("refresh server list failed,ignore it.");
return;
}
boolean changed = false;
List<Server> newServers = (List<Server>) CollectionUtils.subtract(refreshedServers,oldServers);
if (CollectionUtils.isNotEmpty(newServers)) {
servers.addAll(newServers);
changed = true;
Loggers.RAFT.info("server list is updated,new: {} servers: {}",newServers.size(),newServers);
}
List<Server> deadServers = (List<Server>) CollectionUtils.subtract(oldServers,refreshedServers);
if (CollectionUtils.isNotEmpty(deadServers)) {
servers.removeAll(deadServers);
changed = true;
Loggers.RAFT.info("server list is updated,dead: {},servers: {}",deadServers.size(),deadServers);
}
if (changed) {
notifyListeners();
}
} catch (Exception e) {
Loggers.RAFT.info("error while updating server list.",e);
}
}
}
private List<Server> refreshServerList() {
List<Server> result = new ArrayList<>();
if (STANDALONE_MODE) {
Server server = new Server();
server.setIp(NetUtils.getLocalAddress());
server.setServePort(RunningConfig.getServerPort());
result.add(server);
return result;
}
List<String> serverList = new ArrayList<>();
try {
serverList = readClusterConf();
} catch (Exception e) {
Loggers.SRV_LOG.warn("failed to get config: " + CLUSTER_CONF_FILE_PATH,e);
}
if (Loggers.SRV_LOG.isDebugEnabled()) {
Loggers.SRV_LOG.debug("SERVER-LIST from cluster.conf: {}",result);
}
//use system env
if (CollectionUtils.isEmpty(serverList)) {
serverList = SystemUtils.getIPsBySystemEnv(UtilsAndCommons.SELF_SERVICE_CLUSTER_ENV);
if (Loggers.SRV_LOG.isDebugEnabled()) {
Loggers.SRV_LOG.debug("SERVER-LIST from system variable: {}",result);
}
}
if (CollectionUtils.isNotEmpty(serverList)) {
for (int i = 0; i < serverList.size(); i++) {
String ip;
int port;
String server = serverList.get(i);
if (server.contains(UtilsAndCommons.IP_PORT_SPLITER)) {
ip = server.split(UtilsAndCommons.IP_PORT_SPLITER)[0];
port = Integer.parseInt(server.split(UtilsAndCommons.IP_PORT_SPLITER)[1]);
} else {
ip = server;
port = RunningConfig.getServerPort();
}
Server member = new Server();
member.setIp(ip);
member.setServePort(port);
result.add(member);
}
}
return result;
}
private void notifyListeners() {
GlobalExecutor.notifyServerListChange(new Runnable() {
@Override
public void run() {
for (ServerChangeListener listener : listeners) {
listener.onChangeServerList(servers);
listener.onChangeHealthyServerList(healthyServers);
}
}
});
}
複製程式碼
- ServerListUpdater實現了Runnable介面,其run方法會通過refreshServerList方法從配置檔案讀取最新的servers配置,然後對比oldServers,看哪些是新增,哪些是刪除的,如果確實有變更則通過notifyListeners方法進行通知回撥ServerChangeListener的onChangeServerList、onChangeHealthyServerList方法
ServerStatusReporter
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/cluster/ServerListManager.java
private class ServerStatusReporter implements Runnable {
@Override
public void run() {
try {
if (RunningConfig.getServerPort() <= 0) {
return;
}
checkDistroHeartbeat();
int weight = Runtime.getRuntime().availableProcessors() / 2;
if (weight <= 0) {
weight = 1;
}
long curTime = System.currentTimeMillis();
String status = LOCALHOST_SITE + "#" + NetUtils.localServer() + "#" + curTime + "#" + weight + "\r\n";
//send status to itself
onReceiveServerStatus(status);
List<Server> allServers = getServers();
if (!contains(NetUtils.localServer())) {
Loggers.SRV_LOG.error("local ip is not in serverlist,ip: {},serverlist: {}",NetUtils.localServer(),allServers);
return;
}
if (allServers.size() > 0 && !NetUtils.localServer().contains(UtilsAndCommons.LOCAL_HOST_IP)) {
for (com.alibaba.nacos.naming.cluster.servers.Server server : allServers) {
if (server.getKey().equals(NetUtils.localServer())) {
continue;
}
Message msg = new Message();
msg.setData(status);
synchronizer.send(server.getKey(),msg);
}
}
} catch (Exception e) {
Loggers.SRV_LOG.error("[SERVER-STATUS] Exception while sending server status",e);
} finally {
GlobalExecutor.registerServerStatusReporter(this,switchDomain.getServerStatusSynchronizationPeriodMillis());
}
}
}
複製程式碼
- ServerStatusReporter實現了Runnable介面,其run方法首先執行checkDistroHeartbeat,然後執行onReceiveServerStatus,最後獲取servers列表通知通過synchronizer.send(server.getKey(),msg)方法通知其他server自己的狀態;最後再次通過registerServerStatusReporter排程下次執行的時間
checkDistroHeartbeat
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/cluster/ServerListManager.java
private void checkDistroHeartbeat() {
Loggers.SRV_LOG.debug("check distro heartbeat.");
List<Server> servers = distroConfig.get(LOCALHOST_SITE);
if (CollectionUtils.isEmpty(servers)) {
return;
}
List<Server> newHealthyList = new ArrayList<>(servers.size());
long now = System.currentTimeMillis();
for (Server s: servers) {
Long lastBeat = distroBeats.get(s.getKey());
if (null == lastBeat) {
continue;
}
s.setAlive(now - lastBeat < switchDomain.getDistroServerExpiredMillis());
}
//local site servers
List<String> allLocalSiteSrvs = new ArrayList<>();
for (Server server : servers) {
if (server.getKey().endsWith(":0")) {
continue;
}
server.setAdWeight(switchDomain.getAdWeight(server.getKey()) == null ? 0 : switchDomain.getAdWeight(server.getKey()));
for (int i = 0; i < server.getWeight() + server.getAdWeight(); i++) {
if (!allLocalSiteSrvs.contains(server.getKey())) {
allLocalSiteSrvs.add(server.getKey());
}
if (server.isAlive() && !newHealthyList.contains(server)) {
newHealthyList.add(server);
}
}
}
Collections.sort(newHealthyList);
float curRatio = (float) newHealthyList.size() / allLocalSiteSrvs.size();
if (autoDisabledHealthCheck
&& curRatio > switchDomain.getDistroThreshold()
&& System.currentTimeMillis() - lastHealthServerMillis > STABLE_PERIOD) {
Loggers.SRV_LOG.info("[NACOS-DISTRO] distro threshold restored and " +
"stable now,enable health check. current ratio: {}",curRatio);
switchDomain.setHealthCheckEnabled(true);
// we must set this variable,otherwise it will conflict with user's action
autoDisabledHealthCheck = false;
}
if (!CollectionUtils.isEqualCollection(healthyServers,newHealthyList)) {
// for every change disable healthy check for some while
if (switchDomain.isHealthCheckEnabled()) {
Loggers.SRV_LOG.info("[NACOS-DISTRO] healthy server list changed," +
"disable health check for {} ms from now on,old: {},new: {}",STABLE_PERIOD,healthyServers,newHealthyList);
switchDomain.setHealthCheckEnabled(false);
autoDisabledHealthCheck = true;
lastHealthServerMillis = System.currentTimeMillis();
}
healthyServers = newHealthyList;
notifyListeners();
}
}
複製程式碼
- checkDistroHeartbeat方法會遍歷servers從distroBeats獲取lastBeat資訊,然後判斷距離lastBeat的時間是否小於distroServerExpiredMillis,小於則alive為true,否則alive為false;之後遍歷servers更新adWeight
- 然後計算curRatio,如果大於distroThreshold且autoDisabledHealthCheck為true且距離lastHealthServerMillis大於STABLE_PERIOD則會執行switchDomain.setHealthCheckEnabled(true)並更新autoDisabledHealthCheck為false
- 最後判斷healthyServers與newHealthyList是否一致,不一致且switchDomain.isHealthCheckEnabled()則會執行switchDomain.setHealthCheckEnabled(false)且設定autoDisabledHealthCheck為true,更新lastHealthServerMillis;最後更新healthyServers,執行notifyListeners方法
小結
- ServerListManager的init方法註冊了ServerListUpdater、ServerStatusReporter兩個定時任務
- ServerListUpdater實現了Runnable介面,其run方法會通過refreshServerList方法從配置檔案讀取最新的servers配置,然後對比oldServers,看哪些是新增,哪些是刪除的,如果確實有變更則通過notifyListeners方法進行通知回撥ServerChangeListener的onChangeServerList、onChangeHealthyServerList方法
- ServerStatusReporter實現了Runnable介面,其run方法首先執行checkDistroHeartbeat,然後執行onReceiveServerStatus,最後獲取servers列表通知通過synchronizer.send(server.getKey(),msg)方法通知其他server自己的狀態;最後再次通過registerServerStatusReporter排程下次執行的時間