devbox
2023面试高手
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

zookeeper源码-watcher监听_eventthread shut down for session: 0x188f1a94ad724

作者：2023面试高手 | 2024-02-16 15:31:19

赞

踩

eventthread shut down for session: 0x188f1a94ad724db

ClientCnxn类的readResponse()方法 

  1.     //客户端读取返回值    
  2.     void readResponse(ByteBuffer incomingBuffer) throws IOException {
  3.             ByteBufferInputStream bbis = new ByteBufferInputStream(
  4.                     incomingBuffer);
  5.             BinaryInputArchive bbia = BinaryInputArchive.getArchive(bbis);
  6.             ReplyHeader replyHdr = new ReplyHeader();
  7.  
  8.             replyHdr.deserialize(bbia, "header");
  9.             if (replyHdr.getXid() == -2) {
  10.                 // -2 is the xid for pings
  11.                 if (LOG.isDebugEnabled()) {
  12.                     LOG.debug("Got ping response for sessionid: 0x"
  13.                             + Long.toHexString(sessionId)
  14.                             + " after "
  15.                             + ((System.nanoTime() - lastPingSentNs) / 1000000)
  16.                             + "ms");
  17.                 }
  18.                 return;
  19.             }
  20.             if (replyHdr.getXid() == -4) {
  21.                 // -4 is the xid for AuthPacket               
  22.                 if(replyHdr.getErr() == KeeperException.Code.AUTHFAILED.intValue()) {
  23.                     state = States.AUTH_FAILED;                    
  24.                     eventThread.queueEvent( new WatchedEvent(Watcher.Event.EventType.None, 
  25.                             Watcher.Event.KeeperState.AuthFailed, null) );            		            		
  26.                 }
  27.                 if (LOG.isDebugEnabled()) {
  28.                     LOG.debug("Got auth sessionid:0x"
  29.                             + Long.toHexString(sessionId));
  30.                 }
  31.                 return;
  32.             }
  33.             //-1代表是通知
  34.             if (replyHdr.getXid() == -1) {
  35.                 // -1 means notification
  36.                 if (LOG.isDebugEnabled()) {
  37.                     LOG.debug("Got notification sessionid:0x"
  38.                         + Long.toHexString(sessionId));
  39.                 }
  40. //从服务器返回读取事件数据
  41.                 WatcherEvent event = new WatcherEvent();
  42.                 event.deserialize(bbia, "response");
  43.                 //转换server到client
  44.                 // convert from a server path to a client path
  45.                 if (chrootPath != null) {
  46.                     String serverPath = event.getPath();
  47.                     if(serverPath.compareTo(chrootPath)==0)
  48.                         event.setPath("/");
  49.                     else if (serverPath.length() > chrootPath.length())
  50.                         event.setPath(serverPath.substring(chrootPath.length()));
  51.                     else {
  52.                     	LOG.warn("Got server path " + event.getPath()
  53.                     			+ " which is too short for chroot path "
  54.                     			+ chrootPath);
  55.                     }
  56.                 }
  57.                 //触发watch事件，并将事件放进队列
  58.                 WatchedEvent we = new WatchedEvent(event);
  59.                 if (LOG.isDebugEnabled()) {
  60.                     LOG.debug("Got " + we + " for sessionid 0x"
  61.                             + Long.toHexString(sessionId));
  62.                 }
  63.  
  64.                 eventThread.queueEvent( we );
  65.                 return;
  66.             }
  67.  
  68.             // If SASL authentication is currently in progress, construct and
  69.             // send a response packet immediately, rather than queuing a
  70.             // response as with other packets.
  71.             if (clientTunneledAuthenticationInProgress()) {
  72.                 GetSASLRequest request = new GetSASLRequest();
  73.                 request.deserialize(bbia,"token");
  74.                 zooKeeperSaslClient.respondToServer(request.getToken(),
  75.                   ClientCnxn.this);
  76.                 return;
  77.             }
  78.  
  79.             Packet packet;
  80.             synchronized (pendingQueue) {
  81.                 if (pendingQueue.size() == 0) {
  82.                     throw new IOException("Nothing in the queue, but got "
  83.                             + replyHdr.getXid());
  84.                 }
  85.                 packet = pendingQueue.remove();
  86.             }
  87.             /*
  88.              * Since requests are processed in order, we better get a response
  89.              * to the first request!
  90.              */
  91.             try {
  92.                 if (packet.requestHeader.getXid() != replyHdr.getXid()) {
  93.                     packet.replyHeader.setErr(
  94.                             KeeperException.Code.CONNECTIONLOSS.intValue());
  95.                     throw new IOException("Xid out of order. Got Xid "
  96.                             + replyHdr.getXid() + " with err " +
  97.                             + replyHdr.getErr() +
  98.                             " expected Xid "
  99.                             + packet.requestHeader.getXid()
  100.                             + " for a packet with details: "
  101.                             + packet );
  102.                 }
  103.  
  104.                 packet.replyHeader.setXid(replyHdr.getXid());
  105.                 packet.replyHeader.setErr(replyHdr.getErr());
  106.                 packet.replyHeader.setZxid(replyHdr.getZxid());
  107.                 if (replyHdr.getZxid() > 0) {
  108.                     lastZxid = replyHdr.getZxid();
  109.                 }
  110.                 if (packet.response != null && replyHdr.getErr() == 0) {
  111.                     packet.response.deserialize(bbia, "response");
  112.                 }
  113.  
  114.                 if (LOG.isDebugEnabled()) {
  115.                     LOG.debug("Reading reply sessionid:0x"
  116.                             + Long.toHexString(sessionId) + ", packet:: " + packet);
  117.                 }
  118.             } finally {
  119.                 finishPacket(packet);
  120.             }
  121.         }
queueEvent()方法属于ClientCnxn类的内部类EventThread
  1.     class EventThread extends ZooKeeperThread {
  2.         private final LinkedBlockingQueue<Object> waitingEvents =
  3.             new LinkedBlockingQueue<Object>();
  4.  
  5.         /** This is really the queued session state until the event
  6.          * thread actually processes the event and hands it to the watcher.
  7.          * But for all intents and purposes this is the state.
  8.          */
  9.         private volatile KeeperState sessionState = KeeperState.Disconnected;
  10.  
  11.        private volatile boolean wasKilled = false;
  12.        private volatile boolean isRunning = false;
  13.  
  14.         EventThread() {
  15.             super(makeThreadName("-EventThread"));
  16.             setDaemon(true);
  17.         }  
  18.  
  19.  
  20.    
  21.     private final LinkedBlockingQueue<Object> waitingEvents =
  22.             new LinkedBlockingQueue<Object>();
  23.         //初始是关闭
  24.         private volatile KeeperState sessionState = KeeperState.Disconnected;
  25.  
  26. public void queueEvent(WatchedEvent event) {
  27. //如果KeeperState状态关闭，并且Event状态是-1，则直接退出
  28.             if (event.getType() == EventType.None
  29.                     && sessionState == event.getState()) {
  30.                 return;
  31.             }
  32.             sessionState = event.getState();
  33.             //这是Event参数
  34.             // materialize the watchers based on the event
  35.             WatcherSetEventPair pair = new WatcherSetEventPair(
  36.             //触发materialize方法,返回对应的目录文件数据
  37.                     watcher.materialize(event.getState(), event.getType(),
  38.                             event.getPath()),
  39.                             event);
  40.             // queue the pair (watch set & event) for later processing
  41.             //放进waiting事件
  42.             waitingEvents.add(pair);
  43.         }

materialize方法

  1.         @Override
  2.         public Set<Watcher> materialize(Watcher.Event.KeeperState state,
  3.                                         Watcher.Event.EventType type,
  4.                                         String clientPath)
  5.         {
  6.             Set<Watcher> result = new HashSet<Watcher>();
  7.  
  8.             switch (type) {
  9. //当事件类型为None时，也就是客户端和服务端的租约到期，或者认证失败或者断开连接或者建立连接时（建立连接时不会清除watcher,客户端收到反馈后会返回所有事件并触发，当事件类型是节点数据变化或者节点创建时则触发dataWatches和existWatches，当事件类型是子节点变化时childWatches被触发，当事件类型是删除节点时三种watcher都会触发。
  10.             case None:
  11.                 result.add(defaultWatcher);
  12. //满足两种情况则清空dataWatches,existWatches，childWatches  
  13.                 //1.状态不是SyncConnected（客户端与服务器端已建立链接）  
  14.                 //2.设置了System property:zookeepr.disableAutoWatchReset 为true  
  15.                 boolean clear = ClientCnxn.getDisableAutoResetWatch() &&
  16.                         state != Watcher.Event.KeeperState.SyncConnected;
  17.                 synchronized(dataWatches) {
  18.                     for(Set<Watcher> ws: dataWatches.values()) {
  19.                         result.addAll(ws);
  20.                     }
  21.                     if (clear) {
  22.                         dataWatches.clear();
  23.                     }
  24.                 }
  25.  
  26.                 synchronized(existWatches) {
  27.                     for(Set<Watcher> ws: existWatches.values()) {
  28.                         result.addAll(ws);
  29.                     }
  30.                     if (clear) {
  31.                         existWatches.clear();
  32.                     }
  33.                 }
  34.  
  35.                 synchronized(childWatches) {
  36.                     for(Set<Watcher> ws: childWatches.values()) {
  37.                         result.addAll(ws);
  38.                     }
  39.                     if (clear) {
  40.                         childWatches.clear();
  41.                     }
  42.                 }
  43.  
  44.                 return result;
  45.             case NodeDataChanged:
  46.             //当为创建时候，根据子类事件类型触发相应事件
  47.             case NodeCreated:
  48.                 synchronized (dataWatches) {
  49.                     addTo(dataWatches.remove(clientPath), result);
  50.                 }
  51.                 synchronized (existWatches) {
  52.                     addTo(existWatches.remove(clientPath), result);
  53.                 }
  54.                 break;
  55. //当为子目录事件时候，根据子类事件类型触发相应事件
  56.             case NodeChildrenChanged:
  57.                 synchronized (childWatches) {
  58.                     addTo(childWatches.remove(clientPath), result);
  59.                 }
  60.                 break;
  61. //当为删除事件时候，根据子类事件类型触发相应事件
  62.             case NodeDeleted:
  63.                 synchronized (dataWatches) {
  64.                     addTo(dataWatches.remove(clientPath), result);
  65.                 }
  66.                 // XXX This shouldn't be needed, but just in case
  67.                 synchronized (existWatches) {
  68.                     Set<Watcher> list = existWatches.remove(clientPath);
  69.                     if (list != null) {
  70.                         addTo(existWatches.remove(clientPath), result);
  71.                         LOG.warn("We are triggering an exists watch for delete! Shouldn't happen!");
  72.                     }
  73.                 }
  74.                 synchronized (childWatches) {
  75.                     addTo(childWatches.remove(clientPath), result);
  76.                 }
  77.                 break;
  78.             default:
  79.                 String msg = "Unhandled watch event type " + type
  80.                     + " with state " + state + " on path " + clientPath;
  81.                 LOG.error(msg);
  82.                 throw new RuntimeException(msg);
  83.             }
  84.  
  85.             return result;
  86.         }
  87.  
  88.  
  89.         //将form添加至to
  90.         final private void addTo(Set<Watcher> from, Set<Watcher> to) {
  91.             if (from != null) {
  92.                 to.addAll(from);
  93.             }
  94.         }

EventThread线程的run方法

  1.         @Override
  2.         public void run() {
  3.            try {
  4.               isRunning = true;
  5.               while (true) {
  6. //从队列中获取事件数据
  7.                  Object event = waitingEvents.take();
  8.                  if (event == eventOfDeath) {
  9.                  //是否关闭线程
  10.                     wasKilled = true;
  11.                  } else {
  12. //处理事件
  13.                     processEvent(event);
  14.                  }
  15.                  if (wasKilled)
  16.                     synchronized (waitingEvents) {
  17.                        if (waitingEvents.isEmpty()) {
  18.                           isRunning = false;
  19.                           break;
  20.                        }
  21.                     }
  22.               }
  23.            } catch (InterruptedException e) {
  24.               LOG.error("Event thread exiting due to interruption", e);
  25.            }
  26.  
  27.             LOG.info("EventThread shut down for session: 0x{}",
  28.                      Long.toHexString(getSessionId()));
  29.         }

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/2023面试高手/article/detail/94747
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号