tomcat源码解析(三)--请求过程之数据的接收

本章只分析Http11NioProtocol处理请求的过程,该方法也是目前我分析的版本默认的处理方式.

根据第一章的分析知道会在StandardService类的startInternal方法方法里面启动监听,部分代码如下:

@Override
protected void startInternal() throws LifecycleException {

......

synchronized (executors) {
for (Executor executor: executors) {
executor.start();
}
}

......

synchronized (connectorsLock) {
for (Connector connector: connectors) {
......
connector.start();
......
}
}
}

在这里需要注意的是也启动的线程池,该线程城池将用于处理tomcat下的所有连接请求.好了接着看到Connector 里面的start方法,因为Connector继承了LifecycleMBeanBase,所以最终会调用它的startInternal方法,部分代码如下:

@Override
protected void startInternal() throws LifecycleException {
......
protocolHandler.start();
......
}

这里的protocolHandler是什么呢,好了接着看它的定义.找到Connector的构造方法,部分代码如下:

public Connector() {
this(null);
}
public Connector(String protocol) {
setProtocol(protocol);
......
ProtocolHandler p = null;
try {
Class<?> clazz = Class.forName(protocolHandlerClassName);
p = (ProtocolHandler) clazz.newInstance();
} catch (Exception e) {
......
} finally {
this.protocolHandler = p;
}
......
}

从构造函数this.protocolHandler = p;在这里通过反射创建了protocolHandler 的实例,那么现在要去找protocolHandlerClassName的值.看到setProtocol方法,部分代码如下:

public void setProtocol(String protocol) {
......

if ("HTTP/1.1".equals(protocol) || protocol == null) {
if (aprConnector) {
setProtocolHandlerClassName("org.apache.coyote.http11.Http11AprProtocol");
} else {
setProtocolHandlerClassName("org.apache.coyote.http11.Http11NioProtocol");
}
} else if ("AJP/1.3".equals(protocol)) {
if (aprConnector) {
setProtocolHandlerClassName("org.apache.coyote.ajp.AjpAprProtocol");
} else {
setProtocolHandlerClassName("org.apache.coyote.ajp.AjpNioProtocol");
}
} else {
setProtocolHandlerClassName(protocol);
}

}

注意通过Server.xml创建的时候,

<Connector port="8080" protocol="HTTP/1.1"
connectionTimeout="20000"
redirectPort="8443">

是没有调用setProtocolHandlerClassName设置任何值的因此,Connector 的protocolHandlerClassName属性使用的是默认值,

protected String protocolHandlerClassName =
"org.apache.coyote.http11.Http11NioProtocol";

好了接着分析setProtocol方法,因为该方法的protocol等于null,并且aprConnector的值为false,所以最后protocolHandlerClassName的值还是设置为org.apache.coyote.http11.Http11NioProtocol.

那么现在就去看一下Http11NioProtocol的做了什么吧.

看到该类的构造方法:

public Http11NioProtocol() {
super(new NioEndpoint());
}

好了回到Connector类的startInternal方法中,知道该方法调用了Http11NioProtocol的start方法,分析该类的时候它没有start方法,于是乎找到了它的父类AbstractProtocol的start方法里面,部分代码如下:

@Override
public void start() throws Exception {
......
endpoint.start();
......

// 这个是1秒检测servlet的异步请求有没有死掉或者超时
asyncTimeout = new AsyncTimeout();
Thread timeoutThread = new Thread(asyncTimeout, getNameInternal() + "-AsyncTimeout");
timeoutThread.setPriority(endpoint.getThreadPriority());
timeoutThread.setDaemon(true);
timeoutThread.start();
}

注意到endpoint.start();刚才在分析Http11NioProtocol的构造方法的时候,有这么一句super(new NioEndpoint());所以很容易知道,这个endpoint就是NioEndpoint类啦.那么接着看到它的start方法,该方法的定义是在其父类AbstractEndpoint里面定义的,代码如下:

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

该方法主要调用了bind和startInternal方法.先看一下bind方法,bind方法是在NioEndpoint里面定义的,部分代码如下:

@Override
public void bind() throws Exception {

serverSock = ServerSocketChannel.open();
socketProperties.setProperties(serverSock.socket());
InetSocketAddress addr = (getAddress()!=null?new InetSocketAddress(getAddress(),getPort()):new InetSocketAddress(getPort()));
serverSock.socket().bind(addr,getBacklog());
serverSock.configureBlocking(true); //mimic APR behavior
serverSock.socket().setSoTimeout(getSocketProperties().getSoTimeout());

if (acceptorThreadCount == 0) {
acceptorThreadCount = 1; // 说明accptor的线程只有一个
}
if (pollerThreadCount <= 0) {
//minimum one poller thread
pollerThreadCount = 1;
}
stopLatch = new CountDownLatch(pollerThreadCount);

// Initialize SSL if needed
initialiseSsl();  // 这个是支持SSL的

selectorPool.open(); // 开启一个selector池
}

好啦,熟悉java socket编程的话,看到该方法是不是特别熟悉呢.其实该方法就是绑定监口的.还有这里处理accept请求的线程只有一个,也许有人会问只有一条线程处理accept请求,性能会不会跟不上.其实,根据我分析多个有名网络库基本都是一条线程,更甚者把收发数据的工作也放在处理accept请求的线程里面,但是性能还是超高的.其实只要了解socket收发数据的过程,就很容易理解为什么要这么做.好啦,这方面的知识,大家可以去搜索一下.回到正题.

selectorPool.open();这个open的代码如下:

public void open() throws IOException {
enabled = true;
getSharedSelector();
if (SHARED) {
blockingSelector = new NioBlockingSelector();
blockingSelector.open(getSharedSelector());
}

}

这里SHARED的意思是,如果为真的话,那么全部的请求共用一个selector.继续分析startInternal方法,

@Override
public void startInternal() throws Exception {
......
// 这部分主要创建一些类的缓存池的
processorCache = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
socketProperties.getProcessorCache());
eventCache = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
socketProperties.getEventCache());
nioChannels = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
socketProperties.getBufferPool());

// 创建线程池
if ( getExecutor() == null ) {
createExecutor();
}
initializeConnectionLatch();

......
// getPollerThreadCount()方法返回的值就是在bind方法里面提到的pollerThreadCount
pollers = new Poller[getPollerThreadCount()];
for (int i=0; i<pollers.length; i++) {
pollers[i] = new Poller();
Thread pollerThread = new Thread(pollers[i], getName() + "-ClientPoller-"+i);
pollerThread.setPriority(threadPriority);
pollerThread.setDaemon(true);
pollerThread.start();
}
startAcceptorThreads();
}

从该方知道,根据pollerThreadCount 的值创建了对应数量的Poller线程,该线程主要是用来做什么的呢,具体过程后面分析,在这里就说一下它的作用:主要用来接收客户端发送过来的数据的,也就是read事件.

接着看到startAcceptorThreads,该方法的部分代码如下:

protected final void startAcceptorThreads() {
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();
}
}

从这里知道在bind方法里面知道acceptorThreadCount 的值为1,因此int count = getAcceptorThreadCount()的值为1,就是只创建了一条Acceptor的线程,那么看到 createAcceptor方法,内部主要是创建了Acceptor的实例,Acceptor是Runnable的子类,那么看到run方法,部分代码如下:

@Override
public void run() {
......
while (running) {

......
state = AcceptorState.RUNNING;

......
SocketChannel socket = null;
......

socket = serverSock.accept();
......

if (running && !paused) {

if (!setSocketOptions(socket)) {// 在这里处理accept请求的
countDownConnection();
closeSocket(socket);
}
} else {
countDownConnection();
closeSocket(socket);
}
......
}
}

是不是看到熟悉的代码了呢.这个就是接受socket的连接的请求的了,接着看到setSocketOptions方法,因为该类主要处理的就是连接进来之后的处理,部分代码如下:

protected boolean setSocketOptions(SocketChannel socket) {
......
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 {
channel = new NioChannel(socket, bufhandler);
}
} else {
channel.setIOChannel(socket);
channel.reset();
}
getPoller0().register(channel); // 注册到之前的poller
......
}

这个方法的实现,大家应该也很熟悉吧.在这里看到NioChannel channel = nioChannels.pop,还记得startInternal方法里面创建一些类的缓存吧,就是复用了里面的实例了.接着看到getPoller0().register(channel),看一下getPoller0()方法,部分代码如下:

public Poller getPoller0() {
int idx = Math.abs(pollerRotater.incrementAndGet()) % pollers.length;
return pollers[idx];
}

该方法就是获取startInternal创建的Poller实例,该方法目的就是把连接进来的socket均衡的分配到每一个Poller线程里面去,接着看到Poller类的register方法,代码如下:

public void register(final NioChannel socket) {
socket.setPoller(this);
NioSocketWrapper ka = new NioSocketWrapper(socket, NioEndpoint.this);
socket.setSocketWrapper(ka);
ka.setPoller(this);
ka.setReadTimeout(getSocketProperties().getSoTimeout());
ka.setWriteTimeout(getSocketProperties().getSoTimeout());
ka.setKeepAliveLeft(NioEndpoint.this.getMaxKeepAliveRequests());
ka.setSecure(isSSLEnabled());
ka.setReadTimeout(getSoTimeout());
ka.setWriteTimeout(getSoTimeout());
PollerEvent r = eventCache.pop();
ka.interestOps(SelectionKey.OP_READ);// 这里就是设置selector的监听类型为read
if ( r==null) r = new PollerEvent(socket,ka,OP_REGISTER);
else r.reset(socket,ka,OP_REGISTER);
addEvent(r);
}

其实该方法主要是做一些配置,并把请求包装成一个PollerEvent 类,扔到Poller的处理任务队列里面.

执行完register方法之后,accptor就返回继续监听端口了,后面的socket的数据接收就交给Poller线程了.

PollerEvent 也是runnable的一个具体实现,看到run方法,部分代码如下:

@Override
public void run() {
if (interestOps == OP_REGISTER) {
......
socket.getIOChannel().register(
socket.getPoller().getSelector(), SelectionKey.OP_READ, socketWrapper);
......
} else {
......
}
}

该方法主要作用就是把socket注册到一个selector里面去,并设置OP_READ.

到这Connector的startInternal方法就分析完啦.

下面接着分析有数据到来的时候是怎么处理的.从Poller的构造方法知道,每一个Poller都会拥有自己的一个Selector,代码如下

public Poller() throws IOException {
this.selector = Selector.open();
}

接着看到Poller的run方法,部分代码如下:

@Override
public void run() {
while (true) {
......
hasEvents = events();//  这个event全部处理完所有的任务
if (wakeupCounter.getAndSet(-1) > 0) {

keyCount = selector.selectNow();
} else {
keyCount = selector.select(selectorTimeout);
}
wakeupCounter.set(0); // 任务队列为0,后面添加的话需要唤醒一下

Iterator<SelectionKey> iterator =

while (iterator != null && iterator.hasNext()) {
SelectionKey sk = iterator.next();
NioSocketWrapper attachment = (NioSocketWrapper)sk.attachment();

if (attachment == null) {
iterator.remove();
} else {
iterator.remove();
processKey(sk, attachment);
}
}

......
}
}

在这里events这个方法,就处理了register添加到该队列里面的PollerEvent 任务了,过程较简单就不具体分析了.直接看到processKey的方法,部分代码如下:

protected void processKey(SelectionKey sk, NioSocketWrapper attachment) {
......
if ( isWorkerAvailable() ) {
unreg(sk, attachment, sk.readyOps());
boolean closeSocket = false;

if (sk.isReadable()) {
if (!processSocket(attachment, SocketEvent.OPEN_READ, true)) {
closeSocket = true;
}
}
if (!closeSocket && sk.isWritable()) {
if (!processSocket(attachment, SocketEvent.OPEN_WRITE, true)) {
closeSocket = true;
}
}
......
}
}

unreg(sk, attachment, sk.readyOps());

为什么要注销OP_READ的监听呢,因为每次有数据来的时候,都会把接收数据的请求放在之前创建的线程池里面接收数据,如果不注销的时候,就会造成并发接收数据的情况.

看到processSocket的方法,部分代码如下:

protected boolean processSocket(NioSocketWrapper attachment, SocketEvent status, boolean dispatch) {
......
SocketProcessor sc = processorCache.pop(); // 这里是解析器
if ( sc == null ) sc = new SocketProcessor(attachment, status);
else sc.reset(attachment, status);
Executor executor = getExecutor(); // 这里没一个请求都会放在线程池里面执行
if (dispatch && executor != null) {
executor.execute(sc);
} else {
sc.run();
}
......
}

在这里看到了吧,接收数据的时候就把任务包装成SocketProcessor 放到线程池里面处理,并没有在poller线程里面处理.接着看到SocketProcessor 类的run方法,部分代码如下:

@Override
public void run() {
NioChannel socket = ka.getSocket();
......
SelectionKey key = socket.getIOChannel().keyFor(
socket.getPoller().getSelector());
synchronized (socket) {
try {
......
if (handshake == 0) {
SocketState state = SocketState.OPEN;
if (status == null) {
state = getHandler().process(ka, SocketEvent.OPEN_READ); // 在这里处理请求了
} else {
state = getHandler().process(ka, status);
}
......
}
......
}
}
}

因为handshake 第一次有数据的时候会为0的,因此看到getHandler().process的方法.getHandler获取到的是ConnectionHandler的实例,看到该类的process方法,部分代码如下:

@Override
public SocketState process(SocketWrapperBase<S> wrapper, SocketEvent status) {
......
Processor processor = connections.get(socket);

try {
......
SocketState state = SocketState.CLOSED;
do {
state = processor.process(wrapper, status);//----------处理请求-------------------
......
} while ( state == SocketState.UPGRADING);

if (state == SocketState.LONG) {
longPoll(wrapper, processor);
if (processor.isAsync()) {
getProtocol().addWaitingProcessor(processor);
}
} else if (state == SocketState.OPEN) {
connections.remove(socket);
release(processor);
wrapper.registerReadInterest();
} else if (state == SocketState.SENDFILE) {
connections.remove(socket);
release(processor);
}
......
return state; // 返回LONG
}
......
}

这里注意到processor.process(wrapper, status)的status的值为SocketEvent.OPEN_READ,接着分析process方法,部分代码如下:

@Override
public SocketState process(SocketWrapperBase<?> socketWrapper, SocketEvent status)
throws IOException {

SocketState state = SocketState.CLOSED;
Iterator<DispatchType> dispatches = null;
do {
if (dispatches != null) {
DispatchType nextDispatch = dispatches.next();
state = dispatch(nextDispatch.getSocketStatus());
} else if (status == SocketEvent.DISCONNECT) {
} else if (isAsync() || isUpgrade() || state == SocketState.ASYNC_END) {
state = dispatch(status);
if (state == SocketState.OPEN) {
state = service(socketWrapper);
}
} else if (status == SocketEvent.OPEN_WRITE) {
state = SocketState.LONG;
} else {
state = service(socketWrapper);
}

if (state != SocketState.CLOSED && isAsync()) {
state = asyncPostProcess(); // 判断是不是异步处理额
}
......
if (dispatches == null || !dispatches.hasNext()) {
dispatches = getIteratorAndClearDispatches();
}
} while (state == SocketState.ASYNC_END ||dispatches != null && state != SocketState.CLOSED);

return state; // 返回LONG
}

在这里因为传进来的status 的值为SocketEvent.OPEN_READ,使用直接看到

else {
state = service(socketWrapper);
}

分析service方法,该方法的实现是在Http11Processor的,部分代码如下:

@Override
public SocketState service(SocketWrapperBase<?> socketWrapper)
throws IOException {
RequestInfo rp = request.getRequestProcessor();
......

// 对发送和接收缓冲区做设置
setSocketWrapper(socketWrapper);
inputBuffer.init(socketWrapper);
outputBuffer.init(socketWrapper);

keepAlive = true;
openSocket = false;
readComplete = true;
boolean keptAlive = false;

while (!getErrorState().isError() && keepAlive && !isAsync() &&
upgradeToken == null && !endpoint.isPaused()) {

if (!inputBuffer.parseRequestLine(keptAlive)) { // 在这里只要分析请求头的第一行
if (inputBuffer.getParsingRequestLinePhase() == -1) {
return SocketState.UPGRADING;
} else if (handleIncompleteRequestLineRead()) {
break;
}
}

......
keptAlive = true;

request.getMimeHeaders().setLimit(endpoint.getMaxHeaderCount());
if (!inputBuffer.parseHeaders()) { // 该方法是解析请求头第一行意外的其他请求
// 这两个值需要注意一下
openSocket = true;
readComplete = false; // 第一次连接进来是false
break;
}

......
if (!getErrorState().isError()) {
rp.setStage(org.apache.coyote.Constants.STAGE_SERVICE);
getAdapter().service(request, response); // 在这分配进去----主机匹配也是在这里----------------
......

}

......

}

if (getErrorState().isError() || endpoint.isPaused()) {
return SocketState.CLOSED;
} else if (isAsync()) {
return SocketState.LONG;
} else if (isUpgrade()) {
return SocketState.UPGRADING;
} else {
if (sendfileData != null) {
return SocketState.SENDFILE;
} else {
if (openSocket) {
if (readComplete) {
return SocketState.OPEN;
} else {
return SocketState.LONG;// readComplete = false;
}
} else {
return SocketState.CLOSED;
}
}
}
}

该方法主要做了:

1,inputBuffer.parseRequestLine(keptAlive)解析请求头的第一行信息,如GET 请求路径如/index ,还有当前的http协议.如HTTP1.1等.

2,inputBuffer.parseHeaders()解析除第一行以外的请求信息.

3,getAdapter().service(request, response);这里就是处理完整的请求了.

如果发送过来的数据不完整会怎么办呢.比如说解析第一行数据不完整的时候回怎么办

if (!inputBuffer.parseRequestLine(keptAlive)) { // 在这里只要分析请求头的第一行
if (inputBuffer.getParsingRequestLinePhase() == -1) {
return SocketState.UPGRADING;
} else if (handleIncompleteRequestLineRead()) {
break;
}
}

parseRequestLine该方法只要没有解析完第一行数据就会返回false,至于该方法的具体分析过程,后面有时间的时候话再写.因为第一次调用parseRequestLine的话getParsingRequestLinePhase()会返回1,接着就会执行handleIncompleteRequestLineRead方法了,部分代码如下:

private boolean handleIncompleteRequestLineRead() {
openSocket = true;
if (inputBuffer.getParsingRequestLinePhase() > 1) {
if (endpoint.isPaused()) {
......
return false;
} else {
readComplete = false;
}
}
return true;
}

注意到该方法把 openSocket 设置为 true了,有因为如果是第一次的话getParsingRequestLinePhase()返回的值为1,使用该方法返回的值就是true了,接着就跳出while循环了,然后service方法的返回值是什么呢?

if (getErrorState().isError() || endpoint.isPaused()) {
return SocketState.CLOSED;
} else if (isAsync()) {
return SocketState.LONG;
} else if (isUpgrade()) {
return SocketState.UPGRADING;
} else {
if (sendfileData != null) {
return SocketState.SENDFILE;
} else {
if (openSocket) {
if (readComplete) {
return SocketState.OPEN;
} else {
return SocketState.LONG;// readComplete = false;
}
} else {
return SocketState.CLOSED;
}
}
}

看到这里,以为openSocket在handleIncompleteRequestLineRead里面为设置为true了,然后readComplete为false,那返回的就是SocketState.LONG了.那么回到ConnectionHandler类的process方法看到

state = processor.process(wrapper, status);

因此这时候state 的值为SocketState.LONG,使用

if (state == SocketState.LONG) {
longPoll(wrapper, processor);
if (processor.isAsync()) {
getProtocol().addWaitingProcessor(processor);
}
} else if (state == SocketState.OPEN) {
connections.remove(socket);
release(processor);
wrapper.registerReadInterest();
} else if (state == SocketState.SENDFILE) {
connections.remove(socket);
release(processor);
}

走的是

longPoll(wrapper, processor);

这个分支.

longPoll的代码如下:

protected void longPoll(SocketWrapperBase<?> socket, Processor processor) {
if (!processor.isAsync()) { //
socket.registerReadInterest();
}
}

接着看到registerReadInterest,代码如下

@Override
public void registerReadInterest() {
getPoller().add(getSocket(), SelectionKey.OP_READ);
}

看到了吧,如果读取的数据没有读完,那么从新在之前的poller线程重新监听数据.然后有数据的话就重新之前的动作.

对tomcat处理请求过程总结一下吧,就是在poller里面监听selector事件,如何有数据要读的话就放在线程池里面处理.

好了本章就先分析到这里了,然后下一篇分析就是tomcat如何解析请求头以及如何把对应的请求发送到对应的servlet里面去.