Tomcat7.0.28源代码浅读1

 

1.接收Http连接

 

处理连接涉及的主要类是

org.apache.tomcat.util.net.JIoEndpoint的内部类Acceptor。JIoEndpoint在启动之初会调用它的startInternal()方法，之后会有单独的线程负责激活Acceptor。

    /**
     * 监听TCP/IP连接，并处理他们，分发到HTTP的处理器
     */
    protected class Acceptor extends AbstractEndpoint.Acceptor {

        @Override
        public void run() {

            int errorDelay = 0;
            
            //循环接收tcp-ip连接，直到收到关闭命令
            while (running) {

                // 暂停
                while (paused && running) {
                    state = AcceptorState.PAUSED;
                    try {
                        Thread.sleep(50);
                    } catch (InterruptedException e) {
                        // Ignore
                    }
                }
                
                // 停止
                if (!running) {
                    break;
                }
                state = AcceptorState.RUNNING;

                try {
                	
                	//如果当前处理的tcp请求连接的数目已经很多了，等待（默认是200个）
                    countUpOrAwaitConnection();

                    Socket socket = null;
                    try {
                    	
                    	//等待接收下一个tcp请求
                        socket = serverSocketFactory.acceptSocket(serverSocket);
                    } catch (IOException ioe) {
                        countDownConnection();
                        // Introduce delay if necessary
                        errorDelay = handleExceptionWithDelay(errorDelay);
                        // re-throw
                        throw ioe;
                    }
                    // 成功接受了tcp请求
                    errorDelay = 0;

                    // Configure the socket
                    if (running && !paused && setSocketOptions(socket)) {
                    	//用适当的处理器处理Runnable请求
                        if (!processSocket(socket)) {
                        	
                        	//计数器-1，释放正在处理connection的数字
                            countDownConnection();
                            
                            // 关闭tcp连接
                            closeSocket(socket);
                        }
                    } else {
                        countDownConnection();
                        // Close socket right away
                        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;
        }
    }

 

内部类用一个独立的线程运行方式，运行。在没有收到“关闭”状态命令的时候，一直处理接收tcp-ip请求。

 

该类最重要的方法是countUpOrAwaitConnection();和processSocket(socket)。前者是控制接收tcp-ip并发连接数的等待机制，后面是处理socket的具体逻辑。

countUpOrAwaitConnection()的方法逻辑如下：

    protected void countUpOrAwaitConnection() throws InterruptedException {
    	
    	//最大连接数是非法数字
        if (maxConnections==-1) return;
        
        //占有锁的钥匙
        LimitLatch latch = connectionLimitLatch;
        
        //获取一个队列中的可用资源，如果当前没有可用的资源，则一直处于等待
        if (latch!=null) latch.countUpOrAwait();
    }

 

其中，LimitLatch是关键的类。该类的内部类Sync又是一个重点。它是AbstractQueuedSynchronizer类的子类，用于同步队列的实现。LimitLatch.countUpOrAwait()方法就是调用了

 

sync.acquireSharedInterruptibly(1);

 

重点就是Sync类以及它的父类AbstractQueuedSynchronizer的解释了

Sync实现，API的解释在注释中

		/**
		 * 在失败时返回负值；如果共享模式下的获取成功但其后续共享模式下的获取不能成功，则返回
		 * 0；如果共享模式下的获取成功并且其后续共享模式下的获取可能够成功
		 * ，则返回正值，在这种情况下，后续等待线程必须检查可用性。在成功的时候，此对象已被获取。
		 */
		@Override
		protected int tryAcquireShared(int ignored) {
			
			//当前新的计数数字
			long newCount = count.incrementAndGet();
			if (!released && newCount > limit) {
				
				// 已经越过最小值
				count.decrementAndGet();
				return -1;
			} else {
				return 1;
			}
		}

 释放计数器，实现如下

		@Override
		protected boolean tryReleaseShared(int arg) {
			
			//原子量减1
			count.decrementAndGet();
			return true;
		}

 

下面我们弱弱的看一看AbstractQueuedSynchronizer.acquireSharedInterruptibly方法的实现

    public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();

//尝试调用子类的tryAcquireShared方法，获取失败了走doAcquireSharedInterruptibly逻辑
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }

 

AbstractQueuedSynchronizer维护了一个双向的队列结构，FIFO共享方式，对外进行信号量的加锁，解锁操作。获取不到>0的信号量，那么请求线程会加入到等待队列中等待，知道别的线程释放了信号量（减1），再次获取后，方可进行下一步的服务。

 

这样，完成了多线程同时发起http请求的同步和阻塞（达到最大http连接数后）机制。

 

留个标记：AbstractQueuedSynchronizer值得深入研究

处理socket是JIoEndpoint.processSocket(socket)这句。核心方法体如下

        	//包装一下Socket
            SocketWrapper<Socket> wrapper = new SocketWrapper<Socket>(socket);
            
            //HTTP1.1 的最新特性，维护100个http请求，只使用同一个httpconnection
            wrapper.setKeepAliveLeft(getMaxKeepAliveRequests());
            // During shutdown, executor may be null - avoid NPE
            if (!running) {
                return false;
            }
            
            //线程池执行一个“处理http socket”的线程
            getExecutor().execute(new SocketProcessor(wrapper));

 

在SocketProcessor中就是多次处理Socket请求，到后面还是调用AbstractConnectionHandler类的process方法。

 

关于tomcat接收Http连接的问题。我们看到了，是以上的工作机制。

简单的模块图如下