剑指架构师系列-tomcat6通过伪异步实现connector

Posted 2020-10-06

 

首先在StandardService中start接收请求的线程，如下：

 synchronized (connectors) {
            for (int i = 0; i < connectors.length; i++) {
                try {
                    ((Lifecycle) connectors[i]).start();
                } catch (Exception e) {
                    log.error(sm.getString("standardService.connector.startFailed",connectors[i]), e);
                }
            }
        }

然后进入Connector，在这个类中调用了org.apache.coyote.http11.Http11Protocol类

protocolHandler.start();

在Http11Protocol类中又调用了org.apache.tomcat.util.net.JIoEndpoint类

endpoint.start();

下面看一下JIoEndpoint类中的start源代码，如下：

public void start() throws Exception {
		// Initialize socket if not done before
		if (!initialized) {
			init();
		}
		if (!running) {
			running = true;
			paused = false;

			// Create worker collection
			if (executor == null) {
				workers = new WorkerStack(maxThreads); // maxThreads值为200，可同时处理200个请求
			}

			// Start acceptor threads
			for (int i = 0; i < acceptorThreadCount; i++) {  // acceptorThreadCount值为1，只有一个接收请求的线程
				Thread acceptorThread = new Thread(new Acceptor(), getName() + "-Acceptor-" + i);
				acceptorThread.setPriority(threadPriority);
				acceptorThread.setDaemon(daemon);
				acceptorThread.start();
			}
		}
	}

WorkerStack类使用了定长的数组来方便存取worker,也就是真正处理请求的线程。重点看一下Acceptor这个Runnable类的实现。

/**
	 * Server socket acceptor thread.
	 */
	protected class Acceptor implements Runnable {

		/**
		 * 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 (running) {

				// Loop if endpoint is paused
				while (paused) {
					try {
						Thread.sleep(1000);
					} catch (InterruptedException e) {
						// Ignore
					}
				}

				// Accept the next incoming connection from the server socket
				try {
					Socket socket = serverSocketFactory.acceptSocket(serverSocket);  // 阻塞接收socket请求
					serverSocketFactory.initSocket(socket);
					// Hand this socket off to an appropriate processor
					if (!processSocket(socket)) {   // 处理socket请求
						// Close socket right away
						try {
							socket.close();  
						} catch (IOException e) {
							// Ignore
						}
					}
				} catch (IOException x) {
					if (running)
						log.error(sm.getString("endpoint.accept.fail"), x);
				} catch (Throwable t) {
					log.error(sm.getString("endpoint.accept.fail"), t);
				}
				// The processor will recycle itself when it finishes ???
			}
		}// end run

	}

 最重要的就是processSocket()方法了，看下源代码：

protected boolean processSocket(Socket socket) {
		try {
			if (executor == null) {  // 默认情况
				getWorkerThread().assign(socket);
			} else {  // 用户自己指定了执行任务的线程池
				executor.execute(new SocketProcessor(socket));
			}
		} catch (Throwable t) {
			// This means we got an OOM or similar creating a thread, or that
			// the pool and its queue are full
			log.error(sm.getString("endpoint.process.fail"), t);
			return false;
		}
		return true;
	}

在默认情况下，首先要get到一个Worker的Thread，然后才能assign任务。

看一下getWorkerThread()这条逻辑：

/**
	 * Return a new worker thread, and block while to worker is available.
	 */
	protected Worker getWorkerThread() {
		// Allocate a new worker thread
		synchronized (workers) {   // 获取workers锁
			Worker workerThread;
			while ((workerThread = createWorkerThread()) == null) {
				try {
					workers.wait();  // 没有可用线程时释放workers锁，等待notify
				} catch (InterruptedException e) {
					// Ignore
				}
			}
			return workerThread;
		}
	}

代码要通过createWorkerThread()方法来获取一个workerThread，阅读如下代码就可以知道，这个方法有可能返回null。这样这个线程就需要让锁等待了，直到有线程notify。想一下就知道，肯定是分配出去执行任务的线程执行完成后，就可以notify接口请求的线程。接收请求的线程继续while循环，直到获取到一个workerThread为止。

createWorkerThread()方法源代码：

protected Worker createWorkerThread() {

		synchronized (workers) {
			if (workers.size() > 0) { // 通过WorkerStack提供的方法来操作Worker
				curThreadsBusy++;
				return workers.pop();
			}
			if ((maxThreads > 0) && (curThreads < maxThreads)) { // 保证不能大于指定的最大线程数
				curThreadsBusy++;
				if (curThreadsBusy == maxThreads) {
					log.info(sm.getString("endpoint.info.maxThreads", Integer.toString(maxThreads), address,Integer.toString(port)));
				}
				return (newWorkerThread());
			} else {  
				if (maxThreads < 0) {  // maxThreads小于0时会无限制的new WorkerThread，表示不限制
					curThreadsBusy++;
					return (newWorkerThread());
				} else {  // 当curThreads等于maxThreads或者大于maxThreads且maxThreads大于0的情况
					return (null);
				}
			}
		}

	}　　

recycleWorkerThread()方法源代码：

protected void recycleWorkerThread(Worker workerThread) {
		synchronized (workers) {
			workers.push(workerThread);
			curThreadsBusy--;
			workers.notify();
		}
	}

这个方法被谁调用了呢？当然是被执行任何的线程调用了。　　

下面来看一下最重要的Worker类中非常重要的几个方法，如下：

protected class Worker implements Runnable {

		protected Thread thread = null;
		protected boolean available = false;  // available初始化为false
		protected Socket socket = null;

		/**
		 * 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 (running) {

				// Wait for the next socket to be assigned
				Socket socket = await(); // 1
				if (socket == null)
					continue;

				// Process the request from this socket
				if (!setSocketOptions(socket) || !handler.process(socket)) {
					// Close socket
					try {
						socket.close();
					} catch (IOException e) {
					}
				}
				// Finish up this request
				socket = null;
				recycleWorkerThread(this);
			}

		}

		/**
		 * Start the background processing thread.
		 */
		public void start() {
			thread = new Thread(this);
			thread.setName(getName() + "-" + (++curThreads));
			thread.setDaemon(true);
			thread.start();
		}

	}

这个线程在assign任务之前是start的，看一下run()方法中的第1步调用了await()方法，在await()方法中由于available值默认为false，所以进入了while循环后让出了线程锁并等待assign()方法notifyAll()。

/**
		 * 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();

			return (socket);

		}	

当我们assign任务后，调用的assign()方法如下：

/**
		 * 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.
		 */
		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();

		}

没有进入while循环，置available为true后notifyAll()。这样await()方法就跳出循环并置available为false后返回一个局部变量socket（为什么要返回一个局部变量socket呢？），这样run()方法就可以开始往下走了，完成后调用recycleWorkerThread()方法进行线程回收。　

这个run()方法再次进入while循环，调用await()方法后，由于await()方法在之前跳出循环时将available设置为false，所以就进入了让锁等待，等待请求线程调用assign()方法指定任务，这样就回到了开始叙述的地方了。

为什么在await()方法中使用局部变量socket呢？

摘自深入剖析Tomcat：因为使用局部变量可以在当前Socket对象处理完之前，继续接收下一个Socket对象。　

个人认为是怕在run()方法运行的过程中其它线程调用这个Worker对象的assign()方法，毕竟这个对象的引用是可以被其它线程获取到的。为什么可以调用assign()方法重新指定呢？因为run()方法没有加synchronized关键字，所以不能与assign()方法互斥访问socket资源。还是为了安全性吧。