您好,登錄后才能下訂單哦!
上一章節(jié),我們從全局了解了一下Nacos項目的模塊架構(gòu),做到了心中有數(shù),現(xiàn)在,我們?nèi)ブ鸩饺ネ诰蚶锩娴拇a細節(jié),很多人在學習開源的時候,無從下手,代碼那么多,從哪個地方開始看呢?我們可以從一個接口開始入手,這個接口是你使用過的,知道它大概做什么事,有體感的,大家還記得第一章時,我們寫的HelloWorld嗎,對,就從里面的接口開始剝洋蔥。
https://github.com/alibaba/nacos,這個是Nacos的github代碼地址,開始之前先start關(guān)注一下,加上watch,后續(xù)Nacos的郵件列表也會通知到你,可以關(guān)注到Nacos的最新實時消息,及各大牛之間的精彩討論。
下面這段代碼,是第一章節(jié)發(fā)布一個服務的代碼:
public static void main(String[] args) throws NacosException, InterruptedException { //發(fā)布的服務名 String serviceName = "helloworld.services"; //構(gòu)造一個Nacos實例,入?yún)⑹荖acos server的ip和服務端口 NamingService naming = NacosFactory.createNamingService("100.81.0.34:8080"); //發(fā)布一個服務,該服務對外提供的ip為127.0.0.1,端口為8888 naming.registerInstance(serviceName, "100.81.0.35", 8080); Thread.sleep(Integer.MAX_VALUE);}
其中,第一步,是構(gòu)造一個Nacos服務實例,構(gòu)造實例的入?yún)?,是一個String,值的規(guī)范為ip:port,這個ip,就是我們?nèi)我庖慌_Nacos server的地址,我們點進去看這個方法:
public static NamingService createNamingService(String serverAddr) throws NacosException { return NamingFactory.createNamingService(serverAddr); }
同時我們看下創(chuàng)建配置服務實例的代碼:
public static ConfigService createConfigService(String serverAddr) throws NacosException { return ConfigFactory.createConfigService(serverAddr);}
我們可以看到,NacosFactory實際上是一個服務發(fā)現(xiàn)和配置管理接口的統(tǒng)一入口,再由它不通的方法,創(chuàng)建不同服務的實例,我們可以直接使用NamingFactory,或者ConfigFactory直接創(chuàng)建Nacos的服務實例,也能work
cdn.nlark.com/lark/0/2018/png/4232/1537627839251-445ff4a4-d8ec-4cb0-abf5-ffd86d771bec.png">
接下來,看一下,是如何構(gòu)造出這個Nacos naming實例的:
public static NamingService createNamingService(String serverList) throws NacosException { try { Class<?> driverImplClass = Class.forName("com.alibaba.nacos.client.naming.NacosNamingService"); Constructor constructor = driverImplClass.getConstructor(String.class); NamingService vendorImpl = (NamingService) constructor.newInstance(serverList); return vendorImpl; } catch (Throwable e) { throw new NacosException(-400, e.getMessage()); }}
通過反射實例化出了一個NamingService的實例NacosNamingService,構(gòu)造器是一個帶String入?yún)⒌模覀冺樦驴?,?gòu)造函數(shù)里面做了哪些事情:
public NacosNamingService(String serverList) { this.serverList = serverList; init(); eventDispatcher = new EventDispatcher(); serverProxy = new NamingProxy(namespace, endpoint, serverList); beatReactor = new BeatReactor(serverProxy); hostReactor = new HostReactor(eventDispatcher, serverProxy, cacheDir);}
入?yún)erverList就是我們剛才傳入的服務端地址,值賦給了實例的serverList字段,接下來調(diào)用了一個init方法,這個方法里面如下:
private void init() { namespace = System.getProperty(PropertyKeyConst.NAMESPACE); if (StringUtils.isEmpty(namespace)) { namespace = UtilAndComs.DEFAULT_NAMESPACE_ID; } logName = System.getProperty(UtilAndComs.NACOS_NAMING_LOG_NAME); if (StringUtils.isEmpty(logName)) { logName = "naming.log"; } cacheDir = System.getProperty("com.alibaba.nacos.naming.cache.dir"); if (StringUtils.isEmpty(cacheDir)) { cacheDir = System.getProperty("user.home") + "/nacos/naming/" + namespace; }}
這面做了3件事,給namespace,logName,cacheDir賦值,namespace我們么有傳入,默認是default,namespace在Nacos里面的作用,是用來進行本地緩存隔離的,一臺機器上,啟動一個Nacos的客戶端進程,默認的本地緩存路徑是default,如果再啟動一個,需要重新設(shè)置一個namespace,否則就會復用之前的緩存,造成沖突;logName和cacheDir,這2個字段就不解釋了,字面理解。這里多說一句,這些值的設(shè)置,可以在java啟動時,通過系統(tǒng)參數(shù)的形式傳入,并且是第一優(yōu)先級的。
init方法執(zhí)行完之后,接下來是實例化一些框架組件,EventDispatcher,這個是一個經(jīng)典的事件分發(fā)組件,它的工作模式如下:
會有一個單獨線程從blockQueue中獲取事件,這個事件在Nacos這里, 就是服務端推送下來的數(shù)據(jù),listener在我們訂閱一條數(shù)據(jù)時,會從生成一個listener實例,在事件到了隊列中,找到對應的listener,去執(zhí)行里面listener的回調(diào)函數(shù)onEvent。如果對這個模式不熟悉的同學,可以再翻看下EventDispatcher的代碼,這個屬于基礎(chǔ)知識了,和業(yè)務沒有關(guān)系,這里就不過多詳細講解,篇幅太長。
接下來,實例化了一個NameProxy的組件,這個東西是干嘛的呢?我們看下里面代碼:
public NamingProxy(String namespace, String endpoint, String serverList) { this.namespace = namespace; this.endpoint = endpoint; if (StringUtils.isNotEmpty(serverList)) { this.serverList = Arrays.asList(serverList.split(",")); if (this.serverList.size() == 1) { this.nacosDomain = serverList; } } executorService = new ScheduledThreadPoolExecutor(1, new ThreadFactory() { @Override public Thread newThread(Runnable r) { Thread t = new Thread(r); t.setName("com.taobao.vipserver.serverlist.updater"); t.setDaemon(true); return t; } }); executorService.scheduleWithFixedDelay(new Runnable() { @Override public void run() { refreshSrvIfNeed(); } }, 0, vipSrvRefInterMillis, TimeUnit.MILLISECONDS); refreshSrvIfNeed();}
這里面邏輯有些多,我總結(jié)下,主要是啟動了一個線程,每隔30s,去執(zhí)行refreshSrvIfNeed()這個方法,
refreshSrvIfNeed()這個方法里面,做的事情,是通過一個http請求,去Nacos server獲取一串Nacos server集群的地址列表,具體代碼如下:
private void refreshSrvIfNeed() { try { if (!CollectionUtils.isEmpty(serverList)) { LogUtils.LOG.info("server list provided by user: " + serverList); return; } if (System.currentTimeMillis() - lastSrvRefTime < vipSrvRefInterMillis) { return; } List<String> list = getServerListFromEndpoint(); if (list.isEmpty()) { throw new Exception("Can not acquire vipserver list"); } if (!CollectionUtils.isEqualCollection(list, serversFromEndpoint)) { LogUtils.LOG.info("SERVER-LIST", "server list is updated: " + list); } serversFromEndpoint = list; lastSrvRefTime = System.currentTimeMillis(); } catch (Throwable e) { LogUtils.LOG.warn("failed to update server list", e); } }
獲取完地址列表后,賦值給serversFromEndpoint,并且記錄當前更新時間,在下一次更新時,小于30s,就不更新,避免頻繁更新,總的來說,NameProxy的目的就是定時在客戶端維護Nacos服務端的最新地址列表。
我們繼續(xù)往下看,接下來初始化了BeatReactor這個組件,從名字可以猜測,應該是和心跳相關(guān)的事情,它初始化的代碼如下:
public BeatReactor(NamingProxy serverProxy) { this.serverProxy = serverProxy; executorService.scheduleAtFixedRate(new BeatProcessor(), 0, clientBeatInterval, TimeUnit.MILLISECONDS);}
起了一個定時間隔為10s的任務,去執(zhí)行BeatProcessor里面的邏輯,BeatProcessor的代碼里面,是循環(huán)的去取當前客戶端注冊好的實例,然后向服務端發(fā)送一個http的心跳通知請求,告訴客戶端,這個服務的健康狀態(tài),具體代碼如下:
class BeatTask implements Runnable { BeatInfo beatInfo; public BeatTask(BeatInfo beatInfo) { this.beatInfo = beatInfo; } @Override public void run() { Map<String, String> params = new HashMap<String, String>(2); params.put("beat", JSON.toJSONString(beatInfo)); params.put("dom", beatInfo.getDom()); try { String result = serverProxy.callAllServers(UtilAndComs.NACOS_URL_BASE + "/api/clientBeat", params); JSONObject jsonObject = JSON.parseObject(result); if (jsonObject != null) { clientBeatInterval = jsonObject.getLong("clientBeatInterval"); } } catch (Exception e) { LogUtils.LOG.error("CLIENT-BEAT", "failed to send beat: " + JSON.toJSONString(beatInfo), e); } }}
這里就是naocs的客戶端主動上報服務健康狀況的邏輯了,是服務發(fā)現(xiàn)功能,比較重要的一個概念,服務健康檢查機制,常用的還有服務端主動去探測客戶端的接口返回。
最后一步,就是初始化了一個叫HostReactor的實例,我們來看下,它干了些啥:
public HostReactor(EventDispatcher eventDispatcher, NamingProxy serverProxy, String cacheDir) { this.eventDispatcher = eventDispatcher; this.serverProxy = serverProxy; this.cacheDir = cacheDir; this.serviceInfoMap = new ConcurrentHashMap<>(DiskCache.read(this.cacheDir)); this.failoverReactor = new FailoverReactor(this, cacheDir); this.pushRecver = new PushRecver(this);}
第五行,是從緩存文件中加載數(shù)據(jù)到serviceInfoMap的內(nèi)存map中,接下來,初始化了一個FailoverReactor的組件,這個是Nacos客戶端緩存容災相關(guān)的,它里面的初始化代碼如下:
public void init() { executorService.scheduleWithFixedDelay(new SwitchRefresher(), 0L, 5000L, TimeUnit.MILLISECONDS); executorService.scheduleWithFixedDelay(new DiskFileWriter(), 30, DAY_PERIOD_MINUTES, TimeUnit.MINUTES); // backup file on startup if failover directory is empty. executorService.schedule(new Runnable() { @Override public void run() { try { File cacheDir = new File(failoverDir); if (!cacheDir.exists() && !cacheDir.mkdirs()) { throw new IllegalStateException("failed to create cache dir: " + failoverDir); } File[] files = cacheDir.listFiles(); if (files == null || files.length <= 0) { new DiskFileWriter().run(); } } catch (Throwable e) { LogUtils.LOG.error("NA", "failed to backup file on startup.", e); } } }, 10000L, TimeUnit.MILLISECONDS);}
初始化了3個定時任務,第一個任務的代碼如下:
class SwitchRefresher implements Runnable { long lastModifiedMillis = 0L; @Override public void run() { try { File switchFile = new File(failoverDir + UtilAndComs.FAILOVER_SWITCH); if (!switchFile.exists()) { switchParams.put("failover-mode", "false"); LogUtils.LOG.debug("failover switch is not found, " + switchFile.getName()); return; } long modified = switchFile.lastModified(); if (lastModifiedMillis < modified) { lastModifiedMillis = modified; String failover = ConcurrentDiskUtil.getFileContent(failoverDir + UtilAndComs.FAILOVER_SWITCH, Charset.defaultCharset().toString()); if (!StringUtils.isEmpty(failover)) { List<String> lines = Arrays.asList(failover.split(DiskCache.getLineSeperator())); for (String line : lines) { String line1 = line.trim(); if ("1".equals(line1)) { switchParams.put("failover-mode", "true"); LogUtils.LOG.info("failover-mode is on"); new FailoverFileReader().run(); } else if ("0".equals(line1)) { switchParams.put("failover-mode", "false"); LogUtils.LOG.info("failover-mode is off"); } } } else { switchParams.put("failover-mode", "false"); } } } catch (Throwable e) { LogUtils.LOG.error("NA", "failed to read failover switch.", e); } }}
首先判定下容災開關(guān)是否有,容災開關(guān)是一個磁盤文件的形式存在,通過容災開關(guān)文件名字,判定容災開關(guān)是否打開,1表示打開,0為關(guān)閉,讀取到容災開關(guān)后,將值更新到內(nèi)存中,后續(xù)解析地址列表時,首先會判定一下容災開關(guān)是否打開,如果打開了,就讀緩存的數(shù)據(jù),否則從服務端獲取最新數(shù)據(jù)。
第二個定時任務,做的事情如下:
class DiskFileWriter extends TimerTask { public void run() { Map<String, ServiceInfo> map = hostReactor.getServiceInfoMap(); for (Map.Entry<String, ServiceInfo> entry : map.entrySet()) { ServiceInfo serviceInfo = entry.getValue(); if (StringUtils.equals(serviceInfo.getKey(), UtilAndComs.ALL_IPS) || StringUtils.equals(serviceInfo.getName(), UtilAndComs.ENV_LIST_KEY) || StringUtils.equals(serviceInfo.getName(), "00-00---000-ENV_CONFIGS-000---00-00") || StringUtils.equals(serviceInfo.getName(), "vipclient.properties") || StringUtils.equals(serviceInfo.getName(), "00-00---000-ALL_HOSTS-000---00-00")) { continue; } DiskCache.write(serviceInfo, failoverDir); } }}
每隔24小時,把內(nèi)存中所有的服務數(shù)據(jù),寫一遍到磁盤中,其中需要過濾掉一些非域名數(shù)據(jù)的特殊數(shù)據(jù),具體可看代碼中的描述。最后一個定時任務,是每隔10s,是檢查緩存目錄是否存在,同時如果緩存里面值沒有的話,主動觸發(fā)一次緩存寫磁盤的操作。
以上就是客戶端構(gòu)造一個Nacos實例的初始化全部流程,大部分都是在初始化多個線程池或者定時任務,各司其職,這個也是我們寫后端程序的一些基本套路,提高系統(tǒng)的并發(fā)能力,同時在對任務的分發(fā)和執(zhí)行,引入一些常用的異步編程模型如隊列模型的事件分發(fā),這些都是異步和并發(fā)的很好學習素材,這2點也是寫高性能程序的基本要求。
這一章節(jié),我們通過Nacos的NacosFactory構(gòu)造一個nacos服務實例作為切入點,把客戶端的初始化流程給串了一遍,概述下客戶端初始化流程做的幾件事:
初始化事件分發(fā)組件,用于處理服務端主動通知下來的變更數(shù)據(jù)
初始化Nacos服務集群地址列表更新組件,用于客戶端維護Nacos服務端的最新地址列表
初始化服務健康檢查模塊,主動給服務端上報服務的健康情況
初始化客戶端的緩存,10s檢查一次,如果沒有,則創(chuàng)建
24小時備份一次客戶端的緩存文件
5s檢查一次容災開關(guān),更新到內(nèi)存中,容災模式情況下,服務地址讀的都是緩存
以上就是Nacos客戶端實例初始化的整體流程,是不是感覺做的事情挺多的,還有一些代碼的細節(jié)點,大家自己多精讀一下,如果有什么不明白的,可以留言,或者在社區(qū)找@超哥幫你解答,如果能發(fā)現(xiàn)bug或者其他建議,可以在社區(qū)提issue。
免責聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進行舉報,并提供相關(guān)證據(jù),一經(jīng)查實,將立刻刪除涉嫌侵權(quán)內(nèi)容。