dopost 发表于 2017-2-10 08:15:08

Tomcat源码分析(三)------ 可携带状态的线程池 (转)

最近想实现一个可携带状态的线程池,具体需求就是池中的线程被用来处理某种信息,而此信息可视为线程所依赖的外部状态。如果用简单的线程池来实现,线程初始化时就得赋予某些信息,使得线程无法被再次利用。在看老版Tomcat的源码时,找到了答案,其实现思路主要是利用了线程的等待和唤起,HttpProcessor的实现正好基于此思路,时序图如下所示:



初始化HttpProcessor线程时,没法赋予所需的Socket对象,因为如果在初始化阶段就赋予Socket会导致此线程没法回收用来处理其他Socket。因此,在HttpProcessor的run阶段,先把线程给wait住,具体在await方法里体现,代码如下所示:

/**
* Await a newly assigned Socket from our Connector, or <code>null</code>
* 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);   

}
/**
* Await a newly assigned Socket from our Connector, or <code>null</code>
* 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.assign(socket)方法时,会给此线程赋予Socket对象,并唤起此线程,使其继续执行,assign方法的源码如下所示:

/**
* Process an incoming TCP/IP connection on the specified socket.Any
* exception that occurs during processing must be logged and swallowed.
* <b>NOTE</b>: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");   

}
/**
* Process an incoming TCP/IP connection on the specified socket.Any
* exception that occurs during processing must be logged and swallowed.
* <b>NOTE</b>: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");

}

线程被唤起和赋予socket对象后,继续执行核心的process方法,HttpProcessor.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();   
      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();   
    }   

}



本文来自CSDN博客,转载请标明出处:http://blog.csdn.net/cutesource/archive/2009/12/26/5081916.aspx
页: [1]
查看完整版本: Tomcat源码分析(三)------ 可携带状态的线程池 (转)