汽车最全的网站,杭州产品设计公司有哪些,c++软件开发需要学什么,一个主体如何添加网站目录 回顾回调观察者模式发布订阅模式Zookeeper 客户端/ 服务端 watchgetChildren 为例最后归纳 回顾回调观察者模式发布订阅模式
回调的思想
类A的a()方法调用类B的b()方法类B的b()方法执行完毕主动调用类A的callback()方法
回调分为同步回调和异步回调… 目录 回顾回调观察者模式发布订阅模式Zookeeper 客户端/ 服务端 watchgetChildren 为例最后归纳 回顾回调观察者模式发布订阅模式
回调的思想
类A的a()方法调用类B的b()方法类B的b()方法执行完毕主动调用类A的callback()方法
回调分为同步回调和异步回调, 假如以买彩票的场景来模拟, 我买彩票, 调用彩票网,给我返回的结果确定是否中奖,同步回调就是,我买了彩票之后, 需要等待彩票网给我返回的结果, 这个时候我不能做其他事情, 我必须等待这个结果, 这就叫同步回调, 同步, 就意味着等待, 我不能去做其他事情, 必须等待, 异步回调就是, 我买了彩票之后, 可以去做其他事情, 然后当彩票网有了结果和消息, 再给我返回消息。
观察者模式 发布订阅对比 观察者模式
Zookeeper 客户端/ 服务端 watch
客户端维持的 socket 连接 ClientCnxn
/*** This class manages the socket i/o for the client. ClientCnxn maintains a list* of available servers to connect to and transparently switches servers it is* connected to as needed.**/
public class ClientCnxn {/*** Manage watchers handle events generated by the ClientCnxn object.** We are implementing this as a nested class of ZooKeeper so that* the public methods will not be exposed as part of the ZooKeeper client* API.*/static class ZKWatchManager implements ClientWatchManager {服务端 DataTree
/*** This class maintains the tree data structure. It doesnt have any networking* or client connection code in it so that it can be tested in a stand alone* way.* p* The tree maintains two parallel data structures: a hashtable that maps from* full paths to DataNodes and a tree of DataNodes. All accesses to a path is* through the hashtable. The tree is traversed only when serializing to disk.*/
public class DataTree {getChildren 为例
/*** Return the list of the children of the node of the given path.* p* If the watch is non-null and the call is successful (no exception is thrown),* a watch will be left on the node with the given path. The watch willbe* triggered by a successful operation that deletes the node of the given* path or creates/delete a child under the node.* p* The list of children returned is not sorted and no guarantee is provided* as to its natural or lexical order.* p* A KeeperException with error code KeeperException.NoNode will be thrown* if no node with the given path exists.** param path* param watcher explicit watcher* return an unordered array of children of the node with the given path* throws InterruptedException If the server transaction is interrupted.* throws KeeperException If the server signals an error with a non-zero error code.* throws IllegalArgumentException if an invalid path is specified*/public ListString getChildren(final String path, Watcher watcher)throws KeeperException, InterruptedException{final String clientPath path;PathUtils.validatePath(clientPath);// the watch contains the un-chroot pathWatchRegistration wcb null;if (watcher ! null) {wcb new ChildWatchRegistration(watcher, clientPath);}final String serverPath prependChroot(clientPath);RequestHeader h new RequestHeader();h.setType(ZooDefs.OpCode.getChildren);GetChildrenRequest request new GetChildrenRequest();request.setPath(serverPath);request.setWatch(watcher ! null);GetChildrenResponse response new GetChildrenResponse();ReplyHeader r cnxn.submitRequest(h, request, response, wcb);if (r.getErr() ! 0) {throw KeeperException.create(KeeperException.Code.get(r.getErr()),clientPath);}return response.getChildren();}ReplyHeader r cnxn.submitRequest(h, request, response, wcb); 发送请求给服务端
public ReplyHeader submitRequest(RequestHeader h, Record request, Record response, WatchRegistration watchRegistration) throws InterruptedException {ReplyHeader r new ReplyHeader();// 客户端与服务端的网络传输ClientCnxn.Packet packet this.queuePacket(h, r, request, response, (AsyncCallback)null, (String)null, (String)null, (Object)null, watchRegistration);synchronized(packet) {while(!packet.finished) {packet.wait();}return r;}
}ClientCnxn.Packet queuePacket(RequestHeader h, ReplyHeader r, Record request, Record response, AsyncCallback cb, String clientPath, String serverPath, Object ctx, WatchRegistration watchRegistration) {ClientCnxn.Packet packet null;LinkedList var11 this.outgoingQueue;synchronized(this.outgoingQueue) {// 传输的对象都包装成Packet对象packet new ClientCnxn.Packet(h, r, request, response, watchRegistration);packet.cb cb;packet.ctx ctx;packet.clientPath clientPath;packet.serverPath serverPath;if (this.state.isAlive() !this.closing) {if (h.getType() -11) {this.closing true;}// 放入发送队列中等待发送this.outgoingQueue.add(packet);} else {this.conLossPacket(packet);}}this.sendThread.getClientCnxnSocket().wakeupCnxn();return packet;
}outgoingQueue的处理 服务端org.apache.zookeeper.server.FinalRequestProcessor#processRequest处理 case OpCode.getChildren: {lastOp GETC;GetChildrenRequest getChildrenRequest new GetChildrenRequest();ByteBufferInputStream.byteBuffer2Record(request.request,getChildrenRequest);DataNode n zks.getZKDatabase().getNode(getChildrenRequest.getPath());if (n null) {throw new KeeperException.NoNodeException();}PrepRequestProcessor.checkACL(zks, zks.getZKDatabase().aclForNode(n),ZooDefs.Perms.READ,request.authInfo);// 返回children,// 这里根据客户端设置的是否有watch变量来传入watcher对象// 如果true则将当前的ServerCnxn传入ServerCnxn代表客户端和服务端的连接 ListString children zks.getZKDatabase().getChildren(getChildrenRequest.getPath(), null, getChildrenRequest.getWatch() ? cnxn : null);rsp new GetChildrenResponse(children);break;}将数据节点路径和ServerCnxn对象存储在WatcherManager的watchTable和watch2Paths中 public ListString getChildren(String path, Stat stat, Watcher watcher)throws KeeperException.NoNodeException {DataNode n nodes.get(path);if (n null) {throw new KeeperException.NoNodeException();}synchronized (n) {if (stat ! null) {n.copyStat(stat);}ListString childrennew ArrayListString(n.getChildren());if (watcher ! null) {childWatches.addWatch(path, watcher);}return children;}}当服务端处理完毕之后客户端的SendThread线程负责接收服务端的响应finishPacket方法会从packet中取出WatchRegistration并注册到ZKWatchManager中 /*** This class services the outgoing request queue and generates the heart* beats. It also spawns the ReadThread.*/class SendThread extends ZooKeeperThread {private long lastPingSentNs;private final ClientCnxnSocket clientCnxnSocket;private Random r new Random(System.nanoTime()); private boolean isFirstConnect true;void readResponse(ByteBuffer incomingBuffer) throws IOException {ByteBufferInputStream bbis new ByteBufferInputStream(incomingBuffer);BinaryInputArchive bbia BinaryInputArchive.getArchive(bbis);ReplyHeader replyHdr new ReplyHeader();replyHdr.deserialize(bbia, header);if (replyHdr.getXid() -2) {// -2 is the xid for pingsif (LOG.isDebugEnabled()) {LOG.debug(Got ping response for sessionid: 0x Long.toHexString(sessionId) after ((System.nanoTime() - lastPingSentNs) / 1000000) ms);}return;}if (replyHdr.getXid() -4) {// -4 is the xid for AuthPacket if(replyHdr.getErr() KeeperException.Code.AUTHFAILED.intValue()) {state States.AUTH_FAILED; eventThread.queueEvent( new WatchedEvent(Watcher.Event.EventType.None, Watcher.Event.KeeperState.AuthFailed, null) ); }if (LOG.isDebugEnabled()) {LOG.debug(Got auth sessionid:0x Long.toHexString(sessionId));}return;}if (replyHdr.getXid() -1) {// -1 means notificationif (LOG.isDebugEnabled()) {LOG.debug(Got notification sessionid:0x Long.toHexString(sessionId));}WatcherEvent event new WatcherEvent();event.deserialize(bbia, response);// convert from a server path to a client pathif (chrootPath ! null) {String serverPath event.getPath();if(serverPath.compareTo(chrootPath)0)event.setPath(/);else if (serverPath.length() chrootPath.length())event.setPath(serverPath.substring(chrootPath.length()));else {LOG.warn(Got server path event.getPath() which is too short for chroot path chrootPath);}}WatchedEvent we new WatchedEvent(event);if (LOG.isDebugEnabled()) {LOG.debug(Got we for sessionid 0x Long.toHexString(sessionId));}eventThread.queueEvent( we );return;}// If SASL authentication is currently in progress, construct and// send a response packet immediately, rather than queuing a// response as with other packets.if (tunnelAuthInProgress()) {GetSASLRequest request new GetSASLRequest();request.deserialize(bbia,token);zooKeeperSaslClient.respondToServer(request.getToken(),ClientCnxn.this);return;}Packet packet;synchronized (pendingQueue) {if (pendingQueue.size() 0) {throw new IOException(Nothing in the queue, but got replyHdr.getXid());}packet pendingQueue.remove();}/** Since requests are processed in order, we better get a response* to the first request!*/try {if (packet.requestHeader.getXid() ! replyHdr.getXid()) {packet.replyHeader.setErr(KeeperException.Code.CONNECTIONLOSS.intValue());throw new IOException(Xid out of order. Got Xid replyHdr.getXid() with err replyHdr.getErr() expected Xid packet.requestHeader.getXid() for a packet with details: packet );}packet.replyHeader.setXid(replyHdr.getXid());packet.replyHeader.setErr(replyHdr.getErr());packet.replyHeader.setZxid(replyHdr.getZxid());if (replyHdr.getZxid() 0) {lastZxid replyHdr.getZxid();}if (packet.response ! null replyHdr.getErr() 0) {packet.response.deserialize(bbia, response);}if (LOG.isDebugEnabled()) {LOG.debug(Reading reply sessionid:0x Long.toHexString(sessionId) , packet:: packet);}} finally {finishPacket(packet);}}private void finishPacket(Packet p) {int err p.replyHeader.getErr();if (p.watchRegistration ! null) {p.watchRegistration.register(err);}// Add all the removed watch events to the event queue, so that the// clients will be notified with Data/Child WatchRemoved event type.if (p.watchDeregistration ! null) {MapEventType, SetWatcher materializedWatchers null;try {materializedWatchers p.watchDeregistration.unregister(err);for (EntryEventType, SetWatcher entry : materializedWatchers.entrySet()) {SetWatcher watchers entry.getValue();if (watchers.size() 0) {queueEvent(p.watchDeregistration.getClientPath(), err,watchers, entry.getKey());// ignore connectionloss when removing from local// sessionp.replyHeader.setErr(Code.OK.intValue());}}} catch (KeeperException.NoWatcherException nwe) {LOG.error(Failed to find watcher!, nwe);p.replyHeader.setErr(nwe.code().intValue());} catch (KeeperException ke) {LOG.error(Exception when removing watcher, ke);p.replyHeader.setErr(ke.code().intValue());}}if (p.cb null) {synchronized (p) {p.finished true;p.notifyAll();}} else {p.finished true;eventThread.queuePacket(p);}}触发watcher org.apache.zookeeper.server.WatchManager#triggerWatch SetWatcher triggerWatch(String path, EventType type) {return triggerWatch(path, type, null);}SetWatcher triggerWatch(String path, EventType type, SetWatcher supress) {WatchedEvent e new WatchedEvent(type,KeeperState.SyncConnected, path);HashSetWatcher watchers;// 主要做的就是从watchTable和watch2Paths中移除该路径的watcherWatcher机制是一次性的synchronized (this) {watchers watchTable.remove(path);if (watchers null || watchers.isEmpty()) {if (LOG.isTraceEnabled()) {ZooTrace.logTraceMessage(LOG,ZooTrace.EVENT_DELIVERY_TRACE_MASK,No watchers for path);}return null;}for (Watcher w : watchers) {HashSetString paths watch2Paths.get(w);if (paths ! null) {paths.remove(path);}}}for (Watcher w : watchers) {if (supress ! null supress.contains(w)) {continue;}// 真正的回调和业务逻辑执行都在客户端org.apache.zookeeper.server.NIOServerCnxn#processw.process(e);}return watchers;}最后归纳
流程
客户端把注册的Watcher传到服务端处理请求加入处理队列服务端从处理队列取出事件并处理请求返回给客户端回调Watcher处理在客户端处理并会被删除
