xylz,imxylz

关注后端架构、中间件、分布式和并发编程

   :: 首页 :: 新随笔 :: 联系 :: 聚合  :: 管理 ::
  111 随笔 :: 10 文章 :: 2680 评论 :: 0 Trackbacks

Session连接

Zookeeper客户端和服务端维持一个长连接,每隔10s向服务端发送一个心跳,服务端返回客户端一个响应。这就是一个Session连接,拥有全局唯一的session id。Session连接通常是一直有效,如果因为网络原因断开了连接,客户端会使用相同的session id进行重连。由于服务端保留了session的各种状态,尤其是各种瞬时节点是否删除依赖于session是否失效。

Session失效问题

通常客户端主动关闭连接认为是一次session失效。另外也有可能因为其它未知原因,例如网络超时导致的session失效问题。在服务端看来,无法区分session失效是何种情况,一次一旦发生session失效,一定时间后就会将session持有的所有watcher以及瞬时节点删除。

而对于Zookeeper客户端而言,一旦发生失效不知道是否该重连,这涉及到watcher和瞬时节点问题,因此Zookeeper客户端认为,一旦发生了seesion失效,那么就认为客户端死掉了。从而所有操作都不能够进行。参考 How should I handle SESSION_EXPIRED?

解决方案

对于只是简单查询服务的客户端而言,session失效后只需要重新建立连接即可。而对于需要处理瞬时节点以及各种watcher的服务来说,应用程序需要处理session失效或者重连带来的副作用。

下面的逻辑提供了一种简单的解决session重连的问题,这是指重新生成新的连接。

public static org.apache.zookeeper.ZooKeeper getZooKeeper() {
  if (zookeeper == null) {
    synchronized (ZookeeperClient.class) {
      if (zookeeper == null) {
        latch = new CountDownLatch(1);
        zookeeper = buildClient();//如果失败,下次还有成功的机会
        long startTime = System.currentTimeMillis();
        try {
           latch.await(30, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
           e.printStackTrace();
        } finally {
          final SocketAddress remoteAddres = zookeeper.testableRemoteSocketAddress();
          System.out.println("[SUC-CORE] local host: " + zookeeper.testableLocalSocketAddress());
          System.out.println("[SUC-CORE] remote host: " + remoteAddres);
          System.out.println("[SUC-CORE] zookeeper session id: " + zookeeper.getSessionId());
          final String remoteHost = remoteAddres != null ? ((InetSocketAddress) remoteAddres).getAddress().getHostAddress() : "";
          System.out.println("[SUC-CORE] init cost: " + (System.currentTimeMillis() - startTime) + "(ms) " + remoteHost);
          latch = null;
        }
      }
    }
  }
  return zookeeper;
}

上面的代码只是简单的使用一个Double-Check来维持Zookeeper客户端的单实例。保证总是有机会重建客户端以及只有一个单例(这是因为Zookeeper客户端是线程安全的)。

例外上面的例子中继承了org.apache.zookeeper.ZooKeeper类,以便能够拿到session对于的服务端ip地址以及客户端地址,方便调试问题。

在构建客户端的时候是需要设置session超时的时间,例如下面的代码就是30秒。


private static ZooKeeper buildClient() {
    final String rootPath = getRootPath();
    final String connectString = SystemConfig.getInstance().getString("zookeeper.ips",//
            "192.168.10.1:2181,192.168.10.2:2181,192.168.10.3:2181,192.168.10.4:2181,192.168.10.5:2181");
    System.out.printf("[SUC-CORE] rootPath: %1s\n", rootPath);
    System.out.printf("[SUC-CORE] connectString:%1s\n", connectString);
    try {
        return new ZooKeeper(connectString + rootPath, 30000, new SessionWatcher());
    } catch (IOException e) {
        throw new RuntimeException("init zookeeper fail.", e);
    }
}

为了处理连接建立成功以及断开问题,我们需要一个Watcher来处理此问题。

static class SessionWatcherimplements Watcher {

    public void process(WatchedEvent event) {
        if (event.getState() == KeeperState.SyncConnected) {
            if (latch != null) {
                latch.countDown();
            }
        } else if (event.getState() == KeeperState.Expired) {
            System.out.println("[SUC-CORE] session expired. now rebuilding");

            //session expired, may be never happending.
            
//close old client and rebuild new client
            close();

            getZooKeeper();
        }
    }
}


一旦检测到session失效了(Expired),那么久销毁已经建立的客户端实例,重新生成一个客户端实例。

/**
 * 关闭zookeeper连接,释放资源
 
*/
public static void close() {
    System.out.println("[SUC-CORE] close");
    if (zookeeper != null) {
        try {
            zookeeper.close();
            zookeeper = null;
        } catch (InterruptedException e) {
            //ignore exception
        }
    }
}

这是一种简单的处理Session expired问题的方法,显然这不会处理瞬时节点的问题,因此如果有相关的需求,业务系统(应用程序)需要自己修复问题。

测试方案

根据Is there an easy way to expire a session for testing? 提供的测试方法,写一个简单的例子试验下。

public static void main(String[] args) throws Exception {
  ZooKeeper zk = ZookeeperClient.getZooKeeper();
  long sessionId = zk.getSessionId();
  //
  final String rootPath = ZookeeperClient.getRootPath();
  final String connectString = SystemConfig.getInstance().getString("zookeeper.ips",//
          "192.168.10.1:2181,192.168.10.2:2181,192.168.10.3:2181,192.168.10.4:2181,192.168.10.5:2181");
  // close the old connection
  new ZooKeeper(connectString + rootPath, 30000, null, sessionId, null).close();
  Thread.sleep(10000L);
  //

  
// rebuild a new session
  long newSessionid = ZookeeperClient.getZooKeeper().getSessionId();

  // check the new session
  String status = newSessionid != sessionId ? "OK" : "FAIL";
  System.out.println(format("%s --> %s %s", sessionId, newSessionid, status));

  // close the client
  ZookeeperClient.getZooKeeper().close();
}

最后能够自动处理session expired的问题。实验中看到确实执行了session expired的逻辑重新生成了新的session。



©2009-2014 IMXYLZ |求贤若渴
posted on 2011-12-05 13:57 imxylz 阅读(28541) 评论(8)  编辑  收藏 所属分类: J2EE技术

评论

# re: 处理Zookeeper的session过期问题 2011-12-06 11:03 TBW
xiexie !这个东西我还是要学的  回复  更多评论
  

# re: 处理Zookeeper的session过期问题 2011-12-07 17:09 雪地靴
源码好复杂,转载下。  回复  更多评论
  

# re: 处理Zookeeper的session过期问题 2011-12-15 00:31 Tai
艳兄永远活在我们心中  回复  更多评论
  

# re: 处理Zookeeper的session过期问题 2012-09-24 16:54 凌波微步
每隔10s向服务端发送一个心跳
说的不对!  回复  更多评论
  

# re: 处理Zookeeper的session过期问题[未登录] 2014-04-11 09:36 XX
@凌波微步
是的,这里不是固定10S。  回复  更多评论
  

# re: 处理Zookeeper的session过期问题[未登录] 2014-05-09 16:31 w
学习学习  回复  更多评论
  

# re: 处理Zookeeper的session过期问题 2014-08-07 18:52 iteblog
写的不错,但是有个地方错误:每隔10s向服务端发送一个心跳
这是不对的。多久向Zookeeper Server发送心跳和设置的Session有关。  回复  更多评论
  

# re: 处理Zookeeper的session过期问题 2016-06-17 03:02 codewarrior
"而对于Zookeeper客户端而言,一旦发生失效不知道是否该重连",这句理解不对吧,Zookeeper wiki上明确写了,当客户端断开后,客户端无法得到SESSION_EXPIRED通知,因为链接断了怎么发送消息?而自动重连是ZK client library自己进行的。换言之客户端是无法知道自己是否失效的,SESSION_EXPIRED通知只有当重连成功后才能收到。  回复  更多评论
  


只有注册用户登录后才能发表评论。


网站导航:
 

©2009-2014 IMXYLZ