Zookeeper-Session控制原理与异常详解

Zookeeper构造器：

• Zookeeper(connectString,sessionTimeout,watcher，canBeReadOnly)
• Zookeeper(connectString,sessionTimeout,watcher,sessionId,byte[] sessionPwd，canBeReadOnly),

    上述是创建ZK实例的2个构造方法，需要指定connectionString，sessionTimeout，canBeReadOnly是表示是否允许client链接只读节点，当client找不到R/W server时，是否允许链接那写只读server(OBserver),只读server只能提供read请求，任何write请求都会被拒绝，不过此时client将会开启后台线程继续寻找R/W server。

    创建ZK实例的过程，也就是Session创建的过程，Session的创建是异步的，根据传递的connectString解析出来的server，选择其中一个连接（列表将会被“洗牌”之后），如果被选择的server不能建立连接，将会继续选择下一个，直到找到可用server，或者Session超时。

    对于Session重连接需要把sessionId和sessionPWD交付给zk构造器，需要注意这两个构造器的一个小差异，clientCnxn实例(Client端维护的和Follower建立的连接类)中有个seenRwServerBefore属性，对于新建链接，此属性为false，对于重建连接此属性为true，表示“过去”是否和R/W server通信过（Leader）,对于每个client连接server，如果当前连接的是readOnly服务器，那么client间歇性的去pingRwServer(),【SendThread.run()】，如果此时发现了rwServer将会抛出RwServerFoundException，导致client与此RwServer重新连接（DISTINCTED状态消息）.如果client连接到 rwServer（在server返回的结果中得知连接到了rw server），则会把seenRwServerBefore置位，并获取sessionId。SeenRwServerBeffore一旦置位即不会再改变；sessionId默认为0，在找到R/W server之前（以及新建连接时）用0代替其sessionId。

    ReadOnly服务器,就是此server不能接收"write"操作,比如Observer或者处于Looking状态的Follower;此时Client只能和此Server进行read操作.其中read操作包括:getData,exist等.

    rwServer,就是状态正常的Follower或者Leader,表示此server可以接收客户端的write操作.

 

    ZookeeperServer中processConnectRequest()方法是处理client链接，如果client传递的sessionTimeout不在max和min之间，将会被server会截取为max或者min。检测sessionId，如果client连接请求的sessionId不为0，则serverCnxnFactory（连接管理工厂，持有所有session的连接句柄）移除旧的连接，并重新设置连接信息。如果sessionId为0，则生成新的session，向Leader提交createSession操作。每个session对象(SessionImpl)持有sessionId/timeout/tickTime属性，其中sessionId是一个有server维护的序列化（自增，8字节）的数字，sessionId有Leader生成创建代码如下:

nextSid = (System.currentTimeMillis() << 24) >> 8;
nextSid =  nextSid | (id <<56);

 

其中id为server id，每个server都会在myid中注册自己的id，由此可见尽管sessionId也不会重复。

 

    关于Session跟踪，ZK server有一个SessionTracker（Leader有SessionTrackerImpl，Follower有LearnerSessionTracker），SessionTrackerImpl端维护如下3个Map结构，Session创建后相关数据分别放入这三个Map中：

• Map<Long[sessionId],Session>sessionsById
• Map<Long[sessionId],Integer>sessionsWithTimeout
• Map<Long[tickTime], SessionSet> sessionSets

    其中sessionsById简单用来存放Session对象及校验sessionId是否过期。

    sessionsWithTimeout用来维护，session的持久化：数据会写入snapshot，在Server重启时会从snapshot恢复到sessionsWithTimeout，从而能 够维持跨重启的session状态。它的key为sessionId，value为client传递的sessionTimeout。

 

    LearnerSessionTracker维持了2个Map结构：

• HashMap<Long, Integer> touchTable = new HashMap<Long, Integer>();
• ConcurrentHashMap<Long, Integer> sessionsWithTimeouts;

     touchTable维持了连接此server的session列表，sessionWithTimeouts维持了所有session，用于snapshot和session恢复。LearnerSessionTracker是一个shell（壳），接受creatSession/closeSession时对上述2个map做调整，并在ping时把本地活跃的session(即touchTable)发送leader。

 

    Session对象的tickTime属性表示session的过期时间，新建的session此值为0。sessionSets这个Map会以过期时间为key，将所有过期时间 相同的session收集为一个集合。Server每次接到Client的一个请求或者心跳时，会根据当前时间和其sessionTimeout重新计算 过期时间并更新Session对象和sessionSets。计算出的过期时间点会向上取整为ZKServer的属性tickTime的整数倍。 Server启动时会启动一个独立的线程负责将大于当前时间的所有tickTime对应的Session全部清除关闭。SessionTrackerImpl是一个线程(Leader运行)，在run()方法中，轮询执行，每次都重新计算sessionSet的过期时间（nextExpirationTime += expirationInterval）,如果 nextExpirationTime比当前时间小，就等待(等待指定时间之后,扫描--->再等待-->)，到期后开始删除此时间点相应的session集合，同时指示zookeeper.expire向Leader发送closeSession请求。NIOServerCnxn是server端处理请求的起始类，最终的请求将会有相应的ZK Server处理(ZookeeperServer,FollowerZooKeeperServer,LeaderZooKeeperServer，ObserverZooKeeperServer，ReadOnlyZooKeeperServer)，在处理请求数据时（包括ping），将会出发sessionTracker.touchSession(id,to),此方法就是负责“延续”session过期处理的，计算过期的代码(time / expirationInterval + 1) * expirationInterval。

 

    对于Client和Server建立连接,就会触发Session的创建,Follower将createSession请求交付给Leader,Leader收到请求后，会发起一个createSession的Proposal，如果表决成功(多数派)，最终所有的Server都会在其内存中建立 同样的Session。ZookeeperServer.processTxn(),可知在createSession/closeSession类型时，会对本地zkdatabase进行处理(因为Session信息会被持久存出在ZKDatabase中)。等表决通过后，与客户端建立连接的Server为这个session生成一个password，连同 sessionId，sessionTimeOut一起返回给客户端（ConnectResponse）。任何一个Session只能被一个Server所服务,Leader会保留每个Session被哪个Server所持有.

 

    Follower.processPacket()是处理Folower与Leader之间的各种通信的（Leader发给Follower），其中有个ping，由此可见，在Follower向Leader发送ping消息时，同时会把本地session集合（LearnerSessionTracker中的touchTable）发送给Leader，在LearnerHandler线程中run方法轮询处理Follower发送给Leader的消息，在对待ping类型消息时读取follower发送的session集合，依次遍历，执行touch方法（即sessionTrackerImpl的touch方法，目的是“延续”session过期时间），不过同时需要注意Leader.lead()方法会向Learner发送ping，不过是空的packet。由此可见，session的状态，完全有leader保持，follower只是定期通过ping把自己维护的session信息包括过期时间发送给leader，leader决定是否过期，以及发送closeSession提议。

    Leader会周期性的检测全局Session列表,是否有过期的,如果有,将会向所有的Follower发送cloaseSession提议,Follower在接收到提议后,将Session删除.

    客户端如果需要重连Server，可以新建一个 ZooKeeper实例，将上一个成功连接的ZooKeeper对象的sessionId和password传给Server:

• ZooKeeper zk = new ZooKeeper(serverList, sessionTimeout, watcher, sessionId,passwd):ZKServer会根据sessionId和password为同一个client恢复session，如果还没有过期的话。

 

SessionMovedExcepion触发机制:

1. 当client与server1的链接失效,client和server2重建链接.
2. client发送给server1的请求,因为网络问题,延迟到达server1,且在到达server1之前,client与server2的connect请求已到达server2.
3. server2对于这种重连操作(称之为sessionReopen),将先关闭本地持有的此session的链接(serverCnxn),server2将当前session.owner设置为自己,并发送一个Leader.REVALIDATE类型的请求给leader.参见代码(SessionTracker.setOwner(),LearnerZookeeperServer.revalidateSession(),ZookeeperServer.processConnectRequest()--->reopenSession(...))如果session有效,将会触发reopenSession操作,此操作将会导致想leader发送一个REVALIDATE类型的请求.
4. leader收到此请求后,校验session是否有效(zk.touch(sessionId,timeout)此方法校验session,如果有效导致session过期时间延续),如果有效,leader设置session的owner属性(即此session在哪个server上被建立链接),如果session无效,返回valid标识位.无论有效与否,最后把相应packet加入到广播队列(queuedPackets ),leader将会把此packet依次发送给所有的follower(Leader.sendPacket()/LeaderHander.run())
5. server2收到leader的revalidate消息(Learner.processPacket(..)),如果发现valid为无效,则导致server2的链接关闭,直接导致client与server2的重连操作失败.如果valid有效,server2就认为此时此Session合法且属于自己维护.
6. server1也是如此,在接收到client发送的请求后,校验session时发现的owner标识和request.getOwner(){:::request为server的数据封装,此owner为ServerCnxn的关联对象}不同,将会导致SessionMovedException.
7. SessionMovedException不会对请求不会带来干扰,只是导致server1关闭此session上关联的链接(ServerCnxn,标示一个来自client的链接),SessionMovedException对客户端也不可见,也不影响操作,事实上在此异常发生之前,client交付的操作仍会被执行.
8. 每个ServerCnxn都会实例化一个owner,此owner仅仅是一个new Object().

SessionExpiredException异常(和EXPIRED事件):

    这个异常API有2个,一个是KeeperException(Server端抛出),另一个是IOException(客户端抛出),只所以有2个,可能基于设计的考虑,不过最终想表达的意义是一样的.

    通常是zk客户端与服务器的连接断了，试图连接上新的zk机器，这个过程如果耗时过长，超过 SESSION_TIMEOUT 后还没有成功连接上服务器，那么服务器认为这个session已经结束了（服务器无法确认是因为其它异常原因还是客户端主动结束会话），开始清除和这个会话有关的信息，包括这个会话创建的临时节点和注册的Watcher(参见DataTree.killSession(),清除临时节点)。在这之后，客户端重新连接上了服务器在，但是很不幸，服务器会告诉客户端 SESSIONEXPIRED(发生在sessionTracker.checkSession(),如果找不到session,将会导致sessionExpired)。此时客户端要做的事情就看应用的复杂情况了，总之，要重新实例zookeeper对象，重新操作所有临时数据（包括临时节点和注册Watcher）。

    在PrepRequestProcessor.pRequest(request)中,对于几乎所有的操作都会进行session过期检测,所以对于过期的session是不能做任何有效操作的.KeeperException是server端抛出的异常,此异常将会被封装在响应中(ReplyHeader,err属性),客户端在读取响应时,会检测err内容,并以异常的方式抛出,所以客户端(zookeeper)需要捕获异常(所有的zk客户端异常都继承自IOException,参见ClientCnxn).

    对于EXPIRED事件,则是在zk客户端重连接server时,server发现session已经不存在,则会重置此次链接上的session timeout参数(是否还记得,链接时客户端向server发送自己的session过期最大/最小时间;如果时间不符合server配置,将会被重置为server可接受的值,并通过链接响应反馈给client;那么也就是在此步骤,server在发现session过期时,将会重置session timeout时间为0即立即过期),如果client发现server返回的timeout时间为0,则发布一个本地的Expired watchEvent.(参见代码zookeeperSrever.finishSessionInit(),ClientCnxn.SendThread.onConnected()).

 

SessionTimeoutException异常(即CONNECTION_LOSS):

    这个异常属于链接异常,由Client端抛出(此异常继承自IOException);在zk客户端socket链接控制器上(ClientCnxnSocket),会记录now(链接创建时间)/lastSend(请求最后发送时间)/lastHeard(心跳最后时间),这三个变量用来跟踪链接是否超时以及发送心跳的时机.

connectTimeout = sessionTimeout / hostProvider.size();//链接超时时间
readTimeout = sessionTimeout * 2 / 3;//读取响应超时时间

 

    SendThread会不断轮询执行(执行发送请求packet,读取响应,发送心跳等活动),当发现readTimeout/2 - clientCnxnSocket.getIdleSend()的计算结果<=0时,触发ping操作,其中idleSend为(now-lastHeard).因为client的链接状态有2种:

1)链接可用(isConnected):当发现readTimeout - clientCnxnSocket.getIdleRecv()的计算结果<=0时

2)链接不可用(connecting):当发现connectTimeout - clientCnxnSocket.getIdleRecv()的计算结果<=0时

 

    如果连接等待超时,将会导致SessionTimeout异常被抛出,此异常会被内部捕获,并导致client链接处于Disconnect状态,并触发本地Disconnected事件;此异常并不需要客户端明确的去捕获或者进行其他措施,Client会自动重连,直到SessionExpired.此异常不会直接导致session过期,也不会清除当前client上注册的event集合.但是在尚未重连成功之前,server响应的event会丢失,而且尚未处理的响应(包括等待发送的queue,已经发送尚未接受结果的queue),也会被清除.所以如果想确定此前的操作是否成功需要手动去检测.

 

ConnectionLossException异常:因为ZK Client请求是被队列化的,这个队列化控制有每个客户端控制,如果Client在处理Server的响应结果时发现顺序有错乱,比如当前队列头部待确认的请求和Server交付的响应不是一个(通过xid做比较),那么Client认为在请求操作过程中,可能存在数据丢失的情况,将会触发此异常的发生;此异常会直接导致Client连接被关闭,重新建立连接.

 

客户端修改操作,其他客户端是否立即可见?

    ZK服务不保证,任何更新都能够实时的被客户端所感知,但是server中的data view是一致的.数据的更新会通过消息(假如注册watch)通知感兴趣的client,消息的发送和接受也无可能同时到达.如果想得知当前时刻最新的server数据,可以通过调用sync().

 

F1

1. 当Client与Server建立连接后,Server会根据client传递的sessionTimeout时间做一次计算,比如它不能超过server配置的最大/最小过期时间等.并发回给client一个合理的值.对于重连接操作,client会交付给server现有的sessionId,如果此session已经过期,则返回一个负值.client可以根据此值判断出Session是否已经Expired.
2. ..
3. readOnly参数有客户端指定, 即canReadOnly参数,如果为true则表示当前Client允许和readOnly服务器建立连接,当server处于选举过程中或者"危险期"("少数派")是,server处于readOnly状态;server会把自己的状态在建立连接之后的首个响应包中返回给client;如果此参数无false,则表示client不允许与readOnly服务器建立连接,如果不幸与readOnly服务器建立了连接,则直接导致连接被关闭,client需要重新选择服务器.
4. 如果连接了ReadOnly服务器,此时client的"连接"状态为CONNECTEDREADONLY,否则为CONNECTED;此后可以通过client进行正常的读写操作;如果在readOnly模式下,发送write操作(比如create,setData)等,那么server将会校验请求类型,极有可能在响应结果中告知一个error,此时客户端接收到error后会抛出异常,对于API调用者,则需要捕获并处理.
5. 如果Client连接了一个readOnly服务器,那么在SendThread中会开启一个处理分支,并周期性的检测服务器列表,依次建立连接并查找出RWServer,一旦查到,将会以异常的方式close当前连接,并和RWServer重连.
6. 对于Zookeeper客户端实例,所有的请求都将队列化,首先将请求加入队列,并阻塞当前正在处理的packet,packet为客户端请求的封装对象,阻塞的方式为packet.wait..当server端返回了响应结果,那么导致packet.notifyAll();一个奇怪的问题就是,在client重连操作或者关闭连接时,都将导致请求队列被clean,那么对于API 调用者,则会收到异常,那么调用者需要重新尝试操作.
7. sessionTimeout的值有Server最终决定,Client通过此值用来判断SessionTimeoutException触发的时机..在Client与server交互期间,可能因为IO的阻塞,导致client等待的时间超过限制,最终客户端触发SessionTimeoutException,此异常并非致命,但是client会"觉得"sever可能不可用,则尝试和其他server重新连接.(参考6)).
8. Client为了让Server感知到Session的活性,则周期性的向Server发送ping消息,此周期时间比SessionTimeout值要小.
9. 如果Client重连接到Server之后,server校验发现此Session信息已经不存在(只针对重连),那么Server将会在响应包中告知此异常,此异常直接导致Client和Server的连接被关闭(SendThread状态被置为不可用);此后Client实例将不可用,除非新建Zookeeper实例.

1.  Client是否首次连接,主要基于sessionId的判断,对于首次连接(新会话),那么sessionId为0,对于重连接操作,sessionId基于现有的.
2. createSession过程有Leader操作,Follower只需要把此信息转发给Leader即可.leader根据目前现有的最大SessionId + 1,作为新会话的sessionId,并将信息保存在sessionsWithTimeout,sessionsById,sessionSets这三个数据结构中(参见上述);其中sessionsWithTimeout是ZKDatabase的一部分,如果createSession提议被集群确认,那么session信息将会被持久存在所有server的ZKDatabase中.
3. createSession创建成功后,Follower即可以想Client反馈结果,并交付sessionId,同时表示和Client之间的连接即可以处理正常的业务.
4. 对于reopenSession稍微复杂,多了一步session的过期校验,校验的手段非常简单,就是从当前的session列表中查找,如果sessio还存在,则表示未过期..不过此过程仍然有Leader控制.
5. LeaderZookeeper类使用了SessionTracker来跟踪session,此tracker是一个线程,周期性的检测session过期情况.对于createSession或者reopenSession会导致sessionsWithTimeout,sessionsById,sessionSets三个数据结构中session信息的添加,其中sessionSets是个Map(参见上述),key为tickTime,这个值为session下一次过期的时间戳(System.currentTimeMillis() + timeout,并对"expirationInterval"取整).其主要目的就是把所有的session按照即将过期的时间梯度分类,并递增的逐次去检测过期.
6. Follower/Observer也持有了一个LearnerSessionTracker,这个tracker和Leader持有的SessionTracker有却别,它不是一个线程,它不周期性的检测session过期.它只负责当client请求时,更新session的过期时间.并且在Follower/Observer与Leader的ping消息中,将在自己server上活跃(与当前server建立连接的Session)的session列表发送给Leader;Leader被动的根据此session列表来延迟Session过期操作;说白了,session的活跃性与否是Follower知道,但是Follower需要告诉Leader哪些是活跃的,最终有Leader来检测过期与否.
7. 对于Leader接收到Follower/Observer的活跃session列表之后,将会操作SessionTracker中的sessionSets,并将活跃的session从此当前过期梯度的sessionSets中移除,因此在SessionTracker的下一次过期检测时就不会得到它们..(如果session活跃,那么tracker将会把此session放在当前过期梯度的下一个梯度中,每个梯度的时间差为一个"expirationInterval").
8. 对于Leader : SessionTrackerImpl + LeaderZookeeperServer;对于Follower: LearnerSessionTracker + LearnerZookeeperServer

 关于ZK Session机制中还有很多疑点,路很长,我只能帮到这儿了..

评论

1 楼 爱岩修 2013-11-22  

这是分析源码啊，牛X