tomcat 源码分析之 JIoEndpoint

ouzhoudijie

　　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 = createAcceptor();
String threadName = getName() + "-Acceptor-" + i;
acceptors.setThreadName(threadName);
Thread t = new Thread(acceptors, 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;
}
}