手把手带你撸zookeeper源码-zookeeper集群如何接收客户端的连接的

布满荆棘的人生 2022-11-30 01:34 58阅读 0赞

截止到目前为止, zookeeper集群启动时如何进行leader选举以及正常启动、故障恢复时zookeeper如何去恢复内存数据, 如何去leader中同步数据，这两大点都已经分析完了，感兴趣的小伙伴可以去看一下之前的文章 [手把手带你撸zookeeper源码系列目录][zookeeper]

接下里该分析客户端连接集群的代码了，如: 客户端如何和客户端建立的连接、会话是如何创建的、zk集群如何去维护会话的、客户端的创建、删除、修改、查询等操作、持久化节点、临时节点、以及监听通知如何实现的、写请求如何转发到leader上然后leader如何进行2PC数据的提交的等等

今天进入和客户端相关的第一篇文章分析，主要是分析一下zookeeper集群在什么时候对我们配置的clientPort端口进行监听的，然后以什么样的方式来处理客户端的连接的

现在回到我们的zookeeper的源码入口，我们把之前没有深入分析的代码再来看一遍

在QuorumPeerMain.initializeAndRun()方法中QuorumPeerConfig对象中保存了我们配置的zoo.cfg中的变量

protected void initializeAndRun(String[] args)
            throws ConfigException, IOException
        {
            //用于解析配置文件的类
            QuorumPeerConfig config = new QuorumPeerConfig();
            if (args.length == 1) {
                //如果传入了一个参数，则认为是配置文件 zoo.cfg
                config.parse(args[0]);
            }
        }

接下来会调用到runFromConfig(QuorumPeerConfig)方法，我们先看看下面的代码

ServerCnxnFactory cnxnFactory = ServerCnxnFactory.createFactory();
    cnxnFactory.configure(config.getClientPortAddress(),
                                    config.getMaxClientCnxns());

这块第一行代码主要是创建一个服务连接工厂

static public ServerCnxnFactory createFactory() throws IOException {
            String serverCnxnFactoryName =
                System.getProperty(ZOOKEEPER_SERVER_CNXN_FACTORY);
            if (serverCnxnFactoryName == null) {
                serverCnxnFactoryName = NIOServerCnxnFactory.class.getName();
            }
            try {
                ServerCnxnFactory serverCnxnFactory = (ServerCnxnFactory) Class.forName(serverCnxnFactoryName)
                        .getDeclaredConstructor().newInstance();
                LOG.info("Using {} as server connection factory", serverCnxnFactoryName);
                return serverCnxnFactory;
            } catch (Exception e) {
                IOException ioe = new IOException("Couldn't instantiate "
                        + serverCnxnFactoryName);
                ioe.initCause(e);
                throw ioe;
            }
        }

从这块代码分析，会查找系统环境变量中的zookeeper.serverCnxnFactory属性，是否有配置对应的服务连接工厂名(类名)，如果没有则会初始化NIOServerCnxnFactory类名，然后调用Class.forName加载此类, 并通过反射调用进行对象的实例化

接着进入NIOServerCnxnFactory.configure()方法中

@Override
        public void configure(InetSocketAddress addr, int maxcc) throws IOException {
            configureSaslLogin();
    
            thread = new ZooKeeperThread(this, "NIOServerCxn.Factory:" + addr);
            thread.setDaemon(true);
            maxClientCnxns = maxcc;
            this.ss = ServerSocketChannel.open();
            ss.socket().setReuseAddress(true);
            LOG.info("binding to port " + addr);
            ss.socket().bind(addr);
            ss.configureBlocking(false);
            ss.register(selector, SelectionKey.OP_ACCEPT);
        }

在这块代码中，如果对java nio属性的小伙伴，就能看出来关键代码了

this.ss = ServerSocketChannel.open();// 打开一个socketChannel
            ss.socket().setReuseAddress(true);    // 设置socket复用地址
            LOG.info("binding to port " + addr);
            ss.socket().bind(addr);                //绑定地址
            ss.configureBlocking(false);            // 设置为非阻塞
            ss.register(selector, SelectionKey.OP_ACCEPT);    // 注册OP_ACCEPT事件

我们可以看看bind的是哪个地址，传递进来的addr是哪个

可以从QuorumPeerConfig.parse中找到相关代码

if (key.equals("clientPort")) {
                clientPort = Integer.parseInt(value);
            } else if (key.equals("clientPortAddress")) {
                clientPortAddress = value.trim();
            }
    
            if (clientPortAddress != null) {
                this.clientPortAddress = new InetSocketAddress(
                        InetAddress.getByName(clientPortAddress), clientPort);
            } else {
                this.clientPortAddress = new InetSocketAddress(clientPort);
            }

即就是我们在zoo.cfg中配置的clientPort = 2181的端口号

我们知道在上面的代码中只是绑定了一个地址，但是具体如何去接受客户端的连接的呢？在这个地方实例化了一个ZookeeperThread，一看就知道是一个线程，然后设置此线程是一个守护线程(守护线程就是它的执行不会阻挡主线程的退出，主线程如果退出，则守护线程也随着退出)，肯定有某个地方会启动这个线程，并且NIOServerCnxnFactory本身也实现了Runnable接口也是一个线程，我们接着往下看，进入QuorumPeer.start()方法

@Override
        public synchronized void start() {
            //加载快照文件数据到内存中恢复数据
            loadDataBase();
            cnxnFactory.start();
            //启动leader选举
            startLeaderElection();
            //initLeaderElection() 为leader选举做好初始化工作
            super.start();
        }

在这里发现了一行关键代码cnxnFactory.start()，也是我们之前启动时一直没有分析的代码

@Override
        public void start() {
            // ensure thread is started once and only once
            if (thread.getState() == Thread.State.NEW) {
                thread.start();
            }
        }

然后就会把NIOServerCnxnFactory这个线程启动启动线程，接下来我们看看它的run方法

public void run() {
            while (!ss.socket().isClosed()) {
                try {
                    selector.select(1000);
                    Set<SelectionKey> selected;
                    synchronized (this) {
                        selected = selector.selectedKeys();
                    }
                    ArrayList<SelectionKey> selectedList = new ArrayList<SelectionKey>(
                            selected);
                    // 随机接收客户端请求
                    Collections.shuffle(selectedList);
                    for (SelectionKey k : selectedList) {
                        //连接请求处理
                        if ((k.readyOps() & SelectionKey.OP_ACCEPT) != 0) {
                            SocketChannel sc = ((ServerSocketChannel) k
                                    .channel()).accept();
                            InetAddress ia = sc.socket().getInetAddress();
                            int cnxncount = getClientCnxnCount(ia);
                            if (maxClientCnxns > 0 && cnxncount >= maxClientCnxns){
                                LOG.warn("Too many connections from " + ia
                                         + " - max is " + maxClientCnxns );
                                sc.close();
                            } else {
                                LOG.info("Accepted socket connection from "
                                         + sc.socket().getRemoteSocketAddress());
                                sc.configureBlocking(false);
                                SelectionKey sk = sc.register(selector,
                                        SelectionKey.OP_READ);
                                NIOServerCnxn cnxn = createConnection(sc, sk);
                                sk.attach(cnxn);
                                addCnxn(cnxn);
                            }
                        }
                        //处理请求，并返回响应
                        else if ((k.readyOps() & (SelectionKey.OP_READ | SelectionKey.OP_WRITE)) != 0) {
                            NIOServerCnxn c = (NIOServerCnxn) k.attachment();
                            c.doIO(k);
                        } else {
                            if (LOG.isDebugEnabled()) {
                                LOG.debug("Unexpected ops in select "
                                          + k.readyOps());
                            }
                        }
                    }
                    selected.clear();
                } catch (RuntimeException e) {
                    LOG.warn("Ignoring unexpected runtime exception", e);
                } catch (Exception e) {
                    LOG.warn("Ignoring exception", e);
                }
            }
            closeAll();
            LOG.info("NIOServerCnxn factory exited run method");
        }

接着就进入了一个while循环，然后上来就是selector.select(1000); 每隔1s检查一下是否有客户端要进来连接，当有客户端连接进来的时候，就会向下执行

Set<SelectionKey> selected;
                    synchronized (this) {
                        selected = selector.selectedKeys();
                    }
                    ArrayList<SelectionKey> selectedList = new ArrayList<SelectionKey>(
                            selected);

当有客户端连接进来时，可以通过selector.selectedKeys()来获取当前连接的客户端发送的是什么事件的请求

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td>类型</td> 
   <td>说明</td> 
  </tr> 
  <tr> 
   <td>OP_READ</td> 
   <td>读请求</td> 
  </tr> 
  <tr> 
   <td>OP_WRITE</td> 
   <td>写请求</td> 
  </tr> 
  <tr> 
   <td>OP_CONNECT</td> 
   <td>连接请求(客户端)</td> 
  </tr> 
  <tr> 
   <td>OP_ACCEPT</td> 
   <td>接收请求(服务端)</td> 
  </tr> 
 </tbody> 
</table>

// 随机接收客户端请求
    Collections.shuffle(selectedList);

这行代码就是对连接进来的请求进行随机打乱，打个比方，当有十个客户端同时发来请求的时候，服务端不会根据其顺序来依次进行出来，它会对这些请求进行一个shuffle随机排序，然后去一个个执行，这样就避免了当有一个客户端连续发送进来很多请求之后，zookeeper没法处理其他客户端的请求了。这样会让客户端进来的请求进行以比较均匀的处理，不会因为某一个客户端请求量很大导致其他客户端的请求处于一直阻塞状态

for (SelectionKey k : selectedList) {
                        //连接请求处理
                        if ((k.readyOps() & SelectionKey.OP_ACCEPT) != 0) {
                            SocketChannel sc = ((ServerSocketChannel) k
                                    .channel()).accept();
                            InetAddress ia = sc.socket().getInetAddress();
                            int cnxncount = getClientCnxnCount(ia);
                            if (maxClientCnxns > 0 && cnxncount >= maxClientCnxns){
                                LOG.warn("Too many connections from " + ia
                                         + " - max is " + maxClientCnxns );
                                sc.close();
                            } else {
                                LOG.info("Accepted socket connection from "
                                         + sc.socket().getRemoteSocketAddress());
                                sc.configureBlocking(false);
                                SelectionKey sk = sc.register(selector,
                                        SelectionKey.OP_READ);
                                NIOServerCnxn cnxn = createConnection(sc, sk);
                                sk.attach(cnxn);
                                addCnxn(cnxn);
                            }
                        }
                    }

我们先看一下连接的请求，else中的代码就是客户端发送的读写请求处理

当一个客户端要和服务端进行一个连接时，会先进行三次握手，发送一个OP\_ACCEPT，当服务端接受到这个请求之后，会进入到下面的代码进行等待连接

SocketChannel sc = ((ServerSocketChannel) k.channel()).accept();

调用accept方法，默认是阻塞的，就是当调用了accept()方法时就会进入阻塞状态, 等待客户端的连接。会看我们在上面的代码中设置了ss.configureBlocking(false);设置是非阻塞，如果设置了非阻塞模式，则当调用完accept()方法时，不会进行阻塞，而是会直接返回一个null，线程可以继续去处理其他请求。当有客户端三次握手完毕，则线程会继续向下执行

InetAddress ia = sc.socket().getInetAddress();
                            int cnxncount = getClientCnxnCount(ia);
                            if (maxClientCnxns > 0 && cnxncount >= maxClientCnxns){
                                LOG.warn("Too many connections from " + ia
                                         + " - max is " + maxClientCnxns );
                                sc.close();
                            } else {

这块代码就是校验一下连接是否有超过我们配置的maxClientCnxns最大连接数，如果超过了，则直接把当前socket关闭

sc.configureBlocking(false);
    SelectionKey sk = sc.register(selector,SelectionKey.OP_READ);
    NIOServerCnxn cnxn = createConnection(sc, sk);
    sk.attach(cnxn);
    addCnxn(cnxn);

如果确认连接之后，则和此服务端建立好的socketChannel会注册对OP\_READ感兴趣，接受客户端发送过来的请求，接着创建一个连接对象NIOServerCnxn对象

protected NIOServerCnxn createConnection(SocketChannel sock,
                SelectionKey sk) throws IOException {
            return new NIOServerCnxn(zkServer, sock, sk, this);
        }

public NIOServerCnxn(ZooKeeperServer zk, SocketChannel sock,
                SelectionKey sk, NIOServerCnxnFactory factory) throws IOException {
            this.zkServer = zk;
            this.sock = sock;
            this.sk = sk;
            this.factory = factory;
            if (this.factory.login != null) {
                this.zooKeeperSaslServer = new ZooKeeperSaslServer(factory.login);
            }
            if (zk != null) { 
                outstandingLimit = zk.getGlobalOutstandingLimit();
            }
            sock.socket().setTcpNoDelay(true);// 非延迟发送
            /* set socket linger to false, so that socket close does not
             * block */
            sock.socket().setSoLinger(false, -1);
            InetAddress addr = ((InetSocketAddress) sock.socket()
                    .getRemoteSocketAddress()).getAddress();
            authInfo.add(new Id("ip", addr.getHostAddress()));
            sk.interestOps(SelectionKey.OP_READ);
        }

就是初始化了一些信息，这个类就是来处理客户端发送过来的请求的

接着执行的代码是

sk.attach(cnxn);

就是把当前注册感兴趣的SelectionKey和当前cnxn处理客户端请求的类进行一个绑定

private void addCnxn(NIOServerCnxn cnxn) {
            synchronized (cnxns) {
                cnxns.add(cnxn);
                synchronized (ipMap){
                    InetAddress addr = cnxn.sock.socket().getInetAddress();
                    Set<NIOServerCnxn> s = ipMap.get(addr);
                    if (s == null) {
                        // in general we will see 1 connection from each
                        // host, setting the initial cap to 2 allows us
                        // to minimize mem usage in the common case
                        // of 1 entry --  we need to set the initial cap
                        // to 2 to avoid rehash when the first entry is added
                        s = new HashSet<NIOServerCnxn>(2);
                        s.add(cnxn);
                        ipMap.put(addr,s);
                    } else {
                        s.add(cnxn);
                    }
                }
            }
        }

接着调用的addCnxn()，简单理解为就是把和客户端建立的连接的NIOServerCnxn放入到cnxns集合中保存起来，ipMap集合就是保证一个客户端只有一个一个连接，保存了addr和NIOServerCnxn的关联关系，下次当有对应的连接发送请求时，可以直接从此集合中获取即可

//处理请求，并返回响应
                        else if ((k.readyOps() & (SelectionKey.OP_READ | SelectionKey.OP_WRITE)) != 0) {
                            NIOServerCnxn c = (NIOServerCnxn) k.attachment();
                            c.doIO(k);
                        }

大概看一下else中的代码，就是关注OP\_READ和OP\_WRITE,，当客户端有读写请求的时候都会进入这里面

先从SelectionKey中获取和这个key关联的NIOServerCnxn, 然后执行doIO()方法，这个方法里面是执行客户端发送过来的读写请求的具体代码，今天暂时先不分析了

通过本篇文章大概知道了zookeeper怎么对clientPort端口进行监听的、以及当有客户端要进行连接是zk集群是如何处理的

下篇文章先分析一下客户端如何发起和zk集群进行连接的，然后他们之间就可以进行通信了，接着再具体分析中间的代码如何去执行的，读写请求怎么处理的了

[zookeeper]: https://blog.csdn.net/liereli/article/details/107738119