tomcat 源码分析之 JIoEndpoint

JIoEndpoint类是监听客户端连接的类，启动时会调父类的start方法

 

public final void start() throws Exception {
        if (bindState == BindState.UNBOUND) {
            bind();
            bindState = BindState.BOUND_ON_START;
        }
        startInternal();
    }

 

 

 

bind方法

 

public void bind() throws Exception {

        //初始化接受客户端连接线程
        if (acceptorThreadCount == 0) {
            acceptorThreadCount = 1;
        }
      // 初始化最大连接
        if (getMaxConnections() == 0) {
            //没有设置最大连接就取线程池的最大线程数
            setMaxConnections(getMaxThreadsExecutor(true));
        }

        if (serverSocketFactory == null) {
           //如果是https的
            if (isSSLEnabled()) {
                serverSocketFactory =
                    handler.getSslImplementation().getServerSocketFactory(this);
            } else {
                serverSocketFactory = new DefaultServerSocketFactory(this);
            }
        }

        if (serverSocket == null) {
            try {
                if (getAddress() == null) {
                        serverSocket = serverSocketFactory.createSocket(getPort(),
                                getBacklog());
                } else {
                    //生成服务端的serviceSocket
                    serverSocket = serverSocketFactory.createSocket(getPort(),
                            getBacklog(), getAddress());
                }
            } catch (BindException orig) {
                String msg;
                if (getAddress() == null)
                    msg = orig.getMessage() + " <null>:" + getPort();
                else
                    msg = orig.getMessage() + " " +
                            getAddress().toString() + ":" + getPort();
                BindException be = new BindException(msg);
                be.initCause(orig);
                throw be;
            }
        }

    }

 startInternal方法

 

 public void startInternal() throws Exception {

        if (!running) {
            running = true;
            paused = false;

            // Create worker collection
            if (getExecutor() == null) {
                createExecutor();
            }
          //初始化存放当前连接的对象
            initializeConnectionLatch();
           //启动监听获取客户端socket并进行处理
            startAcceptorThreads();

            //处理超时的连接
            Thread timeoutThread = new Thread(new AsyncTimeout(),
                    getName() + "-AsyncTimeout");
            timeoutThread.setPriority(threadPriority);
            timeoutThread.setDaemon(true);
            timeoutThread.start();
        }
    }

 

startAcceptorThreads方法

 

protected final void startAcceptorThreads() {
        //count为1
        int count = getAcceptorThreadCount();
        acceptors = new Acceptor[count];

        for (int i = 0; i < count; i++) {
            acceptors[i] = createAcceptor();
            String threadName = getName() + "-Acceptor-" + i;
            acceptors[i].setThreadName(threadName);
            Thread t = new Thread(acceptors[i], threadName);
            t.setPriority(getAcceptorThreadPriority());
            t.setDaemon(getDaemon());
            t.start();
        }
    }

 我们看Accept实现类

protected class Acceptor extends AbstractEndpoint.Acceptor {

        @Override
        public void run() {

            int errorDelay = 0;

            while (running) {

                while (paused && running) {
                    state = AcceptorState.PAUSED;
                    try {
                        Thread.sleep(50);
                    } catch (InterruptedException e) {
                       
                    }
                }

                if (!running) {
                    break;
                }
                state = AcceptorState.RUNNING;

                try {
                    //if we have reached max connections, wait
                    countUpOrAwaitConnection();

                    Socket socket = null;
                    try {
                      
                       //获取客户端的socket
                        socket = serverSocketFactory.acceptSocket(serverSocket);
                    } catch (IOException ioe) {
                        countDownConnection();
                       
                        errorDelay = handleExceptionWithDelay(errorDelay);
                     
                        throw ioe;
                    }
                   
                    errorDelay = 0;

                    
                    if (running && !paused && setSocketOptions(socket)) {
                        //处理Socket前求
                        if (!processSocket(socket)) {
                            countDownConnection();
                           
                            closeSocket(socket);
                        }
                    } else {
                        countDownConnection();
                       
                        closeSocket(socket);
                    }
                } catch (IOException x) {
                    if (running) {
                        log.error(sm.getString("endpoint.accept.fail"), x);
                    }
                } catch (NullPointerException npe) {
                    if (running) {
                        log.error(sm.getString("endpoint.accept.fail"), npe);
                    }
                } catch (Throwable t) {
                    ExceptionUtils.handleThrowable(t);
                    log.error(sm.getString("endpoint.accept.fail"), t);
                }
            }
            state = AcceptorState.ENDED;
        }
    }