摘要

每台服务器在启动的过程中，会启动一个QuorumPeerManager，负责各台服务器之间的底层Leader选举过程中的网络通信。
这个类就是QuorumCnxManager
在看本节源码之前，建议先把refer中的内容看完，也不多，对于基本概念有一个理解

本节主要讲解

内部类
  SendWorker类作为网络IO的发送者，从发送队列取出，发给对应sid的机器
  Message类定义了消息结构，包含sid以及消息体ByteBuffer
  RecvWorker类作为网络IO的接受者
  Listener类作为electionPort端口的监听器，等待其他机器的连接
属性
  recvQueue作为接受队列
  queueSendMap表示每个sid对应的发送的发送队列
函数
  连接相关
  sender，recv生产消费相关
  其他
思考以及总结

内部类

QuorumCnxManager的内部类

可以看到有四个内部类
SendWorker类，Message类，RecvWorker类,Listener类

SendWorker

这个类作为“发送者”，继承ZooKeeperThread，线程不断地从发送队列取出，发给对应sid的机器
属性

        Long sid;//目标机器sid，不是当前机器sid
        Socket sock;
        RecvWorker recvWorker;//该sid对应的RecvWorker
        volatile boolean running = true;
        DataOutputStream dout;

主要方法

构造函数

        SendWorker(Socket sock, Long sid) {
            super("SendWorker:" + sid);
            this.sid = sid;
            this.sock = sock;
            recvWorker = null;
            try {
                dout = new DataOutputStream(sock.getOutputStream());
            } catch (IOException e) {
                LOG.error("Unable to access socket output stream", e);
                closeSocket(sock);
                running = false;
            }
            LOG.debug("Address of remote peer: " + this.sid);
        }

run方法

        @Override
        public void run() {
            threadCnt.incrementAndGet();//线程数+1
            try {
                /**
                 * If there is nothing in the queue to send, then we
                 * send the lastMessage to ensure that the last message
                 * was received by the peer. The message could be dropped
                 * in case self or the peer shutdown their connection
                 * (and exit the thread) prior to reading/processing
                 * the last message. Duplicate messages are handled correctly
                 * by the peer.
                 *
                 * If the send queue is non-empty, then we have a recent
                 * message than that stored in lastMessage. To avoid sending
                 * stale message, we should send the message in the send queue.
                 */
                ArrayBlockingQueue<ByteBuffer> bq = queueSendMap.get(sid);//找到sid对应需要send的队列
                if (bq == null || isSendQueueEmpty(bq)) {
                   ByteBuffer b = lastMessageSent.get(sid);//如果没有什么发的，就把上一次发的再发一遍(重发能够正确处理)
                   if (b != null) {
                       LOG.debug("Attempting to send lastMessage to sid=" + sid);
                       send(b);//发送
                   }
                }
            } catch (IOException e) {
                LOG.error("Failed to send last message. Shutting down thread.", e);
                this.finish();
            }
            
            try {
                while (running && !shutdown && sock != null) {

                    ByteBuffer b = null;
                    try {
                        ArrayBlockingQueue<ByteBuffer> bq = queueSendMap
                                .get(sid);
                        if (bq != null) {
                            b = pollSendQueue(bq, 1000, TimeUnit.MILLISECONDS);//从发送队列里面取出消息
                        } else {//队列没有记录在map中
                            LOG.error("No queue of incoming messages for " +
                                      "server " + sid);
                            break;
                        }

                        if(b != null){
                            lastMessageSent.put(sid, b);//更新最后一次发送的
                            send(b);//发送
                        }
                    } catch (InterruptedException e) {
                        LOG.warn("Interrupted while waiting for message on queue",
                                e);
                    }
                }
            } catch (Exception e) {
                LOG.warn("Exception when using channel: for id " + sid + " my id = " + 
                        self.getId() + " error = " + e);
            }
            this.finish();
            LOG.warn("Send worker leaving thread");
        }
    }


send方法

        synchronized void send(ByteBuffer b) throws IOException {
            byte[] msgBytes = new byte[b.capacity()];
            try {
                b.position(0);
                b.get(msgBytes);
            } catch (BufferUnderflowException be) {
                LOG.error("BufferUnderflowException ", be);
                return;
            }
            dout.writeInt(b.capacity());
            dout.write(b.array());
            dout.flush();
        }

里面涉及的字段在下面会讲，这里注意
在SendWorker中，一旦Zookeeper发现针对当前服务器的消息发送队列为空，那么此时需要从lastMessageSent中取出一个最近发送过的消息来进行再次发送，这是为了解决接收方在消息接收前或者接收到消息后服务器挂了，导致消息尚未被正确处理。同时，Zookeeper能够保证接收方在处理消息时，会对重复消息进行正确的处理。

Message

这个类定义了server之间传输的消息结构，源码如下

    static public class Message {
        
        Message(ByteBuffer buffer, long sid) {
            this.buffer = buffer;
            this.sid = sid;
        }

        ByteBuffer buffer;
        long sid;
    }

sid为消息来源方的sid，buffer即消息体

RecvWorker

这个类作为“接受者”，类似SendWorker，继承ZooKeeperThread，线程不断地从网络IO中读取数据，放入接收队列
属性

        Long sid;//来源方sid
        Socket sock;
        volatile boolean running = true;
        DataInputStream din;//input
        final SendWorker sw;

主要方法

构造方法

        RecvWorker(Socket sock, Long sid, SendWorker sw) {
            super("RecvWorker:" + sid);
            this.sid = sid;
            this.sock = sock;
            this.sw = sw;
            try {
                din = new DataInputStream(sock.getInputStream());
                // OK to wait until socket disconnects while reading.
                sock.setSoTimeout(0);
            } catch (IOException e) {
                LOG.error("Error while accessing socket for " + sid, e);
                closeSocket(sock);
                running = false;
            }
        }

run方法
        @Override
        public void run() {
            threadCnt.incrementAndGet();
            try {
                while (running && !shutdown && sock != null) {
                    /**
                     * Reads the first int to determine the length of the
                     * message
                     */
                    int length = din.readInt();//获取长度
                    if (length <= 0 || length > PACKETMAXSIZE) {
                        throw new IOException(
                                "Received packet with invalid packet: "
                                        + length);
                    }
                    /**
                     * Allocates a new ByteBuffer to receive the message
                     */
                    byte[] msgArray = new byte[length];
                    din.readFully(msgArray, 0, length);
                    ByteBuffer message = ByteBuffer.wrap(msgArray);//解析出byteBuffer
                    addToRecvQueue(new Message(message.duplicate(), sid));//加入接收队列
                }
            } catch (Exception e) {
                LOG.warn("Connection broken for id " + sid + ", my id = " + 
                        self.getId() + ", error = " , e);
            } finally {
                LOG.warn("Interrupting SendWorker");
                sw.finish();
                if (sock != null) {
                    closeSocket(sock);
                }
            }
        }
    }

Listener

这个类也继承ZooKeeperThread，主要监听electionPort，不断的接收外部连接
run方法核心代码如下

                     while (!shutdown) {
                        Socket client = ss.accept();
                        setSockOpts(client);
                        LOG.info("Received connection request "
                                + client.getRemoteSocketAddress());
                        receiveConnection(client);//不断接受连接
                        numRetries = 0;
                    }

内部类小结

SendWorker和RecvWorker互相依赖对方，原因在下面思考中列出
RecvWorker相比SendWorker代码要好理解

两者都有属性sid，表示每个机器和其他机器连接时，按sid区分不同的RecvWorker和SendWorker
好比sid1和其余(n-1)个server建立连接，那么就按sid分开，有(n-1)个RecvWorker和SendWorker

Message作为消息的封装，包含sid和ByteBuffer作为消息体

Listener主要监听本机配置的electionPort，不断的接收外部连接

属性

将重要属性字段整理如下

比较好理解

函数

请求连接，接收连接相关函数

请求连接相关

connectOne:连接上一个sid的服务器

    synchronized void connectOne(long sid){//连接上某个sid的server
        if (senderWorkerMap.get(sid) == null){//如果没有记录在sender的map里面
            InetSocketAddress electionAddr;
            if (self.quorumPeers.containsKey(sid)) {
                electionAddr = self.quorumPeers.get(sid).electionAddr;//从配置文件获取对应sid机器的选举端口
            } else {
                LOG.warn("Invalid server id: " + sid);
                return;
            }
            try {

                if (LOG.isDebugEnabled()) {
                    LOG.debug("Opening channel to server " + sid);
                }
                Socket sock = new Socket();
                setSockOpts(sock);
                sock.connect(self.getView().get(sid).electionAddr, cnxTO);//连接上对应socket
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Connected to server " + sid);
                }
                initiateConnection(sock, sid);//初始化连接
            } catch (UnresolvedAddressException e) {
                // Sun doesn't include the address that causes this
                // exception to be thrown, also UAE cannot be wrapped cleanly
                // so we log the exception in order to capture this critical
                // detail.
                LOG.warn("Cannot open channel to " + sid
                        + " at election address " + electionAddr, e);
                // Resolve hostname for this server in case the
                // underlying ip address has changed.
                if (self.getView().containsKey(sid)) {
                    self.getView().get(sid).recreateSocketAddresses();
                }
                throw e;
            } catch (IOException e) {
                LOG.warn("Cannot open channel to " + sid
                        + " at election address " + electionAddr,
                        e);
                // We can't really tell if the server is actually down or it failed
                // to connect to the server because the underlying IP address
                // changed. Resolve the hostname again just in case.
                if (self.getView().containsKey(sid)) {
                    self.getView().get(sid).recreateSocketAddresses();
                }
            }
        } else {
            LOG.debug("There is a connection already for server " + sid);
        }
    }

如果senderWorkerMap没有sid对应记录，代表目前没有连接，那么就去连，主要核心代码是调用
initiateConnection函数

initiateConnection:初始化连接

    /**
     * If this server has initiated the connection, then it gives up on the
     * connection if it loses challenge. Otherwise, it keeps the connection.
     */
    public boolean initiateConnection(Socket sock, Long sid) {//初始化连接
        DataOutputStream dout = null;
        try {
            // Sending id and challenge
            dout = new DataOutputStream(sock.getOutputStream());
            dout.writeLong(self.getId());//发送本机sid
            dout.flush();
        } catch (IOException e) {
            LOG.warn("Ignoring exception reading or writing challenge: ", e);
            closeSocket(sock);
            return false;
        }
        
        // If lost the challenge, then drop the new connection
        if (sid > self.getId()) {//发送连接的时候，只让大sid给小sid发送，如果当前sid小，那就close掉
            LOG.info("Have smaller server identifier, so dropping the " +
                     "connection: (" + sid + ", " + self.getId() + ")");
            closeSocket(sock);
            // Otherwise proceed with the connection
        } else {//自己sid大，初始化SendWorker和RecvWorker
            SendWorker sw = new SendWorker(sock, sid);
            RecvWorker rw = new RecvWorker(sock, sid, sw);//rw记录sw
            sw.setRecv(rw);//sw记录rw

            SendWorker vsw = senderWorkerMap.get(sid);
            
            if(vsw != null)
                vsw.finish();//finish掉sid对应的SendWorker,vsw
            
            senderWorkerMap.put(sid, sw);//放入新的SendWorker,sw
            if (!queueSendMap.containsKey(sid)) {
                queueSendMap.put(sid, new ArrayBlockingQueue<ByteBuffer>(
                        SEND_CAPACITY));
            }
            
            sw.start();
            rw.start();
            
            return true;    
            
        }
        return false;
    }

接收连接相关

receiveConnection：接收外部connect

    public void receiveConnection(Socket sock) {//接收connect
        Long sid = null;
        
        try {
            // Read server id
            DataInputStream din = new DataInputStream(sock.getInputStream());
            sid = din.readLong();
            if (sid < 0) { // this is not a server id but a protocol version (see ZOOKEEPER-1633)
                sid = din.readLong();

                // next comes the #bytes in the remainder of the message
                // note that 0 bytes is fine (old servers)
                int num_remaining_bytes = din.readInt();
                if (num_remaining_bytes < 0 || num_remaining_bytes > maxBuffer) {
                    LOG.error("Unreasonable buffer length: {}", num_remaining_bytes);
                    closeSocket(sock);
                    return;
                }
                byte[] b = new byte[num_remaining_bytes];

                // remove the remainder of the message from din
                int num_read = din.read(b);
                if (num_read != num_remaining_bytes) {
                    LOG.error("Read only " + num_read + " bytes out of " + num_remaining_bytes + " sent by server " + sid);
                }
            }
            if (sid == QuorumPeer.OBSERVER_ID) {
                /*
                 * Choose identifier at random. We need a value to identify
                 * the connection.
                 */
                
                sid = observerCounter--;
                LOG.info("Setting arbitrary identifier to observer: " + sid);
            }
        } catch (IOException e) {
            closeSocket(sock);
            LOG.warn("Exception reading or writing challenge: " + e.toString());
            return;
        }
        
        //If wins the challenge, then close the new connection.
        if (sid < self.getId()) {//如果自己id大，就close掉当前连接(当前是小sid发给大sid的连接)，自己再去连对方sid
            /*
             * This replica might still believe that the connection to sid is
             * up, so we have to shut down the workers before trying to open a
             * new connection.
             */
            SendWorker sw = senderWorkerMap.get(sid);
            if (sw != null) {
                sw.finish();
            }

            /*
             * Now we start a new connection
             */
            LOG.debug("Create new connection to server: " + sid);
            closeSocket(sock);
            connectOne(sid);

            // Otherwise start worker threads to receive data.
        } else {
            SendWorker sw = new SendWorker(sock, sid);
            RecvWorker rw = new RecvWorker(sock, sid, sw);
            sw.setRecv(rw);

            SendWorker vsw = senderWorkerMap.get(sid);
            
            if(vsw != null)
                vsw.finish();
            
            senderWorkerMap.put(sid, sw);
            
            if (!queueSendMap.containsKey(sid)) {
                queueSendMap.put(sid, new ArrayBlockingQueue<ByteBuffer>(
                        SEND_CAPACITY));
            }
            
            sw.start();
            rw.start();
            
            return;
        }
    }

这里有个概念，就是win challenge和lose challenge
在zk中，为了保证每一对server只有一个socket，Zookeeper只允许SID大的服务器主动和其他机器建立连接，否则断开连接。
发出连接时，要求自己sid大，完成SendWorker和ReceiveWorker的构造以及线程启动，否则close
接收连接时，要求自己sid小，完成SendWorker和ReceiveWorker的构造以及线程启动，否则close
在“思考”中也会分析

sender生产，消费相关函数

生成，加入sender队列

    private void addToSendQueue(ArrayBlockingQueue<ByteBuffer> queue,
          ByteBuffer buffer) {//发送队列长度为1，如果满了就remove，然后add
        if (queue.remainingCapacity() == 0) {
            try {
                queue.remove();
            } catch (NoSuchElementException ne) {
                // element could be removed by poll()
                LOG.debug("Trying to remove from an empty " +
                        "Queue. Ignoring exception " + ne);
            }
        }
        try {
            queue.add(buffer);
        } catch (IllegalStateException ie) {
            // This should never happen
            LOG.error("Unable to insert an element in the queue " + ie);
        }
    }

每个sender的队列长度都是1,为了避免发送旧的数据，因此会把旧的remove掉

消费sender队列

    private ByteBuffer pollSendQueue(ArrayBlockingQueue<ByteBuffer> queue,
          long timeout, TimeUnit unit) throws InterruptedException {
       return queue.poll(timeout, unit);
    }

recv生产，消费相关函数

recv队列生产

    public void addToRecvQueue(Message msg) {
        synchronized(recvQLock) {
            if (recvQueue.remainingCapacity() == 0) {
                try {
                    recvQueue.remove();
                } catch (NoSuchElementException ne) {
                    // element could be removed by poll()
                     LOG.debug("Trying to remove from an empty " +
                         "recvQueue. Ignoring exception " + ne);
                }
            }
            try {
                recvQueue.add(msg);//加入接收队列
            } catch (IllegalStateException ie) {
                // This should never happen
                LOG.error("Unable to insert element in the recvQueue " + ie);
            }
        }
    }

这里并不是很清楚为什么加入队列时，如果满了要把前面的remove掉，队列的长度上限是100

recv队列消费

    public Message pollRecvQueue(long timeout, TimeUnit unit)
       throws InterruptedException {
       return recvQueue.poll(timeout, unit);
    }

其他函数

toSend

将消息根据sid添加进recv队列或者send队列,间接调用send，recv的生产

    public void toSend(Long sid, ByteBuffer b) {
        /*
         * If sending message to myself, then simply enqueue it (loopback).
         */
        if (self.getId() == sid) {//如果发送给自己,加入recv队列
             b.position(0);
             addToRecvQueue(new Message(b.duplicate(), sid));
            /*
             * Otherwise send to the corresponding thread to send.
             */
        } else {
             /*
              * Start a new connection if doesn't have one already.
              */
             if (!queueSendMap.containsKey(sid)) {//如果发送map没有记录这个sid
                 ArrayBlockingQueue<ByteBuffer> bq = new ArrayBlockingQueue<ByteBuffer>(
                         SEND_CAPACITY);//阻塞队列长度为1
                 queueSendMap.put(sid, bq);
                 addToSendQueue(bq, b);

             } else {
                 ArrayBlockingQueue<ByteBuffer> bq = queueSendMap.get(sid);
                 if(bq != null){
                     addToSendQueue(bq, b);
                 } else {
                     LOG.error("No queue for server " + sid);
                 }
             }
             connectOne(sid);//和这个sid建立连接
                
        }
    }

haveDelivered

是否发送过消息

    /**
     * Check if all queues are empty, indicating that all messages have been delivered.
     */
    boolean haveDelivered() {//如果有一个队列是空的，代表发送过了，和注释不一致
        for (ArrayBlockingQueue<ByteBuffer> queue : queueSendMap.values()) {
            LOG.debug("Queue size: " + queue.size());
            if (queue.size() == 0) {
                return true;
            }
        }

        return false;
    }

这里吐槽一下，代码和注释不一致，在"思考"中进行讲解

连接所有queueSendMap的sid

    public void connectAll(){//把所有需要发送消息的机器sid都连接上
        long sid;
        for(Enumeration<Long> en = queueSendMap.keys();//连接所有queueSendMap记录的sid
            en.hasMoreElements();){
            sid = en.nextElement();
            connectOne(sid);
        }      
    }

思考

tricky方法的体现:每一对server之间只有一个连接

可以理解成n个server，互相之间都要用connection
好比n个点，用无向的边连起来，用[sidn,sid1]表示sidn向sid1建立了连接
那么，[sid1,sidn]就没有存在的必要了，也就是n*(n+1)/2条边就够了

代码里面规定的是，正常条件下只有sid大的server向sid小的server建立连接
体现在哪

连接请求发出时，如果对方sid比自己大，仅仅发送自己sid也就是一个long过去，然后close掉
QuorumCnxManager#initiateConnection

接收连接请求时，如果对方sid比自己小，那么close掉socket然后自己去连接对方sid
QuorumCnxManager#receiveConnection

也就是说，当sid小的向sid大的server发送连接请求时，也只是在告诉对方
“你sid大，你来连我”

SendWorker以及RecvWorker的初始化
如何保证[sidn,sid1]这样的连接中，双方都有初始化两个worker
在连接请求发出时，sid大的一方，也就是sidn初始化两个worker
在接收连接请求时，sid小的一方，也就是sid1初始化两个worker
代码同样体现在上面两个函数中

为什么RecvWorker和SendWorker要互相记录对方

代码里面就是finish的时候
SendWorker#finish调用对应的RecvWorker#finish
RecvWorker#run的finally段也去调用SendWorker#finish

然后变量senderWorkerMap是final ConcurrentHashMap<Long, SendWorker>类型的，
可以看到并不存在对应RecvWorker的map，
所以原因就是 根据sid找到SendWorker，然后方便调用finish方法

为什么发送队列的长度为1,入队时满了就要把前面的踢出去

长度为1 QuorumCnxManager#SEND_CAPACITY
踢出去 QuorumCnxManager#addToSendQueue

应该参考SEND_CAPACITY注释

// Initialized to 1 to prevent sending
// stale notifications to peers

因为是选举leader投票，有特殊的要求:如果之前的票还没有投出去又产生了新的票，那么旧的票就可以直接作废了,不用真正的投出去

发送队列的生产，消费与发送的调用顺序

生产 
QuorumCnxManager#toSend
QuorumCnxManager#addToSendQueue

消费
QuorumCnxManager.SendWorker#run
QuorumCnxManager#pollSendQueue

发送
QuorumCnxManager.SendWorker#run
SendWorker#send

queueSendMap的意义

final ConcurrentHashMap<Long, ArrayBlockingQueue<ByteBuffer>> queueSendMap;//消息发送队列，key为各机器sid

强调一下，key是sid，代表当前机器和对应sid的机器
建立了联系或者有要发送的内容

queueSendMap的改动只有put操作,没有remove等操作
基本都是

    queueSendMap.put(sid, new ArrayBlockingQueue<ByteBuffer>(
                        SEND_CAPACITY));

QuorumCnxManager#haveDelivered 注释和代码不一致

源码在之前贴过了
这里的代码和注释不一致，加入生产队列的函数addToSendQueue只有在QuorumCnxManager#toSend中被调用

toSend调用addToSendQueue

问题

SendWorker发送队列没有东西的时候，把最后一次发送的内容再发一遍

说是这样能解决 ”接收方在消息接收前或者接收到消息后服务器挂了“ 的问题
那么倒数第二条为什么不发，倒数第三条为什么不发？？？

QuorumCnxManager#addToRecvQueue接收队列满的时候就把最前面的删掉

为什么，漏发了没关系吗

总结

这个类就是网络IO的调度器
SendWorker和RecvWorker作为两个线程不断将消息进行收发
Listener监听外部来的连接
主要属性 queueSendMap记录和哪些sid建立联系，对应的发送队列是什么

refer

http://www.cnblogs.com/leesf456/p/6107600.html 第3部分 QuorumCnxManager：网络I/O
《paoxs到zk》 7.6.3