dubbo 注册公网_dubbo之Zookeeper注册中心

╰半橙微兮° 2023-01-08 14:29 258阅读 0赞

目前dubbo支持多种注册中心：Zookeeper、Redis、Simple、Multicast、Etcd3。

本编文章是分析使用Zookeeper作为注册中心，dubbo如何整合Zookeeper进行服务注册和订阅服务。

首先dubbo将服务注册到Zookeeper后，目录结构如下所示：(注册接口名：com.bob.dubbo.service.CityDubboService)

![Image 1][]

在consumer和provider服务启动的时候，去把自身URL格式化成字符串，然后注册到zookeeper相应节点下，作为临时节点，断开连接后，节点删除；consumer启动时，不仅会订阅服务，同时也会将自己的URL注册到zookeeper中；

ZookeeperRegistry

ZookeeperRegistry：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);

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的具体实现原理，并没有进行详细的进行分析，如果想了解的和学习的话，可以自行查找资料。

[Image 1]:

dubbo 注册公网_dubbo之Zookeeper注册中心

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关 spring dubbo使用 Zookeeper 注册中心

相关 dubbo 注册公网_dubbo之Zookeeper注册中心

相关 Dubbo注册中心

相关 Dubbo-----HelloWorld之Zookeeper注册中心（2）

相关 dubbo的Zookeeper注册中心模式

相关 Dubbo介绍、Zookeeper注册中心实例

相关 dubbo的Zookeeper注册中心模式

相关 dubbo之zookeeper注册中心

相关 Dubbo学习一，zookeeper注册中心

相关 dubbo注册中心