在使用dubbo的过程中，当注册中心的数据修改后，新的配置是怎样刷到consumer和provider的？本文以consumer为例，进行分析。
dubbo使用的是zkclient的jar，而zkclient依赖zookeeper的jar。
示例代码：
public class MyI0zk {
public static void main(String[] args) {
ZkClient zkclient = new ZkClient("127.0.0.1:2181");
int countChildren = zkclient.countChildren("/");
System.out.println(countChildren);
}
}
调用栈如下图：
在connect方法中创建了ZooKeeper对象：
// void org.I0Itec.zkclient.ZkConnection.connect(Watcher watcher)
public void connect(Watcher watcher) {
_zookeeperLock.lock();
try {
if (_zk != null) {
throw new IllegalStateException("zk client has already been started");
}
try {
LOG.debug("Creating new ZookKeeper instance to connect to " + _servers + ".");
_zk = new ZooKeeper(_servers, _sessionTimeOut, watcher);
} catch (IOException e) {
throw new ZkException("Unable to connect to " + _servers, e);
}
} finally {
_zookeeperLock.unlock();
}
}
ZkClient自身就是一个Watcher：
//省略其他代码
public class ZkClient implements Watcher {
public void process(WatchedEvent event) {
LOG.debug("Received event: " + event);
_zookeeperEventThread = Thread.currentThread(); boolean stateChanged = event.getPath() == null;
boolean znodeChanged = event.getPath() != null;
boolean dataChanged = event.getType() == EventType.NodeDataChanged || event.getType() == EventType.NodeDeleted || event.getType() == EventType.NodeCreated
|| event.getType() == EventType.NodeChildrenChanged; getEventLock().lock();
try { // We might have to install child change event listener if a new node was created
if (getShutdownTrigger()) {
LOG.debug("ignoring event '{" + event.getType() + " | " + event.getPath() + "}' since shutdown triggered");
return;
}
if (stateChanged) {
processStateChanged(event);
}
if (dataChanged) {
processDataOrChildChange(event);
}
} finally {
if (stateChanged) {
getEventLock().getStateChangedCondition().signalAll(); // If the session expired we have to signal all conditions, because watches might have been removed and
// there is no guarantee that those
// conditions will be signaled at all after an Expired event
// TODO PVo write a test for this
if (event.getState() == KeeperState.Expired) {
getEventLock().getZNodeEventCondition().signalAll();
getEventLock().getDataChangedCondition().signalAll();
// We also have to notify all listeners that something might have changed
fireAllEvents();
}
}
if (znodeChanged) {
getEventLock().getZNodeEventCondition().signalAll();
}
if (dataChanged) {
getEventLock().getDataChangedCondition().signalAll();
}
getEventLock().unlock();
LOG.debug("Leaving process event");
}
} }
1. 再看看dubbo的consumer创建ZkClient的调用栈：
2. consumer的ZkClient注册childListener，等待节点内容改变的通知，对[/dubbo/com.zhang.HelloService/providers, /dubbo/com.zhang.HelloService/configurators, /dubbo/com.zhang.HelloService/routers]分别进行注册：
实际调用exists和getChildren方法，对应zk的stat path watch和ls path watch命令，即当/dubbo/com.zhang.HelloService/providers的值或者子节点改变时，客户端会收到通知：
// ZkClient
public List<String> watchForChilds(final String path) {
if (_zookeeperEventThread != null && Thread.currentThread() == _zookeeperEventThread) {
throw new IllegalArgumentException("Must not be done in the zookeeper event thread.");
}
return retryUntilConnected(new Callable<List<String>>() {
@Override
public List<String> call() throws Exception {
exists(path, true);
try {
return getChildren(path, true);
} catch (ZkNoNodeException e) {
// ignore, the "exists" watch will listen for the parent node to appear
}
return null;
}
});
} public <T> T retryUntilConnected(Callable<T> callable) throws ZkInterruptedException, IllegalArgumentException, ZkException, RuntimeException {
if (_zookeeperEventThread != null && Thread.currentThread() == _zookeeperEventThread) {
throw new IllegalArgumentException("Must not be done in the zookeeper event thread.");
}
while (true) {
try {
return callable.call();
} catch (ConnectionLossException e) {
// we give the event thread some time to update the status to 'Disconnected'
Thread.yield();
waitUntilConnected();
} catch (SessionExpiredException e) {
// we give the event thread some time to update the status to 'Expired'
Thread.yield();
waitUntilConnected();
} catch (KeeperException e) {
throw ZkException.create(e);
} catch (InterruptedException e) {
throw new ZkInterruptedException(e);
} catch (Exception e) {
throw ExceptionUtil.convertToRuntimeException(e);
}
}
}
3. consumer收到通知，重新加watch
consumer的ZkClient收到通知后，会将通知包装成ZkEvent事件，发送给ZkEventThread，ZkEvent的run方法里重新地加上watch：
private void fireChildChangedEvents(final String path, Set<IZkChildListener> childListeners) {
try {
// reinstall the watch
for (final IZkChildListener listener : childListeners) {
_eventThread.send(new ZkEvent("Children of " + path + " changed sent to " + listener) {
@Override
public void run() throws Exception {
try {
// if the node doesn't exist we should listen for the root node to reappear
// 在函数内部，根据是否有listeners，判断加watch
exists(path);
List<String> children = getChildren(path);
listener.handleChildChange(path, children);
} catch (ZkNoNodeException e) {
listener.handleChildChange(path, null);
}
}
});
}
} catch (Exception e) {
LOG.error("Failed to fire child changed event. Unable to getChildren. ", e);
}
}
4. 从ZkEvent的run方法进入，刷新配置：
ZkEventThread线程：
class ZkEventThread extends Thread {
private static final Logger LOG = Logger.getLogger(ZkEventThread.class);
private BlockingQueue<ZkEvent> _events = new LinkedBlockingQueue<ZkEvent>();
private static AtomicInteger _eventId = new AtomicInteger(0);
static abstract class ZkEvent {
private String _description;
public ZkEvent(String description) {
_description = description;
}
public abstract void run() throws Exception;
@Override
public String toString() {
return "ZkEvent[" + _description + "]";
}
}
ZkEventThread(String name) {
setDaemon(true);
setName("ZkClient-EventThread-" + getId() + "-" + name);
}
@Override
public void run() {
LOG.info("Starting ZkClient event thread.");
try {
while (!isInterrupted()) {
ZkEvent zkEvent = _events.take();
int eventId = _eventId.incrementAndGet();
LOG.debug("Delivering event #" + eventId + " " + zkEvent);
try {
zkEvent.run();
} catch (InterruptedException e) {
interrupt();
} catch (ZkInterruptedException e) {
interrupt();
} catch (Throwable e) {
LOG.error("Error handling event " + zkEvent, e);
}
LOG.debug("Delivering event #" + eventId + " done");
}
} catch (InterruptedException e) {
LOG.info("Terminate ZkClient event thread.");
}
}
public void send(ZkEvent event) {
if (!isInterrupted()) {
LOG.debug("New event: " + event);
_events.add(event);
}
}
}