DistroConsistencyServiceImpl

?

我們來看臨時(shí)實(shí)例的一致性實(shí)現(xiàn)：DistroConsistencyServiceImpl類的put方法：

public void put(String key, Record value) throws NacosException {// 先將要更新的實(shí)例信息寫入本地實(shí)例列表onPut(key, value);// 開始集群同步distroProtocol.sync(new DistroKey(key, KeyBuilder.INSTANCE_LIST_KEY_PREFIX), DataOperation.CHANGE,globalConfig.getTaskDispatchPeriod() / 2); }

這里方法只有兩行：

• onPut(key, value)：其中value就是Instances，要更新的服務(wù)信息。這行主要是基于線程池方式，異步的將Service信息寫入注冊(cè)表中(就是那個(gè)多重Map)

• distroProtocol.sync()：就是通過Distro協(xié)議將數(shù)據(jù)同步給集群中的其它Nacos節(jié)點(diǎn)

我們先看onPut方法

更新本地實(shí)例列表

1）放入阻塞隊(duì)列

onPut方法如下：

public void onPut(String key, Record value) {// 判斷是否是臨時(shí)實(shí)例if (KeyBuilder.matchEphemeralInstanceListKey(key)) {// 封裝 Instances 信息到 數(shù)據(jù)集：DatumDatum<Instances> datum = new Datum<>();datum.value = (Instances) value;datum.key = key;datum.timestamp.incrementAndGet();// 放入DataStoredataStore.put(key, datum);}if (!listeners.containsKey(key)) {return;}// 放入阻塞隊(duì)列，這里的 notifier維護(hù)了一個(gè)阻塞隊(duì)列，并且基于線程池異步執(zhí)行隊(duì)列中的任務(wù)notifier.addTask(key, DataOperation.CHANGE); }

notifier的類型就是DistroConsistencyServiceImpl.Notifier，內(nèi)部維護(hù)了一個(gè)阻塞隊(duì)列，存放服務(wù)列表變更的事件：

addTask時(shí)，將任務(wù)加入該阻塞隊(duì)列：

// DistroConsistencyServiceImpl.Notifier類的 addTask 方法： public void addTask(String datumKey, DataOperation action) {if (services.containsKey(datumKey) && action == DataOperation.CHANGE) {return;}if (action == DataOperation.CHANGE) {services.put(datumKey, StringUtils.EMPTY);}// 任務(wù)放入阻塞隊(duì)列tasks.offer(Pair.with(datumKey, action)); }

超強(qiáng)干貨來襲 云風(fēng)專訪：近40年碼齡，通宵達(dá)旦的技術(shù)人生

總結(jié)

以上是生活随笔為你收集整理的Nacos源码DistroConsistencyServiceImpl的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。