从ZooKeeper源代码看如何实现分布式系统（四）session管理

这篇看看ZooKeeper如何管理Session。 Session相关的接口如下： 

Session: 表示session的实体类,维护sessionId和timeout两个主要状态

SessionTracker： Session生命周期管理相关的操作

SessionExpier： Session过期的操作

 

先看看Session接口和它的实现类SessionImpl，维护了5个属性：sessionId， timeout表示超时时间，tickTime表示客户端和服务器的心跳时间，isClosing表示是否关闭，owner表示对应的客户端

 

public static interface Session {
        long getSessionId();
        int getTimeout();
        boolean isClosing();
    }

public static class SessionImpl implements Session {
        SessionImpl(long sessionId, int timeout, long expireTime) {
            this.sessionId = sessionId;
            this.timeout = timeout;
            this.tickTime = expireTime;
            isClosing = false;
        }

        final long sessionId;
        final int timeout;
        long tickTime;
        boolean isClosing;

        Object owner;

        public long getSessionId() { return sessionId; }
        public int getTimeout() { return timeout; }
        public boolean isClosing() { return isClosing; }
    }

 

SessionTracker的实现类是SessionTrackerImpl，它是一个单独运行的线程，根据tick周期来批量检查处理当前的session。SessionTrackerImpl直接继承了Thread类，它的定义和构造函数如下，几个主要的属性：

expirer是SessionExpirer的实现类

expirationInterval表示过期的周期，可以看到它的值是tickTime，即如果服务器端在tickTime里面没有收到客户端的心跳，就认为该session过期了

sessionsWithTimeout是一个ConcurrentHashMap，维护了一组sessionId和它对应的timeout过期时间

nextExpirationTime表示下次过期时间，线程会在nextExpirationTime时间来批量过期session

nextSessionId是根据sid计算出的下一个新建的sessionId

sessionById这个HashMap保存了sessionId和Session对象的映射

sessionSets这个HashMap保存了一个过期时间和一组保存在SessionSet中的Session的映射，用来批量清理过期的Session

 

public interface SessionTracker {
    public static interface Session {
        long getSessionId();
        int getTimeout();
        boolean isClosing();
    }
    public static interface SessionExpirer {
        void expire(Session session);

        long getServerId();
    }

    long createSession(int sessionTimeout);

    void addSession(long id, int to);

    boolean touchSession(long sessionId, int sessionTimeout);

    void setSessionClosing(long sessionId);

    void shutdown();

    void removeSession(long sessionId);

    void checkSession(long sessionId, Object owner) throws KeeperException.SessionExpiredException, SessionMovedException;

    void setOwner(long id, Object owner) throws SessionExpiredException;

    void dumpSessions(PrintWriter pwriter);
}

public class SessionTrackerImpl extends Thread implements SessionTracker {
    private static final Logger LOG = LoggerFactory.getLogger(SessionTrackerImpl.class);

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

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

    ConcurrentHashMap<Long, Integer> sessionsWithTimeout;
    long nextSessionId = 0;
    long nextExpirationTime;

    int expirationInterval;

public SessionTrackerImpl(SessionExpirer expirer,
            ConcurrentHashMap<Long, Integer> sessionsWithTimeout, int tickTime,
            long sid)
    {
        super("SessionTracker");
        this.expirer = expirer;
        this.expirationInterval = tickTime;
        this.sessionsWithTimeout = sessionsWithTimeout;
        nextExpirationTime = roundToInterval(System.currentTimeMillis());
        this.nextSessionId = initializeNextSession(sid);
        for (Entry<Long, Integer> e : sessionsWithTimeout.entrySet()) {
            addSession(e.getKey(), e.getValue());
        }
    }

 

 

看一下SessionTrackerImpl这个线程的run方法实现，实现了批量处理过期Session的逻辑

1. 如果下一次过期时间nextExpirationTime大于当前时间，那么当前线程等待nextExpirationTime - currentTime时间

2. 如果到了过期时间，就从sessionSets里面把当前过期时间对应的一组SessionSet取出

3. 批量关闭和过期这组session

4. 把当前过期时间nextExpirationTime 加上 expirationInterval作为下一个过期时间nextExpiration，继续循环

 

其中expirer.expire(s)这个操作，这里的expirer的实现类是ZooKeeperServer，它的expire方法会给给客户端发送session关闭的请求

 

// SessionTrackerImpl
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!");
    }

// ZookeeperServer
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);
    }

再看一下创建Session的过程

 

1. createSession方法只需要一个sessionTimeout参数来指定Session的过期时间，会把当前全局的nextSessionId作为sessionId传给addSession方法

2. addSession方法先把sessionId和过期时间的映射添加到sessionsWithTimeout这个Map里面来，如果在sessionById这个Map里面没有找到对应sessionId的session对象，就创建一个Session对象，然后放到sessionById Map里面。最后调用touchSession方法来设置session的过期时间等信息

3. touchSession方法首先判断session状态，如果关闭就返回。计算当前session的过期时间，如果是第一次touch这个session，它的tickTime会被设置成它的过期时间expireTime,然后把它加到对应的sessuibSets里面。如果不是第一次touch，那么它的tickTime会是它当前的过期时间，如果还没过期，就返回。如果过期了，就重新计算一个过期时间，并设置给tickTime，然后从对应的sessionSets里面先移出，再加入到新的sessionSets里面。 touchSession方法主要是为了更新session的过期时间。

 

synchronized public long createSession(int sessionTimeout) {
        addSession(nextSessionId, sessionTimeout);
        return nextSessionId++;
    }

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

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

SessionTracker这个接口主要被ZooKeeperServer这个类来使用，ZooKeeperServer表示ZooKeeper的服务器类，负责维护ZooKeeper服务器状态。

 

在ZooKeeperServer的startup方法中，如果sessionTracker对象为空，就先创建一个SessionTracker对象，然后调用startSessionTracker方法启动SessionTrackerImpl这个线程

 

 public void startup() {
        if (sessionTracker == null) {
            createSessionTracker();
        }
        startSessionTracker();
        setupRequestProcessors();

        registerJMX();

        synchronized (this) {
            running = true;
            notifyAll();
        }
    }

 protected void createSessionTracker() {
        sessionTracker = new SessionTrackerImpl(this, zkDb.getSessionWithTimeOuts(),
                tickTime, 1);
    }

protected void startSessionTracker() {
        ((SessionTrackerImpl)sessionTracker).start();
    }

在ZooKeeperServer的shutdown方法中，调用sessionTracker的shutdown方法来关闭sessionTrackerImpl线程

 

 

 public void shutdown() {
        LOG.info("shutting down");

        // new RuntimeException("Calling shutdown").printStackTrace();
        this.running = false;
        // Since sessionTracker and syncThreads poll we just have to
        // set running to false and they will detect it during the poll
        // interval.
        if (sessionTracker != null) {
            sessionTracker.shutdown();
        }
        if (firstProcessor != null) {
            firstProcessor.shutdown();
        }
        if (zkDb != null) {
            zkDb.clear();
        }

        unregisterJMX();
    }

// SessionTrackerImpl
public void shutdown() {
        LOG.info("Shutting down");

        running = false;
        if (LOG.isTraceEnabled()) {
            ZooTrace.logTraceMessage(LOG, ZooTrace.getTextTraceLevel(),
                                     "Shutdown SessionTrackerImpl!");
        }
    }

ZooKeeperServer的createSession方法给连接ServerCnxn创建一个对应的session，然后给客户端发送一个创建了session的请求

 

 

long createSession(ServerCnxn cnxn, byte passwd[], int timeout) {
        long sessionId = sessionTracker.createSession(timeout);
        Random r = new Random(sessionId ^ superSecret);
        r.nextBytes(passwd);
        ByteBuffer to = ByteBuffer.allocate(4);
        to.putInt(timeout);
        cnxn.setSessionId(sessionId);
        submitRequest(cnxn, sessionId, OpCode.createSession, 0, to, null);
        return sessionId;
    }

ZooKeeperServer的reopenSession会给断开了连接后又重新连接的session更新状态，使session继续可用

 

1. 如果session的密码不对，调用finishSessionInit方法来关闭session，如果密码正确，调用revalidateSession方法

2. revalidateSession方法会调用sessionTracker的touchSession，如果session已经过期，rc = false，如果session未过期，更新session的过期时间信息。最后也调用finishSessionInit方法

3. finishSessionInit方法会给客户端发送响应对象ConnectResponse，如果验证不通过，会关闭连接  cnxn.sendBuffer(ServerCnxnFactory.closeConn)。验证通过，调用cnxn.enableRecv(); 方法来设置连接状态，使服务器端连接注册SelectionKey.OP_READ事件，准备接收客户端请求

 

public void reopenSession(ServerCnxn cnxn, long sessionId, byte[] passwd,
            int sessionTimeout) throws IOException {
        if (!checkPasswd(sessionId, passwd)) {
            finishSessionInit(cnxn, false);
        } else {
            revalidateSession(cnxn, sessionId, sessionTimeout);
        }
    }

protected void revalidateSession(ServerCnxn cnxn, long sessionId,
            int sessionTimeout) throws IOException {
        boolean rc = sessionTracker.touchSession(sessionId, sessionTimeout);
        if (LOG.isTraceEnabled()) {
            ZooTrace.logTraceMessage(LOG,ZooTrace.SESSION_TRACE_MASK,
                                     "Session 0x" + Long.toHexString(sessionId) +
                    " is valid: " + rc);
        }
        finishSessionInit(cnxn, rc);
    }

public void finishSessionInit(ServerCnxn cnxn, boolean valid) {
        // register with JMX
        try {
            if (valid) {
                serverCnxnFactory.registerConnection(cnxn);
            }
        } catch (Exception e) {
                LOG.warn("Failed to register with JMX", e);
        }

        try {
            ConnectResponse rsp = new ConnectResponse(0, valid ? cnxn.getSessionTimeout()
                    : 0, valid ? cnxn.getSessionId() : 0, // send 0 if session is no
                            // longer valid
                            valid ? generatePasswd(cnxn.getSessionId()) : new byte[16]);
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            BinaryOutputArchive bos = BinaryOutputArchive.getArchive(baos);
            bos.writeInt(-1, "len");
            rsp.serialize(bos, "connect");
            if (!cnxn.isOldClient) {
                bos.writeBool(
                        this instanceof ReadOnlyZooKeeperServer, "readOnly");
            }
            baos.close();
            ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());
            bb.putInt(bb.remaining() - 4).rewind();
            cnxn.sendBuffer(bb);

            if (!valid) {
                LOG.info("Invalid session 0x"
                        + Long.toHexString(cnxn.getSessionId())
                        + " for client "
                        + cnxn.getRemoteSocketAddress()
                        + ", probably expired");
                cnxn.sendBuffer(ServerCnxnFactory.closeConn);
            } else {
                LOG.info("Established session 0x"
                        + Long.toHexString(cnxn.getSessionId())
                        + " with negotiated timeout " + cnxn.getSessionTimeout()
                        + " for client "
                        + cnxn.getRemoteSocketAddress());
                cnxn.enableRecv();
            }

        } catch (Exception e) {
            LOG.warn("Exception while establishing session, closing", e);
            cnxn.close();
        }
    }

// NIOServerCnxn
public void enableRecv() {
        synchronized (this.factory) {
            sk.selector().wakeup();
            if (sk.isValid()) {
                int interest = sk.interestOps();
                if ((interest & SelectionKey.OP_READ) == 0) {
                    sk.interestOps(interest | SelectionKey.OP_READ);
                }
            }
        }
    }