NIOEndPoint
NIOEndPoint的bind()方法开启一个SocketServer
@Override
public void bind() throws Exception {
//开启一个server socket
serverSock = ServerSocketChannel.open();
//根据配置文件设置server socket的属性
socketProperties.setProperties(serverSock.socket());
InetSocketAddress addr = (getAddress()!=null?new InetSocketAddress(getAddress(),getPort()):new InetSocketAddress(getPort()));
serverSock.socket().bind(addr,getAcceptCount());
serverSock.configureBlocking(true); //mimic APR behavior
// Initialize thread count defaults for acceptor, poller
if (acceptorThreadCount == 0) {
// FIXME: Doesn't seem to work that well with multiple accept threads
// 这个东西和下面的配置有关
acceptorThreadCount = 1;
}
if (pollerThreadCount <= 0) {
//minimum one poller thread
pollerThreadCount = 1;
}
stopLatch = new CountDownLatch(pollerThreadCount);
// 如果需要的话，初始化SSL
initialiseSsl();
//开启Selector，坚挺NIO的 IO的事件
selectorPool.open();
}
·
Acceptor线程接收客户请求
在Tomcat启动的时候会启动一个Endpoint,并会调用它的startInternal方法，在这里开启了一个Acceptor的子线程。利用这个Acceptor子线程来接收Client端的Socket连接
@Override
public void run() {
........................省略代码................................................
// 这里调用上面NIOEndPoint 的serverSocketChannel 来接收一个客户端发送来的socket
socket = endpoint.serverSocketAccept();
........................省略代码................................................
// setSocketOptions() will hand the socket off to
// an appropriate processor if successful
if (!endpoint.setSocketOptions(socket)) {
//及时关闭Socket连接
endpoint.closeSocket(socket);
}
}
从Tomcat8 以后，Tomcat的默认连接为NIO。所以在这里的EndPoint的具体实现是NioEndPoint
@Override
protected boolean setSocketOptions(SocketChannel socket) {
// Process the connection
try {
//disable blocking, APR style, we are gonna be polling it
socket.configureBlocking(false);
Socket sock = socket.socket();
socketProperties.setProperties(sock);
NioChannel channel = nioChannels.pop();
if (channel == null) {
SocketBufferHandler bufhandler = new SocketBufferHandler(
socketProperties.getAppReadBufSize(),
socketProperties.getAppWriteBufSize(),
socketProperties.getDirectBuffer());
if (isSSLEnabled()) {
channel = new SecureNioChannel(socket, bufhandler, selectorPool, this);
} else {
// NIOChannel 实质上对ByteChannel 的一个封装实现
channel = new NioChannel(socket, bufhandler);
}
} else {
// 根据Socket，对设置当前的NioChannel
channel.setIOChannel(socket);
channel.reset();
}
getPoller0().register(channel);
} catch (Throwable t) {
ExceptionUtils.handleThrowable(t);
try {
log.error("",t);
} catch (Throwable tt) {
ExceptionUtils.handleThrowable(tt);
}
// Tell to close the socket
return false;
}
return true;
}
通过阅读代码可以看出其处理过程如下：
设置非阻塞，以及其他的一些参数如SoTimeout、ReceiveBufferSize、SendBufferSize
然后将SocketChannel封装成一个NioChannel，封装过程使用了缓存，即避免了重复创建NioChannel，封装过程中使用了缓存，既避免了重复创建NioChannel对象，直接利用原有的NIOChannel，并将NioChannel中的数据全部清空。
选择一个Poller进行注册
再来看一下Poller
public void register(final NioChannel socket) {
socket.setPoller(this);
NioSocketWrapper ka = new NioSocketWrapper(socket, NioEndpoint.this);
socket.setSocketWrapper(ka);
ka.setPoller(this);
ka.setReadTimeout(getConnectionTimeout());
ka.setWriteTimeout(getConnectionTimeout());
ka.setKeepAliveLeft(NioEndpoint.this.getMaxKeepAliveRequests());
ka.setSecure(isSSLEnabled());
PollerEvent r = eventCache.pop();
ka.interestOps(SelectionKey.OP_READ);//this is what OP_REGISTER turns into.
if ( r==null) r = new PollerEvent(socket,ka,OP_REGISTER);
else r.reset(socket,ka,OP_REGISTER); //
addEvent(r); //将event 加入到事件处理队列中
}
这里又是进行一些参数包装，将socket和Poller的关系绑定，再次从缓存中取出或者重新构建一个PollerEvent，然后将该event放到Poller的事件队列中等待被异步处理
再来看看对于被加入到队列中的events的处理
public boolean events() {
boolean result = false;
PollerEvent pe = null;
while ( (pe = events.poll()) != null ) {
result = true;
try {
pe.run(); //开始处理这个pollerEvent
pe.reset(); //
if (running && !paused) {
eventCache.push(pe);
}
} catch ( Throwable x ) {
log.error("",x);
}
}
return result;
}
Poller的run 方法会对
@Override
public void run() {
.........................省略代码................................
// 遍历并处理 已经准备好的NIO 事件
while (iterator != null && iterator.hasNext()) {
SelectionKey sk = iterator.next();
NioSocketWrapper attachment = (NioSocketWrapper)sk.attachment();
if (attachment == null) {
iterator.remove();
} else {
iterator.remove();
//将已经准备好的IO事件和 绑定的Socket进行分类处理
processKey(sk, attachment);
}
}//while
............................省略代码.............
}
protected void processKey(SelectionKey sk, NioSocketWrapper attachment) {
try {
if ( close ) {
cancelledKey(sk);
} else if ( sk.isValid() && attachment != null ) {
if (sk.isReadable() || sk.isWritable() ) {
if ( attachment.getSendfileData() != null ) {
processSendfile(sk,attachment, false);
} else {
//为了避免多线程对于attach的socketChannel处理的冲突，在这里将socketChannel的
// ready 操作位取反
unreg(sk, attachment, sk.readyOps());
boolean closeSocket = false;
// 如果SelectorKey 为Read，那就去处理Read
if (sk.isReadable()) {
if (!processSocket(attachment, SocketEvent.OPEN_READ, true)) {
closeSocket = true;
}
}
// 如果SelectorKey 为Read，那就去处理Write
if (!closeSocket && sk.isWritable()) {
if (!processSocket(attachment, SocketEvent.OPEN_WRITE, true)) {
closeSocket = true;
}
}
if (closeSocket) {
// 将attach的socketChannel从key中解绑
cancelledKey(sk);
}
}
}
} else {
//invalid key
cancelledKey(sk);
}
} catch ( CancelledKeyException ckx ) {
cancelledKey(sk);
} catch (Throwable t) {
ExceptionUtils.handleThrowable(t);
log.error("",t);
}
}
/**
* Process the given SocketWrapper with the given status. Used to trigger
* processing as if the Poller (for those endpoints that have one)
* selected the socket.
*
* @param socketWrapper The socket wrapper to process
* @param event The socket event to be processed
* @param dispatch Should the processing be performed on a new
* container thread
*
* @return if processing was triggered successfully
*/
public boolean processSocket(SocketWrapperBase<S> socketWrapper,
SocketEvent event, boolean dispatch) {
try {
if (socketWrapper == null) {
return false;
}
SocketProcessorBase<S> sc = processorCache.pop();
if (sc == null) {
sc = createSocketProcessor(socketWrapper, event);
} else {
sc.reset(socketWrapper, event);
}
Executor executor = getExecutor();
if (dispatch && executor != null) {
//将SocketProcessorBase交给线程池中的线程来处理
executor.execute(sc);
} else {
sc.run();
}
} catch (RejectedExecutionException ree) {
getLog().warn(sm.getString("endpoint.executor.fail", socketWrapper) , ree);
return false;
} catch (Throwable t) {
ExceptionUtils.handleThrowable(t);
// This means we got an OOM or similar creating a thread, or that
// the pool and its queue are full
getLog().error(sm.getString("endpoint.process.fail"), t);
return false;
}
return true;
}
线程池中的线程即为 NioEndPonint$SocketProcessor 内部类来看一下其内部的doRun()方法
doRun{
.....................
//如果握手已经完成....
if (handshake == 0) {
SocketState state = SocketState.OPEN;
// Process the request from this socket
if (event == null) {
state = getHandler().process(socketWrapper, SocketEvent.OPEN_READ);
} else {
// 这里是关键
state = getHandler().process(socketWrapper, event);
}
if (state == SocketState.CLOSED) {
close(socket, key);
}
}
....................
}
从上面代码中可以看出，注意它调用了hanler.process(socket)来生成响应数据。并且根据处理之后的返回状态来决定是否关闭Socket连接
对于handler.process(socket)的处理。
NioEndpoint类中的Handler接口的具体实现是静态类 AbstractProtocol$ConnectionHandler<S>通过查看其process(socket)方法.这个方法逻辑很长并且很复杂。说多了也没用。
它在这个方法里做的动作
- 通过协议类型得到相应的Socket的 Processor,并cache起来,在这里这个Processor就是
Http11Processor - 通过一个ThreadLocal 类标识现在的这个线程正在处理一个请求
- 调用Processor的process方法。
上面的Processor的process方法通过抽象类间接的调用了Http11Processor的service()方法。这个service()方法也是相当复杂`
它主要完成的动作有
- 填充Request，Reponse属性
- 调用
CoyoteAdapter的service()方法
通过以上不走，一个请求连接就从Connector走到了Container
小结：
实现一个tomcat连接器Connector就是实现ProtocolHander接口的过程。Connector用来接收Socket Client端的请求，通过内置的线程池去调用Servlet Container生成响应结果，并将响应结果同步或异步的返回给Socket Client。在第三方应用集成tomcat作为Web容器时，一般不会动Servlet Container端的代码，那么connector的性能将是整个Web容器性能的关键。
Tomcat 对于Socket的处理.png
