[ZooKeeper]分布式Session超时

前面分析知道session超时由leader负责，假设某个session长时间没心跳超时，SessionTrackImpl入口

if (set != null) {
                   for (SessionImpl s : set.sessions) {
                       setSessionClosing(s.sessionId);
                       expirer.expire(s);
                   }
               }

 session失效是一个proposal过程

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

 leader段执行处理链，PrepRequestProcessor执行，closeSession的request没有txn内容

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());
        //session下的所有临时节点，需要删除，添加到outstandingChanges队列
                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));
                    }
            //设置状态为closing
                    zks.sessionTracker.setSessionClosing(request.sessionId);
                }

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

 之后leader通过ProposalRequestProcessor发起投票，并写入log，follower处理投票，写入log并ack。然后leader commit请求，leader和follower都进入FinalRequestProcessor处理。先是datatree处理

case OpCode.closeSession:
                   killSession(header.getClientId(), header.getZxid());
                   break;

 

void killSession(long session, long zxid) {
        // the list is already removed from the ephemerals
        // so we do not have to worry about synchronizing on
        // the list. This is only called from FinalRequestProcessor
        // so there is no need for synchronization. The list is not
        // changed here. Only create and delete change the list which
        // are again called from FinalRequestProcessor in session下所有临时节点删除.
    //sequence
        HashSet<String> list = ephemerals.remove(session);
        if (list != null) {
            for (String path : list) {
                try {
                    deleteNode(path, zxid);
                    if (LOG.isDebugEnabled()) {
                        LOG
                                .debug("Deleting ephemeral node " + path
                                        + " for session 0x"
                                        + Long.toHexString(session));
                    }
                } catch (NoNodeException e) {
                    LOG.warn("Ignoring NoNodeException for path " + path
                            + " while removing ephemeral for dead session 0x"
                            + Long.toHexString(session));
                }
            }
        }
    }

 然后FinalRequestProcessor删除session

else if (opCode == OpCode.closeSession) {
            sessionTracker.removeSession(sessionId);
        }

 然后FinalRequestProcessor关闭连接

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;
            }
        }

 closeSession就是把session从内存数据结构中删除。

遍历连接集合，关闭sessionid对应的那个连接

private void closeSessionWithoutWakeup(long sessionId) {
     HashSet<NIOServerCnxn> cnxns;
     synchronized (this.cnxns) {
         cnxns = (HashSet<NIOServerCnxn>)this.cnxns.clone();
     }

     for (NIOServerCnxn cnxn : cnxns) {
         if (cnxn.getSessionId() == sessionId) {
             try {
                 cnxn.close();
             } catch (Exception e) {
                 LOG.warn("exception during session close", e);
             }
             break;
         }
     }
 }

 以上就完成了server端session超时的处理
1.leader发起closeSession的Proposal
2.该session对应的server的FinalRequestProcessor中关闭连接
3.server端发起的session超时处理不需要返回client信息，如果是client主动关闭session，则需要返回closeConn的信息，然后再关闭连接