【Zookeeper】源码分析之网络通信（三）之NettyServerCnxn

    private class SendBufferWriter extends Writer {
        private StringBuffer sb = new StringBuffer();
        
        /**
         * Check if we are ready to send another chunk.
         * @param force force sending, even if not a full chunk
         */
        // 是否准备好发送另一块
        private void checkFlush(boolean force) {
            if ((force && sb.length() > 0) || sb.length() > 2048) { // 当强制发送并且sb大小大于0，或者sb大小大于2048即发送缓存
                sendBuffer(ByteBuffer.wrap(sb.toString().getBytes()));
                // clear our internal buffer
                sb.setLength(0);
            }
        }

        @Override
        public void close() throws IOException {
            if (sb == null) return;
            // 关闭之前需要强制性发送缓存
            checkFlush(true);
            sb = null; // clear out the ref to ensure no reuse
        }

        @Override
        public void flush() throws IOException {
            checkFlush(true);
        }

        @Override
        public void write(char[] cbuf, int off, int len) throws IOException {
            sb.append(cbuf, off, len);
            checkFlush(false);
        }
    }

    static class ResumeMessageEvent implements MessageEvent {
        // 通道
        Channel channel;
        
        // 构造函数
        ResumeMessageEvent(Channel channel) {
            this.channel = channel;
        }
        @Override
        public Object getMessage() {return null;}
        @Override
        public SocketAddress getRemoteAddress() {return null;}
        @Override
        public Channel getChannel() {return channel;}
        @Override
        public ChannelFuture getFuture() {return null;}
    };

    private abstract class CommandThread /*extends Thread*/ {
        PrintWriter pw;
        
        CommandThread(PrintWriter pw) {
            this.pw = pw;
        }
        
        public void start() {
            run();
        }

        public void run() {
            try {
                commandRun();
            } catch (IOException ie) {
                LOG.error("Error in running command ", ie);
            } finally {
                cleanupWriterSocket(pw);
            }
        }
        
        public abstract void commandRun() throws IOException;
    }

public class NettyServerCnxn extends ServerCnxn {
    // 日志
    Logger LOG = LoggerFactory.getLogger(NettyServerCnxn.class);
    
    // 通道
    Channel channel;
    
    // 通道缓存
    ChannelBuffer queuedBuffer;
    
    // 节流与否
    volatile boolean throttled;
    
    // Byte缓冲区
    ByteBuffer bb;
    
    // 四个字节的缓冲区
    ByteBuffer bbLen = ByteBuffer.allocate(4);
    
    // 会话ID
    long sessionId;
    
    // 会话超时时间
    int sessionTimeout;
    
    // 计数
    AtomicLong outstandingCount = new AtomicLong();

    /** The ZooKeeperServer for this connection. May be null if the server
     * is not currently serving requests (for example if the server is not
     * an active quorum participant.
     */
    // Zookeeper服务器
    private volatile ZooKeeperServer zkServer;

    // NettyServerCnxn工厂
    NettyServerCnxnFactory factory;
    
    // 初始化与否
    boolean initialized;
    
    // 四个字节
    private static final byte[] fourBytes = new byte[4];
    
    private static final String ZK_NOT_SERVING =
        "This ZooKeeper instance is not currently serving requests";
}

    NettyServerCnxn(Channel channel, ZooKeeperServer zks, NettyServerCnxnFactory factory) {
        // 给属性赋值
        this.channel = channel;
        this.zkServer = zks;
        this.factory = factory;
        if (this.factory.login != null) { // 需要登录信息(用户名和密码登录)
            this.zooKeeperSaslServer = new ZooKeeperSaslServer(factory.login);
        }
    }

    public void receiveMessage(ChannelBuffer message) {
        try {
            while(message.readable() && !throttled) { // 当writerIndex > readerIndex，并且不节流时，满足条件
                if (bb != null) { // 不为null
                    if (LOG.isTraceEnabled()) {
                        LOG.trace("message readable " + message.readableBytes()
                                + " bb len " + bb.remaining() + " " + bb);
                        ByteBuffer dat = bb.duplicate();
                        dat.flip();
                        LOG.trace(Long.toHexString(sessionId)
                                + " bb 0x"
                                + ChannelBuffers.hexDump(
                                        ChannelBuffers.copiedBuffer(dat)));
                    }

                    if (bb.remaining() > message.readableBytes()) { // bb剩余空间大于message中可读字节大小
                        // 确定新的limit
                        int newLimit = bb.position() + message.readableBytes();
                        bb.limit(newLimit);
                    }
                    // 将message写入bb中
                    message.readBytes(bb);
                    // 重置bb的limit
                    bb.limit(bb.capacity());

                    if (LOG.isTraceEnabled()) {
                        LOG.trace("after readBytes message readable "
                                + message.readableBytes()
                                + " bb len " + bb.remaining() + " " + bb);
                        ByteBuffer dat = bb.duplicate();
                        dat.flip();
                        LOG.trace("after readbytes "
                                + Long.toHexString(sessionId)
                                + " bb 0x"
                                + ChannelBuffers.hexDump(
                                        ChannelBuffers.copiedBuffer(dat)));
                    }
                    if (bb.remaining() == 0) { // 已经读完message，表示内容已经全部接收
                        // 统计接收信息
                        packetReceived();
                        // 翻转，可读
                        bb.flip();

                        ZooKeeperServer zks = this.zkServer;
                        if (zks == null) { // Zookeeper服务器为空
                            throw new IOException("ZK down");
                        }
                        if (initialized) { // 未被初始化
                            // 处理bb中包含的包信息
                            zks.processPacket(this, bb);

                            if (zks.shouldThrottle(outstandingCount.incrementAndGet())) { // 是否已经节流
                                // 不接收数据
                                disableRecvNoWait();
                            }
                        } else { // 已经初始化
                            LOG.debug("got conn req request from "
                                    + getRemoteSocketAddress());
                            // 处理连接请求
                            zks.processConnectRequest(this, bb);
                            initialized = true;
                        }
                        bb = null;
                    }
                } else { // bb为null
                    if (LOG.isTraceEnabled()) {
                        LOG.trace("message readable "
                                + message.readableBytes()
                                + " bblenrem " + bbLen.remaining());
                        // 复制bbLen缓冲
                        ByteBuffer dat = bbLen.duplicate();
                        // 翻转
                        dat.flip();
                        LOG.trace(Long.toHexString(sessionId)
                                + " bbLen 0x"
                                + ChannelBuffers.hexDump(
                                        ChannelBuffers.copiedBuffer(dat)));
                    }

                    if (message.readableBytes() < bbLen.remaining()) { // bb剩余空间大于message中可读字节大小
                        // 重设bbLen的limit
                        bbLen.limit(bbLen.position() + message.readableBytes());
                    }
                    // 将message内容写入bbLen中
                    message.readBytes(bbLen);
                    // 重置bbLen的limit
                    bbLen.limit(bbLen.capacity());
                    if (bbLen.remaining() == 0) { // 已经读完message，表示内容已经全部接收
                        // 翻转
                        bbLen.flip();

                        if (LOG.isTraceEnabled()) {
                            LOG.trace(Long.toHexString(sessionId)
                                    + " bbLen 0x"
                                    + ChannelBuffers.hexDump(
                                            ChannelBuffers.copiedBuffer(bbLen)));
                        }
                        // 读取position后四个字节
                        int len = bbLen.getInt();
                        if (LOG.isTraceEnabled()) {
                            LOG.trace(Long.toHexString(sessionId)
                                    + " bbLen len is " + len);
                        }
                        
                        // 清除缓存
                        bbLen.clear();
                        if (!initialized) { // 未被初始化
                            if (checkFourLetterWord(channel, message, len)) { // 是否是四个字母的命令
                                return;
                            }
                        }
                        if (len < 0 || len > BinaryInputArchive.maxBuffer) {
                            throw new IOException("Len error " + len);
                        }
                        // 根据len重新分配缓冲，以便接收内容
                        bb = ByteBuffer.allocate(len);
                    }
                }
            }
        } catch(IOException e) {
            LOG.warn("Closing connection to " + getRemoteSocketAddress(), e);
            close();
        }
    }