Tomcat Connector の役割とパッケージ構成
Tomcat アーキテクチャにおいて、Connector コンポーネントはソケットレベルでの通信処理を担います。具体的には、クライアントからの socket 接続を受け入れ、プロトコルに基づいてリクエストを解析する役割を担います。ソースコード上では、主に org.apache.catalina.connector および org.apache.coyote パッケージ下に実装されています。
Connector は Service コンテナの子要素として定義され、Service はさらに Server コンテナに所属します。これらのインスタンス化は、server.xml 設定ファイルの記述に基づき、Catalina クラス内で Digester を利用して行われます。デフォルト設定では、HTTP リクエスト処理用と AJP プロトコル用の 2 種類の Connector が用意されています。
サポートされる Connector の種類
Tomcat が提供する主要な Connector 実装は以下の通りです。
- HTTP Connector: HTTP プロトコルを解析します。実装方式により、ブロッキング IO を使用する BIO と、ノンブロッキング IO を使用する NIO に分類されます。本稿では NIO 実装に焦点を当てます。
- AJP Connector: Apache JServ Protocol を使用し、Apache HTTP Server などのウェブサーバーと Tomcat を連携させる際に利用されます。通信効率に優れています。
- APR HTTP Connector: C 言語で実装されたネイティブライブラりを JNI 経由で呼び出します。静的コンテンツの配信性能向上を目的としています。
利用するプロトコルは、server.xml 内の protocol 属性で指定可能です。
<Connector port="8080" protocol="HTTP/1.1"
connectionTimeout="20000"
redirectPort="8443" />Connector の初期化処理
Connector クラスのコンストラクタでは、指定されたプロトコル名に基づいて適切な ProtocolHandler をインスタンス化します。
public Connector(final String protocolSpec) {
configureProtocol(protocolSpec);
ProtocolHandler handler = null;
try {
final Class<?> handlerClass = Class.forName(this.handlerClassName);
handler = (ProtocolHandler) handlerClass.getDeclaredConstructor().newInstance();
} catch (Exception ex) {
log.error("ProtocolHandler initialization failed", ex);
} finally {
this.protocolHandler = handler;
}
if (Globals.STRICT_SERVLET_COMPLIANCE) {
uriCharset = StandardCharsets.ISO_8859_1;
} else {
uriCharset = StandardCharsets.UTF_8;
}
}
private void configureProtocol(String protocol) {
boolean useApr = AprLifecycleListener.isAprAvailable() &&
AprLifecycleListener.getUseAprConnector();
if ("HTTP/1.1".equals(protocol) || protocol == null) {
if (useApr) {
setHandlerClass("org.apache.coyote.http11.Http11AprProtocol");
} else {
setHandlerClass("org.apache.coyote.http11.Http11NioProtocol");
}
} else if ("AJP/1.3".equals(protocol)) {
if (useApr) {
setHandlerClass("org.apache.coyote.ajp.AjpAprProtocol");
} else {
setHandlerClass("org.apache.coyote.ajp.AjpNioProtocol");
}
} else {
setHandlerClass(protocol);
}
}この処理により、各 Connector は固有の ProtocolHandler を保持し、特定のポートでネットワークリクエストを監視する準備が整います。ただし、実際のビジネスロジック処理は Container コンポーネントが担当します。
ライフサイクルの開始段階である startInternal メソッドにおいて、ProtocolHandler の起動が行われます。
protected void startInternal() throws LifecycleException {
if (getPort() < 0) {
throw new LifecycleException("Invalid port configuration");
}
setState(LifecycleState.STARTING);
try {
protocolHandler.start();
} catch (Exception e) {
throw new LifecycleException("ProtocolHandler start failed", e);
}
}HTTP/1.1 および NIO を使用する場合、Http11NioProtocol は org.apache.tomcat.util.net.NioEndpoint インスタンスを生成し、ポート監視およびリクエスト解析の実際の作業を委譲します。
NioEndpoint のスレッドモデル
HTTP リクエストの解析プロセスにおいて、Tomcat は以下の 3 種類のスレッドを活用します。
- Acceptor: ソケット接続の受け入れ
- Poller: 準備完了したソケットの監視（セレクタ.poll）
- Worker: 実際のリクエスト処理
1. Acceptor スレッド
Acceptor は Runnable インターフェースを実装しており、その名の通り接続受け入れを専門とします。serverSocket.accept() を呼び出して SocketChannel を取得し、それを Tomcat 独自の NioChannel へラップします。NIO を採用していても、接続受付自体はブロッキング方式で行われます。Acceptor はスレッドプールによって管理され、NioEndpoint の起動時に生成されます。
public void startInternal() throws Exception {
if (!running) {
running = true;
paused = false;
// キャッシュの初期化
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();
// Poller スレッドの起動
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();
}
}
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();
}
}Acceptor の主要なロジックは run メソッド内にあります。ここでは接続数の制限チェックを行い、ソケットを受け入れた後に設定処理へ渡します。
protected class Acceptor extends AbstractEndpoint.Acceptor {
@Override
public void run() {
int retryWait = 0;
while (this.isActive) {
while (paused && this.isActive) {
state = AcceptorState.PAUSED;
try { Thread.sleep(50); } catch (InterruptedException e) {}
}
if (!this.isActive) break;
state = AcceptorState.RUNNING;
try {
countUpOrAwaitConnection();
SocketChannel channel = null;
try {
channel = serverSocket.accept();
} catch (IOException ioe) {
countDownConnection();
if (this.isActive) {
retryWait = handleExceptionWithDelay(retryWait);
throw ioe;
} else {
break;
}
}
retryWait = 0;
if (this.isActive && !paused) {
if (!configureSocket(channel)) {
closeSocket(channel);
}
} else {
closeSocket(channel);
}
} catch (Throwable t) {
ExceptionUtils.handleThrowable(t);
log.error("Accept failed", t);
}
}
state = AcceptorState.ENDED;
}
private void closeSocket(SocketChannel channel) {
countDownConnection();
try { channel.socket().close(); } catch (IOException ioe) {}
try { channel.close(); } catch (IOException ioe) {}
}
}ソケット受け入れ後、configureSocket (元々 setSocketOptions) が呼び出され、ここで非ブロッキング設定および Poller への登録が行われます。
protected boolean configureSocket(SocketChannel channel) {
try {
channel.configureBlocking(false);
Socket sock = channel.socket();
socketProperties.setProperties(sock);
NioChannel nioChan = nioChannels.pop();
if (nioChan == null) {
SocketBufferHandler bufhandler = new SocketBufferHandler(
socketProperties.getAppReadBufSize(),
socketProperties.getAppWriteBufSize(),
socketProperties.getDirectBuffer());
if (isSSLEnabled()) {
nioChan = new SecureNioChannel(sock, bufhandler, selectorPool, this);
} else {
nioChan = new NioChannel(sock, bufhandler);
}
} else {
nioChan.setIOChannel(channel);
nioChan.reset();
}
getPoller0().register(nioChan);
} catch (Throwable t) {
ExceptionUtils.handleThrowable(t);
return false;
}
return true;
}2. Poller スレッド
Poller も Runnable を実装する NioEndpoint の内部クラスです。主な責務は、セレクタを輪詢し、読み書き準備が整ったソケットを検出することです。これにより IO のマルチプレクスが実現されます。
public Poller() throws IOException {
this.selector = Selector.open();
}Acceptor がソケットをacceptした後、register メソッドを介して PollerEvent が生成され、イベントキューに追加されます。
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());
PollerEvent r = eventCache.pop();
ka.interestOps(SelectionKey.OP_READ);
if (r == null) r = new PollerEvent(socket, ka, OP_REGISTER);
else r.reset(socket, ka, OP_REGISTER);
addEvent(r);
}Poller の run メソッドでは、イベントキューの処理とセレクタの選択処理が交互に行われます。
public void run() {
while (true) {
boolean hasEvents = false;
try {
if (!close) {
hasEvents = events();
if (wakeupCounter.getAndSet(-1) > 0) {
keyCount = selector.selectNow();
} else {
keyCount = selector.select(selectorTimeout);
}
wakeupCounter.set(0);
}
if (close) {
events();
timeout(0, false);
try { selector.close(); } catch (IOException ioe) {}
break;
}
} catch (Throwable x) {
log.error("", x);
continue;
}
if (keyCount == 0) hasEvents = (hasEvents | events());
Iterator<SelectionKey> iterator = keyCount > 0 ? selector.selectedKeys().iterator() : null;
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);
}
}
timeout(keyCount, hasEvents);
}
getStopLatch().countDown();
}イベントキュー処理 (events) では、登録イベントがセレクタへ実際に登録されます。
public boolean events() {
boolean result = false;
PollerEvent pe = null;
for (int i = 0, size = events.size(); i < size && (pe = events.poll()) != null; i++) {
result = true;
try {
pe.run();
pe.reset();
if (running && !paused) {
eventCache.push(pe);
}
} catch (Throwable x) {
log.error("", x);
}
}
return result;
}PollerEvent の実行時、ソケットチャネルがセレクタへ登録され、読込イベントが監視対象となります。
3. Worker スレッド (SocketProcessor)
Worker スレッドは実際のリクエスト処理を担当します。Poller が準備完了したソケットを検出すると、processKey を経由して処理を委譲します。
protected void processKey(SelectionKey sk, NioSocketWrapper attachment) {
try {
if (close) {
cancelledKey(sk);
} else if (sk.isValid() && attachment != null) {
if (sk.isReadable() || sk.isWritable()) {
if (attachment.getSendfileData() != null) {
processSendfile(sk, attachment, false);
} else {
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;
}
}
if (closeSocket) cancelledKey(sk);
}
}
} else {
cancelledKey(sk);
}
} catch (CancelledKeyException ckx) {
cancelledKey(sk);
} catch (Throwable t) {
log.error("", t);
}
}processSocket メソッドでは、スレッドプールから SocketProcessor を取得し、タスクとして実行します。
public boolean processSocket(SocketWrapperBase<S> socketWrapper,
SocketEvent event, boolean dispatch) {
try {
if (socketWrapper == null) return false;
SocketProcessorBase<S> sc = processorCache.pop();
if (sc == null) {
sc = createSocketProcessor(socketWrapper, event);
} else {
sc.reset(socketWrapper, event);
}
Executor executor = getExecutor();
if (dispatch && executor != null) {
executor.execute(sc);
} else {
sc.run();
}
} catch (RejectedExecutionException ree) {
getLog().warn("Executor failed", ree);
return false;
} catch (Throwable t) {
getLog().error("Processing failed", t);
return false;
}
return true;
}スレッド間の連携フロー
Http11NioProtocol は Java NIO を基盤とし、Acceptor、Poller、Worker の 3 種類のスレッドによって構成されています。Acceptor はソケット接続を受け入れる生産者として機能し、生成されたイベントを EventQueue へ投入します。Poller は消費者としてイベントキューを監視し、セレクタを通じて IO 準備状態を確認します。準備が整った接続は、最終的に Worker スレッドプールへ渡され、ビジネスロジックの実行へと至ります。このモデルにより、高い同時接続処理能力と効率的なリソース利用が実現されています。