摘要:目前支持多種注冊中心。本編文章是分析使用作為注冊中心,如何整合進行服務注冊和訂閱服務。
目前dubbo支持多種注冊中心:Zookeeper、Redis、Simple、Multicast、Etcd3。
本編文章是分析使用Zookeeper作為注冊中心,dubbo如何整合Zookeeper進行服務注冊和訂閱服務。
首先dubbo將服務注冊到Zookeeper后,目錄結構如下所示:(注冊接口名:com.bob.dubbo.service.CityDubboService)
在consumer和provider服務啟動的時候,去把自身URL格式化成字符串,然后注冊到zookeeper相應節點下,作為臨時節點,斷開連接后,節點刪除;consumer啟動時,不僅會訂閱服務,同時也會將自己的URL注冊到zookeeper中;
ZookeeperRegistryZookeeperRegistry:dubbo與zookeeper交互主要的類,已下結合源碼進行分析,先來看
doSubcribe()
這個方法主要是用于訂閱服務,添加監聽器,動態監聽提供者列表變化:
</>復制代碼
@Override
public void doSubscribe(final URL url, final NotifyListener listener) {
try {
// 處理所有service層發起的訂閱,例如監控中心的訂閱
if (Constants.ANY_VALUE.equals(url.getServiceInterface())) {
String root = toRootPath();
ConcurrentMap listeners = zkListeners.get(url);
if (listeners == null) {
zkListeners.putIfAbsent(url, new ConcurrentHashMap<>());
listeners = zkListeners.get(url);
}
ChildListener zkListener = listeners.get(listener);
if (zkListener == null) {
listeners.putIfAbsent(listener, (parentPath, currentChilds) -> {
for (String child : currentChilds) {
child = URL.decode(child);
if (!anyServices.contains(child)) {
anyServices.add(child);
subscribe(url.setPath(child).addParameters(Constants.INTERFACE_KEY, child,
Constants.CHECK_KEY, String.valueOf(false)), listener);
}
}
});
zkListener = listeners.get(listener);
}
zkClient.create(root, false);
List services = zkClient.addChildListener(root, zkListener);
if (services != null && !services.isEmpty()) {
for (String service : services) {
service = URL.decode(service);
anyServices.add(service);
subscribe(url.setPath(service).addParameters(Constants.INTERFACE_KEY, service,
Constants.CHECK_KEY, String.valueOf(false)), listener);
}
}
// 處理指定service層發起的訂閱,例如服務消費者的訂閱
} else {
List urls = new ArrayList<>();
// 循環分類數組 , router, configurator, provider
for (String path : toCategoriesPath(url)) {
// 獲得 url 對應的監聽器集合
ConcurrentMap listeners = zkListeners.get(url);
if (listeners == null) {// 不存在,進行創建
zkListeners.putIfAbsent(url, new ConcurrentHashMap<>());
listeners = zkListeners.get(url);
}
// 獲得 ChildListener 對象
ChildListener zkListener = listeners.get(listener);
if (zkListener == null) {// 不存在子目錄的監聽器,進行創建 ChildListener 對象
// 訂閱父級目錄, 當有子節點發生變化時,觸發此回調函數,回調listener中的notify()方法
listeners.putIfAbsent(listener, (parentPath, currentChilds) -> ZookeeperRegistry.this.notify(url, listener, toUrlsWithEmpty(url, parentPath, currentChilds)));
zkListener = listeners.get(listener);
}
創建Type節點,此節點為持久節點
zkClient.create(path, false);
// 向 Zookeeper ,PATH 節點,發起訂閱,返回此節點下的所有子元素 path : /根節點/接口全名/providers, 比如 : /dubbo/com.bob.service.CityService/providers
List children = zkClient.addChildListener(path, zkListener);
if (children != null) {
urls.addAll(toUrlsWithEmpty(url, path, children));
}
}
// 首次全量數據獲取完成時,調用 `#notify(...)` 方法,回調 NotifyListener, 在這一步從連接Provider,實例化Invoker
notify(url, listener, urls);
}
} catch (Throwable e) {
throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
}
}
register()
ZookeeperRegistry父類FailbackRegistry中的方法,用于將服務注冊到zookeeper,具體代碼如下:
</>復制代碼
@Override
public void register(URL url) {
// 調用父類AbstractRegistry中的register()方法,將url存儲到注冊集合中
super.register(url);
// 如果之前這個url注冊失敗,則會從注冊失敗集合中刪除
removeFailedRegistered(url);
removeFailedUnregistered(url);
try {
// 像注冊中心發送注冊請求
doRegister(url);
} catch (Exception e) {
Throwable t = e;
// If the startup detection is opened, the Exception is thrown directly.
boolean check = getUrl().getParameter(Constants.CHECK_KEY, true)
&& url.getParameter(Constants.CHECK_KEY, true)
&& !Constants.CONSUMER_PROTOCOL.equals(url.getProtocol());
boolean skipFailback = t instanceof SkipFailbackWrapperException;
if (check || skipFailback) {
if (skipFailback) {
t = t.getCause();
}
throw new IllegalStateException("Failed to register " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t);
} else {
logger.error("Failed to register " + url + ", waiting for retry, cause: " + t.getMessage(), t);
}
// 將url存入注冊失敗集合中,進行重試try()
addFailedRegistered(url);
}
}
doRegister()
ZookeeperRegistry類中的方法
</>復制代碼
@Override
public void doRegister(URL url) {
try {
// 通過zookeeper客戶端向注冊中心發送服務注冊請求,在zookeeper下創建服務對應的節點
zkClient.create(toUrlPath(url), url.getParameter(Constants.DYNAMIC_KEY, true));
} catch (Throwable e) {
throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
}
}
在介紹注冊registry()方法的時候,解析到了FailbackRegistry類,接下來咱們來分析一下這個類的作用:
FailbackRegistry這個類是ZookeeperRegistry的父類,通過分析該類的結構,主要是用于服務的注冊、訂閱、重試,而服務具體的注冊、訂閱又在ZookeeperRegistry子類進行了實現,現在我們來分析重試這個功能,服務暴露和訂閱的配置文件中一般會設置重試這個屬性,如下所示:
</>復制代碼
上面是一個服務暴露的示例,設置了retries屬性,表示重試的次數。接下來咱們就以注冊重試進行分析(服務訂閱是同樣的原理):在注冊registry()方法中(代碼上面已提供),在異常catch{}代碼塊中有一個addFailedRegistered(url)方法,這個就是將注冊失敗的url添加到集合中,并創建一個重試的任務FailedRegisteredTask(url, this),代碼如下:
</>復制代碼
private void addFailedRegistered(URL url) {
// 先從集合中獲取,如果存在,直接返回
FailedRegisteredTask oldOne = failedRegistered.get(url);
if (oldOne != null) {
return;
}
// 本地集合不存在,則創建重試定時任務,默認每隔5s執行
FailedRegisteredTask newTask = new FailedRegisteredTask(url, this);
oldOne = failedRegistered.putIfAbsent(url, newTask);
if (oldOne == null) {
// 將定時任務放置在HashedWheelTimer這個處理定時任務的容器,(HashedWheelTimer執行原理,可以自行查找資料,這里就不介紹)
retryTimer.newTimeout(newTask, retryPeriod, TimeUnit.MILLISECONDS);
}
}
咱們下來看FailedRegisteredTask這個定時任務,有哪些東西,FailedRegisteredTask是AbstractRetryTask的子類,在執行new FailedRegisteredTask(url, this)代碼時,其實調用的是父類構造函數,其中retryTimes表示重試的次數,在沒有配置的情況下,默認重試三次:
</>復制代碼
AbstractRetryTask(URL url, FailbackRegistry registry, String taskName) {
if (url == null || StringUtils.isBlank(taskName)) {
throw new IllegalArgumentException();
}
this.url = url;
this.registry = registry;
this.taskName = taskName;
cancel = false;
this.retryPeriod = url.getParameter(Constants.REGISTRY_RETRY_PERIOD_KEY, Constants.DEFAULT_REGISTRY_RETRY_PERIOD);
// 重試次數,默認情況下重試三次
this.retryTimes = url.getParameter(Constants.REGISTRY_RETRY_TIMES_KEY, Constants.DEFAULT_REGISTRY_RETRY_TIMES);
}
在AbstractRetryTask類中有一個run()方法,在run()方法會根據XML配置文件中的retries屬性值進行比較來進行重試,如果沒有達到重試次數,則會調用doRetry(url, registry, timeout),而這個方法又在子類具體實現,這里我以注冊FailedRegisteredTask舉例:
</>復制代碼
@Override
public void run(Timeout timeout) throws Exception {
if (timeout.isCancelled() || timeout.timer().isStop() || isCancel()) {
// other thread cancel this timeout or stop the timer.
return;
}
// 重試次數與設置的retries進行比較,超過則不在進行重試
if (times > retryTimes) {
// reach the most times of retry.
logger.warn("Final failed to execute task " + taskName + ", url: " + url + ", retry " + retryTimes + " times.");
return;
}
if (logger.isInfoEnabled()) {
logger.info(taskName + " : " + url);
}
try {
// 調用子類實現,進行重試
doRetry(url, registry, timeout);
} catch (Throwable t) { // Ignore all the exceptions and wait for the next retry
logger.warn("Failed to execute task " + taskName + ", url: " + url + ", waiting for again, cause:" + t.getMessage(), t);
// reput this task when catch exception.
reput(timeout, retryPeriod);
}
}
在子類FailedRegisteredTask中doRetry()方法具體實現:
</>復制代碼
public final class FailedRegisteredTask extends AbstractRetryTask {
private static final String NAME = "retry register";
public FailedRegisteredTask(URL url, FailbackRegistry registry) {
super(url, registry, NAME);
}
@Override
protected void doRetry(URL url, FailbackRegistry registry, Timeout timeout) {
// 調用ZookeeperRegistry類中的doRegister()方法進行注冊
registry.doRegister(url);
registry.removeFailedRegisteredTask(url);
}
}
分析到這里,有個疑問:重試任務已經封裝了,任務什么時候去執行,怎么執行的?其實在上面咱們就分析到過,就是使用了HashedWheelTimer,這個類是在ZookeeperRegistry類初始化的時候就會去初始化:
</>復制代碼
public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter) {
// 這個地方進行初始化的:初始化父類FailbackRegistry
super(url);
if (url.isAnyHost()) {
throw new IllegalStateException("registry address == null");
}
String group = url.getParameter(Constants.GROUP_KEY, DEFAULT_ROOT);
if (!group.startsWith(Constants.PATH_SEPARATOR)) {
group = Constants.PATH_SEPARATOR + group;
}
this.root = group;
zkClient = zookeeperTransporter.connect(url);
zkClient.addStateListener(state -> {
if (state == StateListener.RECONNECTED) {
try {
recover();
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
});
}
</>復制代碼
public FailbackRegistry(URL url) {
super(url);
this.retryPeriod = url.getParameter(Constants.REGISTRY_RETRY_PERIOD_KEY, Constants.DEFAULT_REGISTRY_RETRY_PERIOD);
// 創建HashedWheelTimer對象
retryTimer = new HashedWheelTimer(new NamedThreadFactory("DubboRegistryRetryTimer", true), retryPeriod, TimeUnit.MILLISECONDS, 128);
}
然后在addFailedRegistered()方法中有retryTimer.newTimeout(newTask, retryPeriod, TimeUnit.MILLISECONDS);這樣的一條代碼,這個就是執行任務的開始點:
</>復制代碼
@Override
public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
if (task == null) {
throw new NullPointerException("task");
}
if (unit == null) {
throw new NullPointerException("unit");
}
long pendingTimeoutsCount = pendingTimeouts.incrementAndGet();
if (maxPendingTimeouts > 0 && pendingTimeoutsCount > maxPendingTimeouts) {
pendingTimeouts.decrementAndGet();
throw new RejectedExecutionException("Number of pending timeouts ("
+ pendingTimeoutsCount + ") is greater than or equal to maximum allowed pending "
+ "timeouts (" + maxPendingTimeouts + ")");
}
// 開啟輪詢任務
start();
// Add the timeout to the timeout queue which will be processed on the next tick.
// During processing all the queued HashedWheelTimeouts will be added to the correct HashedWheelBucket.
long deadline = System.nanoTime() + unit.toNanos(delay) - startTime;
// Guard against overflow.
if (delay > 0 && deadline < 0) {
deadline = Long.MAX_VALUE;
}
HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);
timeouts.add(timeout);
return timeout;
}
調用start()方法時,開啟一個線程work去輪詢存儲到HashedWheelTimer容器的任務,然后調用任務中的run()方法,
</>復制代碼
public void start() {
switch (WORKER_STATE_UPDATER.get(this)) {
case WORKER_STATE_INIT:
if (WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_INIT, WORKER_STATE_STARTED)) {
// 開啟work線程,執行work線程中的run()方法
workerThread.start();
}
break;
case WORKER_STATE_STARTED:
break;
case WORKER_STATE_SHUTDOWN:
throw new IllegalStateException("cannot be started once stopped");
default:
throw new Error("Invalid WorkerState");
}
// Wait until the startTime is initialized by the worker.
while (startTime == 0) {
try {
startTimeInitialized.await();
} catch (InterruptedException ignore) {
// Ignore - it will be ready very soon.
}
}
}
</>復制代碼
@Override
public void run() {
// Initialize the startTime.
startTime = System.nanoTime();
if (startTime == 0) {
// We use 0 as an indicator for the uninitialized value here, so make sure it"s not 0 when initialized.
startTime = 1;
}
// Notify the other threads waiting for the initialization at start().
startTimeInitialized.countDown();
do {
final long deadline = waitForNextTick();
if (deadline > 0) {
int idx = (int) (tick & mask);
processCancelledTasks();
HashedWheelBucket bucket =
wheel[idx];
transferTimeoutsToBuckets();
// 執行重試任務
bucket.expireTimeouts(deadline);
tick++;
}
} while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_STARTED);
// Fill the unprocessedTimeouts so we can return them from stop() method.
for (HashedWheelBucket bucket : wheel) {
bucket.clearTimeouts(unprocessedTimeouts);
}
for (; ; ) {
HashedWheelTimeout timeout = timeouts.poll();
if (timeout == null) {
break;
}
if (!timeout.isCancelled()) {
unprocessedTimeouts.add(timeout);
}
}
processCancelledTasks();
}
</>復制代碼
void expireTimeouts(long deadline) {
HashedWheelTimeout timeout = head;
// process all timeouts
while (timeout != null) {
// 輪詢獲取重試任務
HashedWheelTimeout next = timeout.next;
if (timeout.remainingRounds <= 0) {
next = remove(timeout);
if (timeout.deadline <= deadline) {
// 執行重試任務
timeout.expire();
} else {
// The timeout was placed into a wrong slot. This should never happen.
throw new IllegalStateException(String.format(
"timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline));
}
} else if (timeout.isCancelled()) {
next = remove(timeout);
} else {
timeout.remainingRounds--;
}
timeout = next;
}
}
</>復制代碼
public void expire() {
if (!compareAndSetState(ST_INIT, ST_EXPIRED)) {
return;
}
try {
// 調用任務中的run()方法,(如:AbstractRetryTask任務中的run()方法,在去調用子類FailedRegisteredTask中的doRetry()方法進行重試注冊)
task.run(this);
} catch (Throwable t) {
if (logger.isWarnEnabled()) {
logger.warn("An exception was thrown by " + TimerTask.class.getSimpleName() + ".", t);
}
}
}
在上面對于HashedWheelTimer的具體實現原理,并沒有進行詳細的進行分析,如果想了解的和學習的話,可以自行查找資料。
文章版權歸作者所有,未經允許請勿轉載,若此文章存在違規行為,您可以聯系管理員刪除。
轉載請注明本文地址:http://m.specialneedsforspecialkids.com/yun/73033.html
摘要:是目前非常流行的分布式服務技術,很多公司都在用。空閑之余,搭了個,分享給大家。本文下載下載是服務的注冊中心,下載后進入到安裝目錄。雙擊即可啟動注冊中心服務。 dubbo是目前非常流行的分布式服務技術,很多公司都在用。空閑之余,搭了個helloworld,分享給大家。本文demo下載1.下載 zookeeperzookeeper是服務的注冊中心,下載后進入到安裝目錄G:bakCenter...
摘要:啟動容器,加載,運行服務提供者。服務提供者在啟動時,在注冊中心發布注冊自己提供的服務。注冊中心返回服務提供者地址列表給消費者,如果有變更,注冊中心將基于長連接推送變更數據給消費者。 一 為什么需要 dubbo 很多時候,其實我們使用這個技術的時候,可能都是因為項目需要,所以,我們就用了,但是,至于為什么我們需要用到這個技術,可能自身并不是很了解的,但是,其實了解技術的來由及背景知識,對...
Github 地址:https://github.com/Snailclimb/springboot-integration-examples ,歡迎各位 Star。 目錄: 使用 SpringBoot+Dubbo 搭建一個簡單分布式服務 實戰之前,先來看幾個重要的概念 什么是分布式? 什么是 Duboo? Dubbo 架構 什么是 RPC? 為什么要用 Dubbo? 開始實戰 1 ...
摘要:構建服務接口創建一個簡單的項目,并在下面定義一個抽象接口,比如構建服務接口提供方第一步創建一個項目,在中引入第一步中構建的包以及對和的依賴,比如第一步中構建的包這里需要注意兩點必須包含包,不然啟動會報錯。 很早以前,在剛開始搞Spring Cloud基礎教程的時候,寫過這樣一篇文章:《微服務架構的基礎框架選擇:Spring Cloud還是Dubbo?》,可能不少讀者也都看過。之后也就一...
摘要:在微服務架構中,注冊中心是核心的基礎服務之一。在微服務架構流行之前,注冊中心就已經開始出現在分布式架構的系統中。服務提供者注冊到注冊中心,服務消費者到注冊中心訂閱,同時,注冊中心中的變更也會通知服務消費者。 在微服務架構中,注冊中心是核心的基礎服務之一。在微服務架構流行之前,注冊中心就已經開始出現在分布式架構的系統中。Dubbo是一個在國內比較流行的分布式框架,被大量的中小型互聯網公司...
閱讀 2667·2019-08-30 15:53
閱讀 2881·2019-08-29 16:20
閱讀 1088·2019-08-29 15:10
閱讀 1029·2019-08-26 10:58
閱讀 2199·2019-08-26 10:49
閱讀 640·2019-08-26 10:21
閱讀 708·2019-08-23 18:30
閱讀 1641·2019-08-23 15:58