這篇文章主要介紹“Java中怎么用nacos配置中心”，在日常操作中，相信很多人在Java中怎么用nacos配置中心問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對(duì)大家解答”Java中怎么用nacos配置中心”的疑惑有所幫助！接下來，請跟著小編一起來學(xué)習(xí)吧！

讓客戶滿意是我們工作的目標(biāo)，不斷超越客戶的期望值來自于我們對(duì)這個(gè)行業(yè)的熱愛。我們立志把好的技術(shù)通過有效、簡單的方式提供給客戶，將通過不懈努力成為客戶在信息化領(lǐng)域值得信任、有價(jià)值的長期合作伙伴，公司提供的服務(wù)項(xiàng)目有：域名注冊、雅安服務(wù)器托管、營銷軟件、網(wǎng)站建設(shè)、臨沭網(wǎng)站維護(hù)、網(wǎng)站推廣。

配置的發(fā)布與訂閱

我們先來看看如何使用nacos提供的api來實(shí)現(xiàn)配置的發(fā)布與訂閱

發(fā)布配置:

public class ConfigPub {

    public static void main(String[] args) throws NacosException {

        final String dataId="test";

        final String group="DEFAULT_GROUP";

        ConfigService configService= NacosFactory.createConfigService("localhost:8848");

        configService.publishConfig(dataId,group,"test config body");
    }
}

訂閱配置:

   public static void main(String[] args) throws NacosException, InterruptedException {

        final String dataId="test";

        final String group="DEFAULT_GROUP";

        ConfigService configService= NacosFactory.createConfigService("localhost:8848");

        configService.addListener(dataId, group, new Listener() {
            @Override
            public Executor getExecutor() {
                return null;
            }

            @Override
            public void receiveConfigInfo(String configInfo) {

                System.out.println("receiveConfigInfo:"+configInfo);
            }
        });

        Thread.sleep(Integer.MAX_VALUE);
    }
}

根據(jù)上面的demo可以看到通過dataId和group可以定位一個(gè)配置文件。

深入了解配置發(fā)布

1-發(fā)布的配置信息會(huì)通過http請求調(diào)用具體的服務(wù)

agent.httpPost(url, headers, params, encode, POST_TIMEOUT);

服務(wù)類為 ConfigController：處理配置相關(guān)的http請求

persistService
      .insertOrUpdateTag(configInfo, tag, srcIp, srcUser, time, false);
EventDispatcher.fireEvent(
      new ConfigDataChangeEvent(false, dataId, group, tenant, tag,
            time.getTime()));

可以看到發(fā)布的配置首先會(huì)進(jìn)行持久化,然后會(huì)觸發(fā)變更通知。

持久化這里就不做分析,我們來看看fireEvent這個(gè)方法:

EventDispatcher.fireEvent:
static public void fireEvent(Event event) {
    if (null == event) {
        throw new IllegalArgumentException("event is null");
    }

    for (AbstractEventListener listener : getEntry(event.getClass()).listeners) {
        try {
            listener.onEvent(event);
        } catch (Exception e) {
            log.error(e.toString(), e);
        }
    }
}

這里可以看到具體調(diào)用了listener.onEvent(event);
這里只要找到AbstractEventListener 具體的實(shí)現(xiàn)類是哪個(gè)就可以。
AbstractEventListener主要有兩個(gè)實(shí)現(xiàn)類:
AsyncNotifyService
LongPollingService

我們可以通過event的類型去判斷,因?yàn)檫@里onEvent的參數(shù)類型為ConfigDataChangeEvent,
所以我們可以清楚的知道我們要找的實(shí)現(xiàn)類是AsyncNotifyService。
每個(gè)AbstractEventListener初始化的時(shí)候都會(huì)先將自己加入到listeners中
final CopyOnWriteArrayList listeners;
public AbstractEventListener(
) {
    /**
     * automatic register
     */
    EventDispatcher.addEventListener(this);
}

我們可以直接看看AsyncNotifyService的onEvent方法:
public void onEvent(Event event) {

   // 并發(fā)產(chǎn)生 ConfigDataChangeEvent
   if (event instanceof ConfigDataChangeEvent) {
      ConfigDataChangeEvent evt = (ConfigDataChangeEvent) event;
      long dumpTs = evt.lastModifiedTs;
      String dataId = evt.dataId;
      String group = evt.group;
      String tenant = evt.tenant;
      String tag = evt.tag;
      //Member{address='192.168.31.192:8848'}
      Collection ipList = memberManager.allMembers();

      // 其實(shí)這里任何類型隊(duì)列都可以
      Queue queue = new LinkedList();
      for (Member member : ipList) {
         queue.add(new NotifySingleTask(dataId, group, tenant, tag, dumpTs,
               member.getAddress(), evt.isBeta));
      }
      EXECUTOR.execute(new AsyncTask(httpclient, queue));
   }
}

上面的方法主要實(shí)現(xiàn)的是:
獲取所有的nacos服務(wù)節(jié)點(diǎn),然后對(duì)其執(zhí)行異步任務(wù)AsyncTask。
AsyncTask中會(huì)從隊(duì)列中獲取每個(gè)節(jié)點(diǎn)的NotifySingleTask信息，然后進(jìn)行http請求,調(diào)用通知配置信息改變
的服務(wù)。具體服務(wù)在CommunicationController中實(shí)現(xiàn)。

/**
 * 通知配置信息改變
 */
@GetMapping("/dataChange")

這個(gè)方法放在后面分析。

深入了解配置訂閱

初始化：

NacosConfigService初始化的時(shí)候構(gòu)造了ClientWorker,并且通過ClientWorker啟動(dòng)了兩個(gè)線程池。
worker = new ClientWorker(agent, configFilterChainManager, properties);
第一個(gè)線程池每10ms執(zhí)行一次checkConfigInfo();
executor.scheduleWithFixedDelay(new Runnable() {
    @Override
    public void run() {
        try {
            checkConfigInfo();
        } catch (Throwable e) {
           LOGGER.error("[" + agent.getName() + "] [sub-check] rotate check 
           error", e);
        }
    }
}, 1L, 10L, TimeUnit.MILLISECONDS);

我們來看看checkConfigInfo具體是做什么的
public void checkConfigInfo() {
    // 分任務(wù)
    int listenerSize = cacheMap.get().size();
    // 向上取整為批數(shù),限制LongPollingRunnable處理配置的個(gè)數(shù)。
    int longingTaskCount =(int) Math.ceil(listenerSize / ParamUtil.getPerTaskConfigSize());
    if (longingTaskCount > currentLongingTaskCount) {
        for (int i = (int) currentLongingTaskCount; i < longingTaskCount; i++) {
            // 要判斷任務(wù)是否在執(zhí)行 這塊需要好好想想。 
            //任務(wù)列表現(xiàn)在是無序的。變化過程可能有問題
            executorService.execute(new LongPollingRunnable(i));
            //這里的i就代表taskId
        }
        currentLongingTaskCount = longingTaskCount;
    }
}

這里主要的作用是提交LongPollingRunnable任務(wù)到第二個(gè)線程池中去運(yùn)行。
并且每個(gè)LongPollingRunnable只會(huì)處理3000個(gè)配置。

我們來看看LongPollingRunnable的實(shí)現(xiàn)
List cacheDatas = new ArrayList();
List inInitializingCacheList = new ArrayList();
try {
    // check failover config
    for (CacheData cacheData : cacheMap.get().values()) {
        if (cacheData.getTaskId() == taskId) {
            cacheDatas.add(cacheData);
            ...
        }
    }
cacheMap中保存了配置信息,從磁盤中加載獲取。
通過taskId從 cacheMap中獲取需要被當(dāng)前LongPollingRunnable任務(wù)處理的配置,放入到cacheDatas集合。

我們來看看是在哪里設(shè)置的taskId
int taskId = cacheMap.get().size() / (int) ParamUtil.getPerTaskConfigSize();
cache.setTaskId(taskId);
可以看到這里和上面相對(duì)應(yīng),每3000個(gè)配置的taskId是相同的。因?yàn)槊總€(gè)LongPollingRunnable線程會(huì)處理
3000個(gè)配置。
// check server config  向服務(wù)端請求變化的配置
List changedGroupKeys = checkUpdateDataIds(cacheDatas, inInitializingCacheList);

//從Server獲取值變化了的DataID列表。返回的對(duì)象里只有dataId和group是有效的。 保證不返回NULL。
return checkUpdateConfigStr(sb.toString(), isInitializingCacheList);

這里訂閱配置的客戶端會(huì)向服務(wù)端發(fā)送http長輪詢請求,來獲取變化的配置信息
長輪詢請求不會(huì)立刻返回結(jié)果,而是當(dāng)有配置發(fā)生變化時(shí)返回,設(shè)置了超時(shí)時(shí)間30s,如果超過了設(shè)置的
超時(shí)時(shí)間沒有配置更新,則會(huì)默認(rèn)返回。然后重新發(fā)起一次長輪詢的請求。

HttpResult result = agent.httpPost(Constants.CONFIG_CONTROLLER_PATH + "/listener", 
headers, params,
    agent.getEncode(), readTimeoutMs);

長輪詢的周期默認(rèn)為30s:
timeout=Math.max(NumberUtils.toInt(properties.getProperty(PropertyKeyConst.CONFIG_LONG_POLL_TIMEOUT),
    Constants.CONFIG_LONG_POLL_TIMEOUT), Constants.MIN_CONFIG_LONG_POLL_TIMEOUT);

具體服務(wù)實(shí)現(xiàn)類在ConfigController中:
@PostMapping("/listener")
@Secured(action = ActionTypes.READ, parser = ConfigResourceParser.class)
public void listener(HttpServletRequest request, HttpServletResponse response)
      throws ServletException, IOException {
   ....

   // do long-polling
   inner.doPollingConfig(request, response, clientMd5Map, probeModify.length());
}

doPollingConfig方法:
// 服務(wù)端處理長輪詢請求
if (LongPollingService.isSupportLongPolling(request)) {
    longPollingService.addLongPollingClient(request, response, clientMd5Map, 
    probeRequestSize);
    return HttpServletResponse.SC_OK + "";
}

使用線程池處理請求:
scheduler.execute(
    new ClientLongPolling(asyncContext, clientMd5Map, ip, probeRequestSize, timeout, 
    appName, tag));

接著來看ClientLongPolling是一個(gè)線程實(shí)現(xiàn)類
首先會(huì)觸發(fā)一個(gè)延時(shí)任務(wù),然后將自己加入到隊(duì)列:allSubs.add(this);
allSubs中維護(hù)了所有長輪訓(xùn)請求。

那么肯定會(huì)有一個(gè)地方去消費(fèi)allSubs隊(duì)列中的請求.
這個(gè)消費(fèi)的地方就是onEvent方法:
LongPollingService其實(shí)就是我們上面提到的AbstractEventListener,因此也實(shí)現(xiàn)了onEvent方法。

@Override
public void onEvent(Event event) {
    if (isFixedPolling()) {
        // ignore
    } else {
        if (event instanceof LocalDataChangeEvent) {
            LocalDataChangeEvent evt = (LocalDataChangeEvent)event;
            scheduler.execute(new DataChangeTask(evt.groupKey, evt.isBeta, 
            evt.betaIps));
        }
    }
}

這個(gè)event方法就是去處理配置變化的情況，主要邏輯在DataChangeTask中:
從allSubs獲取維護(hù)的請求中相同dataId+group的請求,比如:（test+DEFAULT_GROUP）
然后進(jìn)行這個(gè)對(duì)長輪詢的請求進(jìn)行返回。
for (Iterator iter = allSubs.iterator(); iter.hasNext(); ) {
    ClientLongPolling clientSub = iter.next();
    //groupKey test+DEFAULT_GROUP
    if (clientSub.clientMd5Map.containsKey(groupKey)) {
        ......
        iter.remove(); // 刪除訂閱關(guān)系
        LogUtil.clientLog.info("{}|{}|{}|{}|{}|{}|{}",
        (System.currentTimeMillis() - changeTime),
        "in-advance",
        RequestUtil.getRemoteIp((HttpServletRequest)clientSub.asyncContext.getRequest()),
            "polling",
            clientSub.clientMd5Map.size(), clientSub.probeRequestSize, groupKey);
        clientSub.sendResponse(Arrays.asList(groupKey));
    }
}
那是哪里觸發(fā)了LongPollingService里面的onEvent 方法呢?
當(dāng)然是在配置發(fā)布后進(jìn)行觸發(fā)的,還記得CommunicationController中的dataChange服務(wù)嗎？
配置發(fā)布后會(huì)通過http請求調(diào)用nacos服務(wù)中的dataChange服務(wù)。通過dataChange服務(wù)就可以通知
nacos服務(wù)中保存的長輪訓(xùn)的請求了。

并且這個(gè)方法是獲取所有nacos服務(wù)節(jié)點(diǎn)去遍歷執(zhí)行的,因此不管變更配置對(duì)應(yīng)的長輪詢保存在哪個(gè)節(jié)點(diǎn),
都會(huì)可以被獲取到。

/**
 * 通知配置信息改變
 */
@GetMapping("/dataChange")

此處會(huì)調(diào)用DumpService中的方法保存配置文件到磁盤，并緩存md5.

DiskUtil.saveToDisk(dataId, group, tenant, content);

public static void updateMd5(String groupKey, String md5, long lastModifiedTs) {
    CacheItem cache = makeSure(groupKey);
    if (cache.md5 == null || !cache.md5.equals(md5)) {
        cache.md5 = md5;
        cache.lastModifiedTs = lastModifiedTs;
        EventDispatcher.fireEvent(new LocalDataChangeEvent(groupKey));
    }
}

可以看到當(dāng)配置變更,就會(huì)觸發(fā)fireEvent的LocalDataChangeEvent事件。

到此，關(guān)于“Java中怎么用nacos配置中心”的學(xué)習(xí)就結(jié)束了，希望能夠解決大家的疑惑。理論與實(shí)踐的搭配能更好的幫助大家學(xué)習(xí)，快去試試吧！若想繼續(xù)學(xué)習(xí)更多相關(guān)知識(shí)，請繼續(xù)關(guān)注創(chuàng)新互聯(lián)網(wǎng)站，小編會(huì)繼續(xù)努力為大家?guī)砀鄬?shí)用的文章！