tomcat的IO模型

Java I/O 模型

• 用户线程等待内核将数据从网卡拷贝到内核空间。

• 内核将数据从内核空间拷贝到用户空间。

操作系统相关知识

进程和线程

ssize_t read(int fd,void *buf,size_t nbyte)

阻塞与唤醒

一次 Socket read 系统调用的过程

非阻塞I/O: NioEndpoint 组件

LimitLatch

public class LimitLatch {
    private class Sync extends AbstractQueuedSynchronizer {
     
        @Override
        protected int tryAcquireShared() {
            long newCount = count.incrementAndGet();
            if (newCount > limit) {
                count.decrementAndGet();
                return -1;
            } else {
                return 1;
            }
        }
 
        @Override
        protected boolean tryReleaseShared(int arg) {
            count.decrementAndGet();
            return true;
        }
    }
 
    private final Sync sync;
    private final AtomicLong count;
    private volatile long limit;
    
    // 线程调用这个方法来获得接收新连接的许可，线程可能被阻塞
    public void countUpOrAwait() throws InterruptedException {
      sync.acquireSharedInterruptibly(1);
    }
 
    // 调用这个方法来释放一个连接许可，那么前面阻塞的线程可能被唤醒
    public long countDown() {
      sync.releaseShared(0);
      long result = getCount();
      return result;
   }
}

Acceptor

serverSock = ServerSocketChannel.open();
serverSock.socket().bind(addr,getAcceptCount());
serverSock.configureBlocking(true);

Poller

private final SynchronizedQueue<PollerEvent> events = new SynchronizedQueue<>();

SocketProcessor

Executor

高并发思路

异步I/O : Nio2Endpoint

Java NIO.2 回顾

public class Nio2Server {
 
   void listen(){
      //1. 创建一个线程池
      ExecutorService es = Executors.newCachedThreadPool();
 
      //2. 创建异步通道群组
      AsynchronousChannelGroup tg = AsynchronousChannelGroup.withCachedThreadPool(es, 1);
      
      //3. 创建服务端异步通道
      AsynchronousServerSocketChannel assc = AsynchronousServerSocketChannel.open(tg);
 
      //4. 绑定监听端口
      assc.bind(new InetSocketAddress(8080));
 
      //5. 监听连接，传入回调类处理连接请求
      assc.accept(this, new AcceptHandler()); 
   }
}

1. 创建一个线程池，这个线程池用来执行来自内核的回调请求。

2. 创建一个 AsynchronousChannelGroup，并绑定一个线程池。

3. 创建 AsynchronousServerSocketChannel，并绑定到 AsynchronousChannelGroup。

4. 绑定一个监听端口。

5. 调用 accept 方法开始监听连接请求，同时传入一个回调类去处理连接请求。请你注意，accept 方法的第一个参数是 this 对象，就是 Nio2Server 对象本身，我在下文还会讲为什么要传入这个参数。

//AcceptHandler 类实现了 CompletionHandler 接口的 completed 方法。它还有两个模板参数，第一个是异步通道，第二个就是 Nio2Server 本身
public class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel, Nio2Server> {
 
   // 具体处理连接请求的就是 completed 方法，它有两个参数：第一个是异步通道，第二个就是上面传入的 NioServer 对象
   @Override
   public void completed(AsynchronousSocketChannel asc, Nio2Server attachment) {      
      // 调用 accept 方法继续接收其他客户端的请求
      attachment.assc.accept(attachment, this);
      
      //1. 先分配好 Buffer，告诉内核，数据拷贝到哪里去
      ByteBuffer buf = ByteBuffer.allocate(1024);
      
      //2. 调用 read 函数读取数据，除了把 buf 作为参数传入，还传入读回调类
      channel.read(buf, buf, new ReadHandler(asc)); 
 
}

public interface CompletionHandler<V,A> {
 
    void completed(V result, A attachment);
 
    void failed(Throwable exc, A attachment);
}

public class ReadHandler implements CompletionHandler<Integer, ByteBuffer> {   
    // 读取到消息后的处理  
    @Override  
    public void completed(Integer result, ByteBuffer attachment) {  
        //attachment 就是数据，调用 flip 操作，其实就是把读的位置移动最前面
        attachment.flip();  
        // 读取数据
        ... 
    }  
 
    void failed(Throwable exc, A attachment){
        ...
    }
}

Nio2Endpoint

Nio2Acceptor

serverSock.accept(null, this);

protected class Nio2Acceptor extends Acceptor<AsynchronousSocketChannel>
    implements CompletionHandler<AsynchronousSocketChannel, Void> {
    
@Override
public void completed(AsynchronousSocketChannel socket,
        Void attachment) {
        
    if (isRunning() && !isPaused()) {
        if (getMaxConnections() == -1) {
            // 如果没有连接限制，继续接收新的连接
            serverSock.accept(null, this);
        } else {
            // 如果有连接限制，就在线程池里跑 Run 方法，Run 方法会检查连接数
            getExecutor().execute(this);
        }
        // 处理请求
        if (!setSocketOptions(socket)) {
            closeSocket(socket);
        }
    } 
}

• 如果没有连接限制，继续在本线程中调用 accept 方法接收新的连接。

• 如果有连接限制，就在线程池里跑 run 方法去接收新的连接。那为什么要跑 run 方法呢，因为在 run 方法里会检查连接数，当连接达到最大数时，线程可能会被 LimitLatch 阻塞。为什么要放在线程池里跑呢？这是因为如果放在当前线程里执行，completed 方法可能被阻塞，会导致这个回调方法一直不返回。

Nio2SocketWrapper

• 第一次 read 调用：连接刚刚建立好后，Acceptor 创建 SocketProcessor 任务类交给线程池去处理，Http11Processor 在处理请求的过程中，会调用 Nio2SocketWrapper 的 read 方法发出第一次读请求，同时注册了回调类 readCompletionHandler，因为数据没读到，Http11Processor 把当前的 Nio2SocketWrapper 标记为数据不完整
接着 SocketProcessor 线程被回收，Http11Processor 并没有阻塞等待数据。
这里请注意，Http11Processor 维护了一个 Nio2SocketWrapper 列表，也就是维护了连接的状态。

• 第二次 read 调用：当数据到达后，内核已经把数据拷贝到 Http11Processor 指定的 Buffer 里，同时回调类 readCompletionHandler 被调用，在这个回调处理方法里会重新创建一个新的 SocketProcessor 任务来继续处理这个连接，而这个新的 SocketProcessor 任务类持有原来那个 Nio2SocketWrapper，这一次 Http11Processor 可以通过 Nio2SocketWrapper 读取数据了，因为数据已经到了应用层的 Buffer。

this.readCompletionHandler = new CompletionHandler<Integer, SocketWrapperBase<Nio2Channel>>() {
    public void completed(Integer nBytes, SocketWrapperBase<Nio2Channel> attachment) {
        ...
        // 通过附件类 SocketWrapper 拿到所有的上下文
        Nio2SocketWrapper.this.getEndpoint().processSocket(attachment, SocketEvent.OPEN_READ, false);
    }
 
    public void failed(Throwable exc, SocketWrapperBase<Nio2Channel> attachment) {
        ...
    }
}

Tomcat APR 优化

The APR based Apache Tomcat Native library which allows optimal performance in production environments was not found on the java.library.path: ***

AprEndpoint 工作过程

Acceptor

public class Socket {
  ...
  // 用 native 修饰这个方法，表明这个函数是 C 语言实现
  public static native long create(int family, int type,
                                 int protocol, long cont)
                                 
  public static native int bind(long sock, long sa);
  
  public static native int listen(long sock, int backlog);
  
  public static native long accept(long sock)
}

// 注意函数的名字要符合 JNI 规范的要求
JNIEXPORT jint JNICALL 
Java_org_apache_tomcat_jni_Socket_bind(JNIEnv *e, jlong sock,jlong sa)
	{
	    jint rv = APR_SUCCESS;
	    tcn_socket_t *s = (tcn_socket_t *）sock;
	    apr_sockaddr_t *a = (apr_sockaddr_t *) sa;
	
        // 调用 APR 库自己实现的 bind 函数
	    rv = (jint)apr_socket_bind(s->sock, a);
	    return rv;
	}

Poller

APR 提升性能的秘密

JVM 堆 VS 本地内存

java my.class

// 分配 HeapByteBuffer
ByteBuffer buf = ByteBuffer.allocate(1024);
 
// 分配 DirectByteBuffer
ByteBuffer buf = ByteBuffer.allocateDirect(1024);

sendfile

1. 从磁盘读取 HTML 到内存。

2. 将这段内存的内容通过 Socket 发送出去。

• 读取文件时，首先是内核把文件内容读取到内核缓冲区。

• 如果使用 HeapByteBuffer，文件数据从内核到 JVM 堆内存需要经过本地内存中转。

• 同样在将文件内容推入网络时，从 JVM 堆到内核缓冲区需要经过本地内存中转。

• 最后还需要把文件从内核缓冲区拷贝到网卡缓冲区。

sendfile(socket, file, len);