Zookeeper-watcher机制源码分析(一)Watcher的基本流程

Watcher的基本流程

ZooKeeper 的 Watcher 机制,总的来说可以分为三个过程:客户端注册 Watcher、服务器处理 Watcher 和客户端回调 Watcher

客户端注册watcher有3种方式,getData、exists、getChildren;以如下代码为例来分析整个触发机制的原理

|

ZooKeeper zookeeper=new ZooKeeper(“192.168.11.152:2181”,4000,new Watcher(){

public void processor(WatchedEvent event){

System.out.println(“event.type”);

}

});

zookeeper.create(“/mic”,”0”.getByte(),ZooDefs.Ids. OPEN_ACL_UNSAFE,CreateModel. PERSISTENT); //创建节点

zookeeper.exists(“/mic”,true); //注册监听

zookeeper.setData(“/mic”, “1”.getByte(),-1) ; //修改节点的值触发监听

|

ZooKeeper API的初始化过程

|

ZooKeeper zookeeper=new ZooKeeper(“192.168.11.152:2181”,4000,new Watcher(){

public void processor(WatchedEvent event){

System.out.println(“event.type”);

}

});

|

在创建一个 ZooKeeper 客户端对象实例时,我们通过new Watcher()向构造方法中传入一个默认的 Watcher, 这个 Watcher 将作为整个 ZooKeeper会话期间的默认 Watcher,会一直被保存在客户端 ZKWatchManager 的 defaultWatcher 中;代码如下

|

public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,

boolean canBeReadOnly, HostProvider aHostProvider,

ZKClientConfig clientConfig) throws IOException {

LOG.info("Initiating client connection, connectString=" + connectString

  • " sessionTimeout=" + sessionTimeout + " watcher=" + watcher);

if (clientConfig == null) {

clientConfig = new ZKClientConfig();

}

this.clientConfig = clientConfig;

watchManager = defaultWatchManager();

watchManager.defaultWatcher = watcher; --在这里将watcher设置到ZKWatchManager

ConnectStringParser connectStringParser = new ConnectStringParser(

connectString);

hostProvider = aHostProvider;

--初始化了ClientCnxn,并且调用cnxn.start()方法

cnxn = new ClientCnxn(connectStringParser.getChrootPath(),

hostProvider, sessionTimeout, this, watchManager,

getClientCnxnSocket(), canBeReadOnly);

cnxn.start();

}

|

ClientCnxn:是Zookeeper客户端和Zookeeper服务器端进行通信和事件通知处理的主要类,它内部包含两个类,

1. SendThread :负责客户端和服务器端的数据通信, 也包括事件信息的传输

2. EventThread : 主要在客户端回调注册的Watchers进行通知处理

ClientCnxn初始化

|

public ClientCnxn(String chrootPath, HostProvider hostProvider, int sessionTimeout, ZooKeeper zooKeeper,

ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket,

long sessionId, byte[] sessionPasswd, boolean canBeReadOnly) {

this.zooKeeper = zooKeeper;

this.watcher = watcher;

this.sessionId = sessionId;

this.sessionPasswd = sessionPasswd;

this.sessionTimeout = sessionTimeout;

this.hostProvider = hostProvider;

this.chrootPath = chrootPath;

connectTimeout = sessionTimeout / hostProvider.size();

readTimeout = sessionTimeout * 2 / 3;

readOnly = canBeReadOnly;

sendThread = new SendThread(clientCnxnSocket); --初始化sendThread

eventThread = new EventThread(); --初始化eventThread

this.clientConfig=zooKeeper.getClientConfig();

}

public void start() { --启动两个线程

sendThread.start();

eventThread.start();

}

|

客户端通过exists注册监听

|

zookeeper.exists(“/mic”,true); //注册监听

|

通过exists方法来注册监听,代码如下

|

public Stat exists(final String path, Watcher watcher)

throws KeeperException, InterruptedException

{

final String clientPath = path;

PathUtils.validatePath(clientPath);

// the watch contains the un-chroot path

WatchRegistration wcb = null;

if (watcher != null) {

wcb = new ExistsWatchRegistration(watcher, clientPath); //构建ExistWatchRegistration

}

final String serverPath = prependChroot(clientPath);

RequestHeader h = new RequestHeader();

h.setType(ZooDefs.OpCode.exists); //设置操作类型为exists

ExistsRequest request = new ExistsRequest(); // 构造ExistsRequest

request.setPath(serverPath);

request.setWatch(watcher != null); //是否注册监听

SetDataResponse response = new SetDataResponse(); //设置服务端响应的接收类

//将封装的RequestHeader、ExistsRequest、SetDataResponse、WatchRegistration添加到发送队列

ReplyHeader r = cnxn.submitRequest(h, request, response, wcb);

if (r.getErr() != 0) {

if (r.getErr() == KeeperException.Code.NONODE.intValue()) {

return null;

}

throw KeeperException.create(KeeperException.Code.get(r.getErr()),

clientPath);

}

//返回exists得到的结果(Stat信息)

return response.getStat().getCzxid() == -1 ? null : response.getStat();

}

|

cnxn.submitRequest

|

public ReplyHeader submitRequest(RequestHeader h, Record request,

Record response, WatchRegistration watchRegistration,

WatchDeregistration watchDeregistration)

throws InterruptedException {

ReplyHeader r = new ReplyHeader();

//将消息添加到队列,并构造一个Packet传输对象

Packet packet = queuePacket(h, r, request, response, null, null, null, null, watchRegistration, watchDeregistration);

synchronized (packet) {

while (!packet.finished) { //在数据包没有处理完成之前,一直阻塞

packet.wait();

}

}

return r;

}

|

|

public Packet queuePacket(RequestHeader h, ReplyHeader r, Record request,

Record response, AsyncCallback cb, String clientPath,

String serverPath, Object ctx, WatchRegistration watchRegistration,

WatchDeregistration watchDeregistration) {

//将相关传输对象转化成Packet

Packet packet = null;

packet = new Packet(h, r, request, response, watchRegistration);

packet.cb = cb;

packet.ctx = ctx;

packet.clientPath = clientPath;

packet.serverPath = serverPath;

packet.watchDeregistration = watchDeregistration;

synchronized (state) {

if (!state.isAlive() || closing) {

conLossPacket(packet);

} else {

if (h.getType() == OpCode.closeSession) {

closing = true;

}

outgoingQueue.add(packet); //添加到outgoingQueue

}

}

sendThread.getClientCnxnSocket().packetAdded();//此处是多路复用机制,唤醒Selector,告诉他有数据包添加过来了

return packet;

}

|

在 ZooKeeper 中,Packet 是一个最小的通信协议单元,即数据包。Pakcet 用于进行客户端与服务端之间的网络传输,任何需要传输的对象都需要包装成一个 Packet 对象。在 ClientCnxn 中 WatchRegistration 也会被封装到 Pakcet 中,然后由 SendThread 线程调用queuePacket方法把 Packet 放入发送队列中等待客户端发送,这又是一个异步过程,分布式系统采用异步通信是一个非常常见的手段

SendThread的发送过程

在初始化连接的时候,zookeeper初始化了两个线程并且启动了。接下来我们来分析SendThread的发送过程,因为是一个线程,所以启动的时候会调用SendThread.run方法

|

public void run() {

clientCnxnSocket.introduce(this, sessionId, outgoingQueue);

clientCnxnSocket.updateNow();

clientCnxnSocket.updateLastSendAndHeard();

int to;

long lastPingRwServer = Time.currentElapsedTime();

final int MAX_SEND_PING_INTERVAL = 10000; //10 seconds

while (state.isAlive()) {

try {

if (!clientCnxnSocket.isConnected()) {// 如果没有连接:发起连接

// don't re-establish connection if we are closing

if (closing) {

break;

}

startConnect(); //发起连接

clientCnxnSocket.updateLastSendAndHeard();

}

if (state.isConnected()) { //如果是连接状态,则处理sasl的认证授权

// determine whether we need to send an AuthFailed event.

if (zooKeeperSaslClient != null) {

boolean sendAuthEvent = false;

if (zooKeeperSaslClient.getSaslState() == ZooKeeperSaslClient.SaslState.INITIAL) {

try {

zooKeeperSaslClient.initialize(ClientCnxn.this);

} catch (SaslException e) {

LOG.error("SASL authentication with Zookeeper Quorum member failed: " + e);

state = States.AUTH_FAILED;

sendAuthEvent = true;

}

}

KeeperState authState = zooKeeperSaslClient.getKeeperState();

if (authState != null) {

if (authState == KeeperState.AuthFailed) {

// An authentication error occurred during authentication with the Zookeeper Server.

state = States.AUTH_FAILED;

sendAuthEvent = true;

} else {

if (authState == KeeperState.SaslAuthenticated) {

sendAuthEvent = true;

}

}

}

if (sendAuthEvent == true) {

eventThread.queueEvent(new WatchedEvent(

Watcher.Event.EventType.None,

authState,null));

}

}

to = readTimeout - clientCnxnSocket.getIdleRecv();

} else {

to = connectTimeout - clientCnxnSocket.getIdleRecv();

}

//to,表示客户端距离timeout还剩多少时间,准备发起ping连接

if (to <= 0) {//表示已经超时了。

String warnInfo;

warnInfo = "Client session timed out, have not heard from server in "

  • clientCnxnSocket.getIdleRecv()
  • "ms"
  • " for sessionid 0x"
  • Long.toHexString(sessionId);

LOG.warn(warnInfo);

throw new SessionTimeoutException(warnInfo);

}

if (state.isConnected()) {

//计算下一次ping请求的时间

int timeToNextPing = readTimeout / 2 - clientCnxnSocket.getIdleSend() -

((clientCnxnSocket.getIdleSend() > 1000) ? 1000 : 0);

//send a ping request either time is due or no packet sent out within MAX_SEND_PING_INTERVAL

if (timeToNextPing <= 0 || clientCnxnSocket.getIdleSend() > MAX_SEND_PING_INTERVAL) {

sendPing(); //发送ping请求

clientCnxnSocket.updateLastSend();

} else {

if (timeToNextPing < to) {

to = timeToNextPing;

}

}

}

// If we are in read-only mode, seek for read/write server

if (state == States.CONNECTEDREADONLY) {

long now = Time.currentElapsedTime();

int idlePingRwServer = (int) (now - lastPingRwServer);

if (idlePingRwServer >= pingRwTimeout) {

lastPingRwServer = now;

idlePingRwServer = 0;

pingRwTimeout =

Math.min(2*pingRwTimeout, maxPingRwTimeout);

pingRwServer();

}

to = Math.min(to, pingRwTimeout - idlePingRwServer);

}

调用clientCnxnSocket,发起传输

其中 pendingQueue是一个用来存放已经发送、等待回应的Packet队列,

clientCnxnSocket默认使用ClientCnxnSocketNIO(ps:还记得在哪里初始化吗?在实例化zookeeper的时候)

clientCnxnSocket.doTransport(to, pendingQueue, ClientCnxn.this);

} catch (Throwable e) {

if (closing) {

if (LOG.isDebugEnabled()) {

// closing so this is expected

LOG.debug("An exception was thrown while closing send thread for session 0x"

  • Long.toHexString(getSessionId())
  • " : " + e.getMessage());

}

break;

} else {

// this is ugly, you have a better way speak up

if (e instanceof SessionExpiredException) {

LOG.info(e.getMessage() + ", closing socket connection");

} else if (e instanceof SessionTimeoutException) {

LOG.info(e.getMessage() + RETRY_CONN_MSG);

} else if (e instanceof EndOfStreamException) {

LOG.info(e.getMessage() + RETRY_CONN_MSG);

} else if (e instanceof RWServerFoundException) {

LOG.info(e.getMessage());

} else {

LOG.warn(

"Session 0x"

  • Long.toHexString(getSessionId())
  • " for server "
  • clientCnxnSocket.getRemoteSocketAddress()
  • ", unexpected error"
  • RETRY_CONN_MSG, e);

}

// At this point, there might still be new packets appended to outgoingQueue.

// they will be handled in next connection or cleared up if closed.

cleanup();

if (state.isAlive()) {

eventThread.queueEvent(new WatchedEvent(

Event.EventType.None,

Event.KeeperState.Disconnected,

null));

}

clientCnxnSocket.updateNow();

clientCnxnSocket.updateLastSendAndHeard();

}

}

}

synchronized (state) {

// When it comes to this point, it guarantees that later queued

// packet to outgoingQueue will be notified of death.

cleanup();

}

clientCnxnSocket.close();

if (state.isAlive()) {

eventThread.queueEvent(new WatchedEvent(Event.EventType.None,

Event.KeeperState.Disconnected, null));

}

ZooTrace.logTraceMessage(LOG, ZooTrace.getTextTraceLevel(),

"SendThread exited loop for session: 0x"

  • Long.toHexString(getSessionId()));

}

|

client 和 server的网络交互

|

@Override

void doTransport(int waitTimeOut, List<Packet> pendingQueue, ClientCnxn cnxn) throws IOException, InterruptedException {

try {

if (!firstConnect.await(waitTimeOut, TimeUnit.MILLISECONDS)) {

return;

}

Packet head = null;

if (needSasl.get()) {

if (!waitSasl.tryAcquire(waitTimeOut, TimeUnit.MILLISECONDS)) {

return;

}

} else {

//判断outgoingQueue是否存在待发送的数据包,不存在则直接返回

if ((head = outgoingQueue.poll(waitTimeOut, TimeUnit.MILLISECONDS)) == null) {

return;

}

}

// check if being waken up on closing.

if (!sendThread.getZkState().isAlive()) {

// adding back the patck to notify of failure in conLossPacket().

addBack(head);

return;

}

// channel disconnection happened

if (disconnected.get()) { //异常流程,channel关闭了,讲当前的packet添加到addBack中

addBack(head);

throw new EndOfStreamException("channel for sessionid 0x"

  • Long.toHexString(sessionId)
  • " is lost");

}

if (head != null) { //如果当前存在需要发送的数据包,则调用doWrite方法,pendingQueue表示处于已经发送过等待响应的packet队列

doWrite(pendingQueue, head, cnxn);

}

} finally {

updateNow();

}

}

|

DoWrite方法

|

private void doWrite(List<Packet> pendingQueue, Packet p, ClientCnxn cnxn) {

updateNow();

while (true) {

if (p != WakeupPacket.getInstance()) {

if ((p.requestHeader != null) && //判断请求头以及判断当前请求类型不是ping或者auth操作

(p.requestHeader.getType() != ZooDefs.OpCode.ping) &&

(p.requestHeader.getType() != ZooDefs.OpCode.auth)) {

p.requestHeader.setXid(cnxn.getXid()); //设置xid,这个xid用来区分请求类型

synchronized (pendingQueue) {

pendingQueue.add(p); //将当前的packet添加到pendingQueue队列中

}

}

sendPkt(p); //将数据包发送出去

}

if (outgoingQueue.isEmpty()) {

break;

}

p = outgoingQueue.remove();

}

}

|

sendPkt

|

private void sendPkt(Packet p) {

// Assuming the packet will be sent out successfully. Because if it fails,

// the channel will close and clean up queues.

p.createBB(); //序列化请求数据

updateLastSend(); //更新最后一次发送updateLastSend

sentCount++; //更新发送次数

channel.write(ChannelBuffers.wrappedBuffer(p.bb)); //通过nio channel发送字节缓存到服务端

}

|

createBB

|

public void createBB() {

try {

ByteArrayOutputStream baos = new ByteArrayOutputStream();

BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);

boa.writeInt(-1, "len"); // We'll fill this in later

if (requestHeader != null) {

requestHeader.serialize(boa, "header"); //序列化header头(requestHeader)

}

if (request instanceof ConnectRequest) {

request.serialize(boa, "connect");

// append "am-I-allowed-to-be-readonly" flag

boa.writeBool(readOnly, "readOnly");

} else if (request != null) {

request.serialize(boa, "request"); //序列化request(request)

}

baos.close();

this.bb = ByteBuffer.wrap(baos.toByteArray());

this.bb.putInt(this.bb.capacity() - 4);

this.bb.rewind();

} catch (IOException e) {

LOG.warn("Ignoring unexpected exception", e);

}

}

|

从createBB方法中,我们看到在底层实际的网络传输序列化中,zookeeper只会讲requestHeader和request两个属性进行序列化,即只有这两个会被序列化到底层字节数组中去进行网络传输,不会将watchRegistration相关的信息进行网络传输。

总结

用户调用exists注册监听以后,会做几个事情

  1. 讲请求数据封装为packet,添加到outgoingQueue
  2. SendThread这个线程会执行数据发送操作,主要是将outgoingQueue队列中的数据发送到服务端
  3. 通过clientCnxnSocket.doTransport(to, pendingQueue, ClientCnxn.this); 其中ClientCnxnSocket只zookeeper客户端和服务端的连接通信的封装,有两个具体的实现类ClientCnxnSocketNetty和ClientCnxnSocketNIO;具体使用哪一个类来实现发送,是在初始化过程是在实例化Zookeeper的时候设置的,代码如下

|

cnxn = new ClientCnxn(connectStringParser.getChrootPath(),

hostProvider, sessionTimeout, this, watchManager,

getClientCnxnSocket(), canBeReadOnly);

|

private ClientCnxnSocket getClientCnxnSocket() throws IOException {

String clientCnxnSocketName = getClientConfig().getProperty(

ZKClientConfig.ZOOKEEPER_CLIENT_CNXN_SOCKET);

if (clientCnxnSocketName == null) {

clientCnxnSocketName = ClientCnxnSocketNIO.class.getName();

}

try {

Constructor<?> clientCxnConstructor =

Class.forName(clientCnxnSocketName).getDeclaredConstructor(ZKClientConfig.class);

ClientCnxnSocket clientCxnSocket = (ClientCnxnSocket) clientCxnConstructor.newInstance(getClientConfig());

return clientCxnSocket;

} catch (Exception e) {

IOException ioe = new IOException("Couldn't instantiate "

  • clientCnxnSocketName);

ioe.initCause(e);

throw ioe;

}

}

|

  1. 基于第3步,最终会在ClientCnxnSocketNetty方法中执行sendPkt将请求的数据包发送到服务端

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏葡萄城控件技术团队

Wijmo 更优美的jQuery UI部件集:服务器端Grid魔法

在之前的一篇文章中,我们讨论了如何添加一大堆的超赞的功能到一个标准的HTML表,并把它转换成一个“grid”表格。今天我想要做的事情是向你展示如何将这些功能转到...

22560
来自专栏知识分享

51采集PCF8591数据通过ESP8266上传C#上位机android 之TCP客户端编程ESP8266使用详解NodeMCU初探ESP8266刷AT固件与nodemcu固件ESP8266使用详解-

这两天测试程序还发现一个bug就是如果客户端断开了,应该检测一下哪个断开了,数据就不应该发向那个连接,,,否则就会报错,然后模块会复位重启 所以加上这段代码 c...

61750
来自专栏听雨堂

用Nunit测试通讯程序

    对于Nunit,我是个纯粹的新手,没想到,第一个练手的,居然是一个通讯程序。难度略微大了点。我的通讯程序是一个类似通讯服务器的程序,能够监听和维持多个连...

20870
来自专栏跟着阿笨一起玩NET

使用windows服务和MSMQ和进行日志管理(解决高并发问题)

然后就可以写我们的代码了,我们的服务需要实时监视MSMQ的队列中有没有记录,如果有,就向数据库中插入

58210
来自专栏大内老A

我的WCF之旅(6):在Winform Application中调用Duplex Service出现TimeoutException的原因和解决方案

几个星期之前写了一篇关于如何通过WCF进行 双向通信的文章([原创]我的WCF之旅(3):在WCF中实现双向通信(Bi-directional Communic...

22980
来自专栏一只程序汪的自我修养

ASP.NET MVC 微信JS-SDK认证

31130
来自专栏跟着阿笨一起玩NET

ASP.NET datable导出excel

本文转载:http://www.cnblogs.com/chwkai/archive/2005/10/08/250426.html

10820
来自专栏cnblogs

Reactor模式的.net版本简单实现--DEMO

     近期在学习DotNetty,遇到不少的问题。由于dotnetty是次netty的.net版本的实现。导致在网上叙述dotnetty的原理,以及实现技巧...

42060
来自专栏技术之路

Caliburn.Micro学习笔记(四)----IHandle<T>实现多语言功能

说一下IHandle<T>实现多语言功能 因为Caliburn.Micro是基于MvvM的UI与codebehind分离, binding可以是双向的所以我们想...

26370
来自专栏晓晨的专栏

.NET Core 实现定时抓取博客园首页文章信息并发送到邮箱

大家好,我是晓晨。许久没有更新博客了,今天给大家带来一篇干货型文章,一个每隔5分钟抓取博客园首页文章信息并在第二天的上午9点发送到你的邮箱的小工具。比如我在20...

34180

扫码关注云+社区

领取腾讯云代金券