欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

Zookeeper Session机制

程序员文章站 2022-06-26 19:15:37
...

1、zookeeper session是什么

    客户端与服务端的任何交换操作都与会话相关,包括:临时节点的生命周期,客户端的请求顺序执行以及Watcher通知机制。

 2、客户端操作

2.1、Zookeeper客户端一次创建会话的过程。

Zookeeper Session机制
            
    
    博客分类: Zookeeper  

 

2.2、构造会话请求

long sessId = (seenRwServerBefore) ? sessionId : 0;
ConnectRequest conReq = new ConnectRequest(0, lastZxid,sessionTimeout, sessId, sessionPasswd);
outgoingQueue.addFirst(new Packet(null, null, conReq,null, null, readOnly));

 

2.3、处理Response

ConnectResponse conRsp = new ConnectResponse();
conRsp.deserialize(bbia, "connect");
this.sessionId = conRsp.getSessionId();
sendThread.onConnected(conRsp.getTimeOut(),this.sessionId,conRsp.getPasswd(), isRO);

2.3.1、更新客户端Timeout等信息

void onConnected(int _negotiatedSessionTimeout, long _sessionId,
                byte[] _sessionPasswd, boolean isRO) throws IOException {
            negotiatedSessionTimeout = _negotiatedSessionTimeout;
            if (negotiatedSessionTimeout <= 0) {
                state = States.CLOSED;

                eventThread.queueEvent(new WatchedEvent(
                        Watcher.Event.EventType.None,
                        Watcher.Event.KeeperState.Expired, null));
                eventThread.queueEventOfDeath();
                throw new SessionExpiredException(
                        "Unable to reconnect to ZooKeeper service, session 0x"
                                + Long.toHexString(sessionId) + " has expired");
            }
            if (!readOnly && isRO) {
                LOG.error("Read/write client got connected to read-only server");
            }
            readTimeout = negotiatedSessionTimeout * 2 / 3;
            connectTimeout = negotiatedSessionTimeout / hostProvider.size();
            hostProvider.onConnected();
            sessionId = _sessionId;
            sessionPasswd = _sessionPasswd;
            state = (isRO) ?
                    States.CONNECTEDREADONLY : States.CONNECTED;
            seenRwServerBefore |= !isRO;
            LOG.info("Session establishment complete on server "
                    + clientCnxnSocket.getRemoteSocketAddress()
                    + ", sessionid = 0x" + Long.toHexString(sessionId)
                    + ", negotiated timeout = " + negotiatedSessionTimeout
                    + (isRO ? " (READ-ONLY mode)" : ""));
            KeeperState eventState = (isRO) ?
                    KeeperState.ConnectedReadOnly : KeeperState.SyncConnected;
            eventThread.queueEvent(new WatchedEvent(
                    Watcher.Event.EventType.None,
                    eventState, null));
        }

 

2.4、发送Ping请求

    客户端当sessionTimeout/3(readTimeout=sessionTimeout * 2 /3 )之内都没有发送通信请求后,会主动发起一个ping请求,以维持会话状态.

if (state.isConnected()) {
    int timeToNextPing = readTimeout / 2
                          - clientCnxnSocket.getIdleSend();
   if (timeToNextPing <= 0) {
             sendPing();
            clientCnxnSocket.updateLastSend();
    } else {
             if (timeToNextPing < to) {
                to = timeToNextPing;
              }
     }
}

 

 3、服务端操作 

3.1、服务端会话处理流程示意图

    Zookeeper Session机制
            
    
    博客分类: Zookeeper  

 

3.2、协商sessionTimeout

    2Ticktime <= sessionTimeout <= 20Ticktime(tickTime默认等于2s)

int sessionTimeout = connReq.getTimeOut();
 byte passwd[] = connReq.getPasswd();
 int minSessionTimeout = getMinSessionTimeout();
if (sessionTimeout < minSessionTimeout) {
      sessionTimeout = minSessionTimeout;
}
int maxSessionTimeout = getMaxSessionTimeout();
if (sessionTimeout > maxSessionTimeout) {
     sessionTimeout = maxSessionTimeout;
 }
cnxn.setSessionTimeout(sessionTimeout);

 

3.3、判断是否需要创建新会话

long sessionId = connReq.getSessionId();
 if (sessionId != 0) {
       long clientSessionId = connReq.getSessionId();
       LOG.info("Client attempting to renew session 0x"+ Long.toHexString(clientSessionId) + " at " + cnxn.getRemoteSocketAddress());
       serverCnxnFactory.closeSession(sessionId);
       cnxn.setSessionId(sessionId);
       reopenSession(cnxn, sessionId, passwd, sessionTimeout);
   } else {
        LOG.info("Client attempting to establish new session at "
                    + cnxn.getRemoteSocketAddress());
        createSession(cnxn, passwd, sessionTimeout);
  }

 

3.4、创建sessionId

     每次客户端创建新会话时,zookeeper都会为其分配一个全局唯一的sessionID。

public static long initializeNextSession(long id) {
        long nextSid = 0;
        nextSid = (System.currentTimeMillis() << 24) >> 8;
        nextSid =  nextSid | (id <<56);
        return nextSid;
}
synchronized public long createSession(int sessionTimeout) {
     addSession(nextSessionId, sessionTimeout);
     return nextSessionId++;
}

 

 

 3.5、注册会话

 

SessionTracker中维护两个数据结构sessionsWithTimeout和sessionsById,前者根据sessionID保存了所有会话超时时间,后者则根据sessionID保存了会话实体。

 

  HashMap<Long, SessionImpl> sessionsById = new HashMap<Long, SessionImpl>();
  ConcurrentHashMap<Long, Integer> sessionsWithTimeout; 

   synchronized public void addSession(long id, int sessionTimeout) {
        sessionsWithTimeout.put(id, sessionTimeout);
        if (sessionsById.get(id) == null) {
            SessionImpl s = new SessionImpl(id, sessionTimeout, 0);
            sessionsById.put(id, s);
            if (LOG.isTraceEnabled()) {
                ZooTrace.logTraceMessage(LOG, ZooTrace.SESSION_TRACE_MASK,
                        "SessionTrackerImpl --- Adding session 0x"
                        + Long.toHexString(id) + " " + sessionTimeout);
            }
        } else {
            if (LOG.isTraceEnabled()) {
                ZooTrace.logTraceMessage(LOG, ZooTrace.SESSION_TRACE_MASK,
                        "SessionTrackerImpl --- Existing session 0x"
                        + Long.toHexString(id) + " " + sessionTimeout);
            }
        }
        touchSession(id, sessionTimeout);
    }

 

 3.6、激活会话

 

  3.6.1、分桶策略

   

将类似的会话放在同一区块中进行管理,以便于不同区块的隔离,相同区块的统一处理。

   分配的原则:会话的下一次超时的时间点(ExpirationTime, ExpirationInterval=tickTime 单位:毫秒)。

   ExpirationTime = CurrentTime + SessionTimeout

   ExpirationTime =  (ExpirationTime / ExpirationInterval +1) x ExpirationInterval

   以ExpirationTime为key存入Map中,实现了分桶。

 

 3.6.2、激活

 

   客户端会向服务端发送ping请求来保持会话的有效,服务端收到心跳会不断的更新对应的客户端会话。

  

HashMap<Long, SessionSet> sessionSets = new HashMap<Long, SessionSet>();

private long roundToInterval(long time) {
        // We give a one interval grace period
        return (time / expirationInterval + 1) * expirationInterval;
}
 synchronized public boolean touchSession(long sessionId, int timeout) {
        if (LOG.isTraceEnabled()) {
            ZooTrace.logTraceMessage(LOG,
                                     ZooTrace.CLIENT_PING_TRACE_MASK,
                                     "SessionTrackerImpl --- Touch session: 0x"
                    + Long.toHexString(sessionId) + " with timeout " + timeout);
        }
        SessionImpl s = sessionsById.get(sessionId);
        // Return false, if the session doesn't exists or marked as closing
        if (s == null || s.isClosing()) {
            return false;
        }
        long expireTime = roundToInterval(System.currentTimeMillis() + timeout);
        if (s.tickTime >= expireTime) {
            // Nothing needs to be done
            return true;
        }
        SessionSet set = sessionSets.get(s.tickTime);
        if (set != null) {
            set.sessions.remove(s);
        }
        s.tickTime = expireTime;
        set = sessionSets.get(s.tickTime);
        if (set == null) {
            set = new SessionSet();
            sessionSets.put(expireTime, set);
        }
        set.sessions.add(s);
        return true;
    }

 

3.7、会话清理

 

3.7.1、会话超时检测

       分桶策略的数据迁移是被动触发,没有做迁移的sessionId会一直保存在之前的桶中,SessionTracker线程会检测到过期(没有迁移的)的会话。

 

3.7.2、清理过程

1、标记会话为“已关闭”

2、发起“会话关闭”请求

3、收集要清理的临时节点

4、添加“节点删除”事务

5、删除临时节点

6、移除会话

7、关闭NioServerCnxn

 

    synchronized public void run() {
        try {
            while (running) {
                currentTime = System.currentTimeMillis();
                if (nextExpirationTime > currentTime) {
                    this.wait(nextExpirationTime - currentTime);
                    continue;
                }
                SessionSet set;
                set = sessionSets.remove(nextExpirationTime);
                if (set != null) {
                    for (SessionImpl s : set.sessions) {
                        setSessionClosing(s.sessionId);
                        expirer.expire(s);
                    }
                }
                nextExpirationTime += expirationInterval;
            }
        } catch (InterruptedException e) {
            LOG.error("Unexpected interruption", e);
        }
        LOG.info("SessionTrackerImpl exited loop!");
    }

    synchronized public void setSessionClosing(long sessionId) {
        if (LOG.isTraceEnabled()) {
            LOG.info("Session closing: 0x" + Long.toHexString(sessionId));
        }
        SessionImpl s = sessionsById.get(sessionId);
        if (s == null) {
            return;
        }
        s.isClosing = true;
    }

    public void expire(Session session) {
        long sessionId = session.getSessionId();
        LOG.info("Expiring session 0x" + Long.toHexString(sessionId)
                + ", timeout of " + session.getTimeout() + "ms exceeded");
        close(sessionId);
    }

    private void close(long sessionId) {
        submitRequest(null, sessionId, OpCode.closeSession, 0, null, null);
    }

    

   

PrepRequestProcessor.pRequest2Txn(int type, long zxid, Request request, Record record, boolean deserialize)
case OpCode.closeSession:
                // We don't want to do this check since the session expiration thread
                // queues up this operation without being the session owner.
                // this request is the last of the session so it should be ok
                //zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
                HashSet<String> es = zks.getZKDatabase()
                        .getEphemerals(request.sessionId);
                synchronized (zks.outstandingChanges) {
                    for (ChangeRecord c : zks.outstandingChanges) {
                        if (c.stat == null) {
                            // Doing a delete
                            es.remove(c.path);
                        } else if (c.stat.getEphemeralOwner() == request.sessionId) {
                            es.add(c.path);
                        }
                    }
                    for (String path2Delete : es) {
                        addChangeRecord(new ChangeRecord(request.hdr.getZxid(),
                                path2Delete, null, 0, null));
                    }

                    zks.sessionTracker.setSessionClosing(request.sessionId);
                }

                LOG.info("Processed session termination for sessionid: 0x"
                        + Long.toHexString(request.sessionId));
                break;

 

 

 

FinalRequestProcessor.processRequest(Request request) 
       if (request.hdr != null && request.hdr.getType() == OpCode.closeSession) {
            ServerCnxnFactory scxn = zks.getServerCnxnFactory();
            // this might be possible since
            // we might just be playing diffs from the leader
            if (scxn != null && request.cnxn == null) {
                // calling this if we have the cnxn results in the client's
                // close session response being lost - we've already closed
                // the session/socket here before we can send the closeSession
                // in the switch block below
                scxn.closeSession(request.sessionId);
                return;
            }
        }
 

 

  • Zookeeper Session机制
            
    
    博客分类: Zookeeper  
  • 大小: 342 KB
  • Zookeeper Session机制
            
    
    博客分类: Zookeeper  
  • 大小: 333.3 KB