|
tomcat如何利用多线程实现同时处理多个请求-简化版
说明以下内容以《how tomcat works》第四章为原型整体流程，交互过程以及源码可参考[https://blog.csdn.net/qq_29025955/article/details/120196616](https://blog.csdn.net/qq_29025955/article/details/120196616)
启动应用时就创建多个处理实例（线程）创建实例：执行run()方法，每个实例都执行自己的run方法（即是针对每个实例都启动一个线程）：上面代码片段中的 await()方法如下表明每个实例对应的线程初始化时都处于wait状态
客户端发送请求HttpConnector的run()方法用于监听客户端请求，同时指派HttpProcessor实例对请求进行处理。指派方法 processor.assign(socket)会唤醒在该HttpProcessor实例processor上等待的线程。notifyAll()来实现参考：[java多线程之wait()/notify/notifyAll()](https://blog.csdn.net/qq_29025955/article/details/121704930)
说明
以下内容以《how tomcat works》第四章为原型
整体流程，交互过程以及源码可参考https://blog.csdn.net/qq_29025955/article/details/120196616
启动应用时就创建多个处理实例（线程）
在应用启动时就会去创建多个HttpProcessors实例，然后启动该实例的run()方法，即是创建了多个线程并且让每个线程都处理wait状态。
创建实例：
/**
* Create and return a new processor suitable for processing HTTP
* requests and returning the corresponding responses.
*/
private HttpProcessor newProcessor() {
// if (debug >= 2)
// log("newProcessor: Creating new processor");
HttpProcessor processor = new HttpProcessor(this, curProcessors++);
if (processor instanceof Lifecycle) {
try {
((Lifecycle) processor).start();
} catch (LifecycleException e) {
log("newProcessor", e);
return (null);
}
}
created.addElement(processor);
return (processor);
}
执行run()方法，每个实例都执行自己的run方法（即是针对每个实例都启动一个线程）：
/**
* The background thread that listens for incoming TCP/IP connections and
* hands them off to an appropriate processor.
*/
public void run() {
// Process requests until we receive a shutdown signal
while (!stopped) {
// Wait for the next socket to be assigned
Socket socket = await();
System.out.println(Thread.currentThread().getName()+": i am awake!");
if (socket == null)
continue;
// Process the request from this socket
try {
process(socket);
} catch (Throwable t) {
log("process.invoke", t);
}
// Finish up this request
connector.recycle(this);
}
// Tell threadStop() we have shut ourselves down successfully
synchronized (threadSync) {
threadSync.notifyAll();
}
}
上面代码片段中的 await()方法如下表明每个实例对应的线程初始化时都处于wait状态
/**
* Await a newly assigned Socket from our Connector, or null
* if we are supposed to shut down.
*/
private synchronized Socket await() {
// Wait for the Connector to provide a new Socket
while (!available) {
try {
wait();
} catch (InterruptedException e) {
}
}
// Notify the Connector that we have received this Socket
Socket socket = this.socket;
available = false;
notifyAll();
if ((debug >= 1) && (socket != null))
log(" The incoming request has been awaited");
return (socket);
}
客户端发送请求
HttpConnector实例用于监听客户端请求，当请求到来时会将请求指派给一个具体的HttpProcessor实例进行处理，指派完成后会马上返回继续监听客户端请求，同时被指派的HttpProcessor会对请求进行处理，与HttpConnector实例相互独立运行，互不影响。 当第2个、第3个…请求到来时，HttpConnector同样会指派第2个，第3个HttpProcessor实例进行处理，这1 2 3.。。。个HttpProcessor实例（线程）可以同时独立的运行。
HttpConnector的run()方法用于监听客户端请求，同时指派HttpProcessor实例对请求进行处理。
/**
* The background thread that listens for incoming TCP/IP connections and
* hands them off to an appropriate processor.
*/
public void run() {
// Loop until we receive a shutdown command
while (!stopped) {
// Accept the next incoming connection from the server socket
Socket socket = null;
try {
// if (debug >= 3)
// log("run: Waiting on serverSocket.accept()");
socket = serverSocket.accept();
// if (debug >= 3)
// log("run: Returned from serverSocket.accept()");
if (connectionTimeout > 0)
socket.setSoTimeout(connectionTimeout);
socket.setTcpNoDelay(tcpNoDelay);
} catch (AccessControlException ace) {
log("socket accept security exception", ace);
continue;
} catch (IOException e) {
// if (debug >= 3)
// log("run: Accept returned IOException", e);
try {
// If reopening fails, exit
synchronized (threadSync) {
if (started && !stopped)
log("accept error: ", e);
if (!stopped) {
// if (debug >= 3)
// log("run: Closing server socket");
serverSocket.close();
// if (debug >= 3)
// log("run: Reopening server socket");
serverSocket = open();
}
}
// if (debug >= 3)
// log("run: IOException processing completed");
} catch (IOException ioe) {
log("socket reopen, io problem: ", ioe);
break;
} catch (KeyStoreException kse) {
log("socket reopen, keystore problem: ", kse);
break;
} catch (NoSuchAlgorithmException nsae) {
log("socket reopen, keystore algorithm problem: ", nsae);
break;
} catch (CertificateException ce) {
log("socket reopen, certificate problem: ", ce);
break;
} catch (UnrecoverableKeyException uke) {
log("socket reopen, unrecoverable key: ", uke);
break;
} catch (KeyManagementException kme) {
log("socket reopen, key management problem: ", kme);
break;
}
continue;
}
// Hand this socket off to an appropriate processor
HttpProcessor processor = createProcessor();
if (processor == null) {
try {
log(sm.getString("httpConnector.noProcessor"));
socket.close();
} catch (IOException e) {
;
}
continue;
}
// if (debug >= 3)
// log("run: Assigning socket to processor " + processor);
processor.assign(socket);
// The processor will recycle itself when it finishes
}
// Notify the threadStop() method that we have shut ourselves down
// if (debug >= 3)
// log("run: Notifying threadStop() that we have shut down");
synchronized (threadSync) {
threadSync.notifyAll();
}
}
指派方法 processor.assign(socket)会唤醒在该HttpProcessor实例processor上等待的线程。notifyAll()来实现参考：java多线程之wait()/notify/notifyAll()
/**
* Process an incoming TCP/IP connection on the specified socket. Any
* exception that occurs during processing must be logged and swallowed.
* NOTE: This method is called from our Connector's thread. We
* must assign it to our own thread so that multiple simultaneous
* requests can be handled.
*
* @param socket TCP socket to process
*/
synchronized void assign(Socket socket) {
// Wait for the Processor to get the previous Socket
while (available) {
try {
wait();
} catch (InterruptedException e) {
}
}
// Store the newly available Socket and notify our thread
this.socket = socket;
available = true;
notifyAll();
if ((debug >= 1) && (socket != null))
log(" An incoming request is being assigned");
}
|