1
/**
2
 * Defaults to using HTTP/1.1 NIO implementation.
3
 */
4
public Connector() {
5
    this("HTTP/1.1");
6
}

1
public Http11NioProtocol() {
2
       super(new NioEndpoint());
3
   }

1
@Override
2
    public void bind() throws Exception {
3
        initServerSocket();
4
        setStopLatch(new CountDownLatch(1));
5
        // Initialize SSL if needed
6
        initialiseSsl();
7
        selectorPool.open(getName());
8
    }
9

10
  //socket相关  initServerSocket()具体如下
11
  // Separated out to make it easier for folks that extend NioEndpoint to
12
    // implement custom [server]sockets
13
    protected void initServerSocket() throws Exception {
14
            //.......
15
          //根据平台不同，反回具体底层类对象（windows,linux,unix）
16
            serverSock = ServerSocketChannel.open();
17
            socketProperties.setProperties(serverSock.socket());
18
            //绑定地址和端口号
19
            InetSocketAddress addr = new InetSocketAddress(getAddress(), getPortWithOffset());
20
            serverSock.socket().bind(addr,getAcceptCount());
21
          //.......
22
    }

1
// Create worker collection
2
if (getExecutor() == null) {
3
    //创建线程池
4
    createExecutor();
5
}
6

7
initializeConnectionLatch();
8
// Start poller thread
9
// 创建客户端队列（客户端过来的请求）
10

11
poller = new Poller();
12
Thread pollerThread = new Thread(poller, getName() + "-ClientPoller");
13
pollerThread.setPriority(threadPriority);
14
pollerThread.setDaemon(true);
15
pollerThread.start();
16

17
//创建接收远程请求线程
18
startAcceptorThread();

1
/**
2
         * Processes events in the event queue of the Poller.
3
         *
4
         * @return <code>true</code> if some events were processed,
5
         *   <code>false</code> if queue was empty
6
         */
7
        public boolean events() {
8
            boolean result = false;
9

10
            PollerEvent pe = null;
11
            for (int i = 0, size = events.size(); i < size && (pe = events.poll()) != null; i++ ) {
12
                result = true;
13
                NioSocketWrapper socketWrapper = pe.getSocketWrapper();
14
                SocketChannel sc = socketWrapper.getSocket().getIOChannel();
15
                int interestOps = pe.getInterestOps();
16
                if (sc == null) {
17
                    log.warn(sm.getString("endpoint.nio.nullSocketChannel"));
18
                    socketWrapper.close();
19
                } else if (interestOps == OP_REGISTER) {
20
                    try {
21
                        //注册绑定
22
                        sc.register(getSelector(), SelectionKey.OP_READ, socketWrapper);
23
                    } catch (Exception x) {
24
                        log.error(sm.getString("endpoint.nio.registerFail"), x);
25
                    }
26
                } else {
27
                    final SelectionKey key = sc.keyFor(getSelector());
28
                    if (key == null) {
29
                        // The key was cancelled (e.g. due to socket closure)
30
                        // and removed from the selector while it was being
31
                        // processed. Count down the connections at this point
32
                        // since it won't have been counted down when the socket
33
                        // closed.
34
                        socketWrapper.close();
35
                    } else {
36
                        final NioSocketWrapper attachment = (NioSocketWrapper) key.attachment();
37
                        if (attachment != null) {
38
                            // We are registering the key to start with, reset the fairness counter.
39
                            try {
40
                                int ops = key.interestOps() | interestOps;
41
                                attachment.interestOps(ops);
42
                                key.interestOps(ops);
43
                            } catch (CancelledKeyException ckx) {
44
                                cancelledKey(key, socketWrapper);
45
                            }
46
                        } else {
47
                            cancelledKey(key, socketWrapper);
48
                        }
49
                    }
50
                }
51
                if (running && !paused && eventCache != null) {
52
                    pe.reset();//清空任务socketWrapper
53
                    eventCache.push(pe);
54
                }
55
            }
56

57
            return result;
58
        }

1
protected boolean setSocketOptions(SocketChannel socket) {
2
        NioSocketWrapper socketWrapper = null;
3
        try {
4
            // Allocate channel and wrapper
5
            NioChannel channel = null;
6
            if (nioChannels != null) {
7
                channel = nioChannels.pop();
8
            }
9
            //...... 部分省略
10
            NioSocketWrapper newWrapper = new NioSocketWrapper(channel, this);
11

12
            socketWrapper.setKeepAliveLeft(NioEndpoint.this.getMaxKeepAliveRequests());
13
            poller.register(socketWrapper);
14
            return true;
15
        } catch (Throwable t) {
16
            ExceptionUtils.handleThrowable(t);
17
            try {
18
                log.error(sm.getString("endpoint.socketOptionsError"), t);
19
            } catch (Throwable tt) {
20
                ExceptionUtils.handleThrowable(tt);
21
            }
22
            if (socketWrapper == null) {
23
                destroySocket(socket);
24
            }
25
        }