NIO에 대해 학습하며 작성된 글입니다. 학습 과정에서 작성된 글이기 때문에 잘못된 내용이 있을 수 있으며, 이에 대한 지적이나 피드백은 언제든 환영입니다.
1. 글을 작성하게 된 계기
이전 글에서 서블릿부터 컨트롤러 사이의 동작 과정에 대해 학습했습니다. NIO를 공부하다 보니 사용자 요청이 애플리케이션 Endpoint부터 서블릿 컨테이너까지 어떤 과정을 거쳐 동작하는지에 호기심이 생겼고, 이에 글을 작성하게 되었습니다.
이전 글에서 NIO의 등장 배경, NIO와 연관 개념들에 대해 살펴보았는데, 이번 글에서는 스프링 Endpoint부터 서블릿 컨테이너 전까지의 사용자 요청의 처리 과정에 대해 살펴보겠습니다.
NIO에 대한 이해가 없다면 이전 글을 참조해주세요.
2. 동작 과정
애플리케이션 Endpoint부터 서블릿 컨테이너까지는 아래 그림과 같은 순서로 동작합니다. 먼저 클라이언트 연결 요청이 오면 Acceptor에 의해 수락(1)되며, 해당 연결은 Poller에 전달(2)돼 I/O 이벤트를 기다립니다. I/O 이벤트가 발생하면 이는 Executor에 의해 처리(3)되어, 최종적으로 Http11Processor를 통해 서블릿 객체로 변환되어 서블릿 컨테이너에 전달(4)됩니다.
각 클래스는 다음과 같은 역할을 합니다. 다음으로 해당 클래스들이 어떻게 동작하는지 코드 레벨에서 살펴보겠습니다.
LimitLatch: 동시에 접속할 수 있는 최대 연결 수를 제한합니다.Acceptor: 클라이언트의 연결 요청을 수락하는 역할을 합니다. 연결이 수락되면 이를다른 구성 요소로 전달합니다.Poller: 클라이언트와의 연결을 모니터링하며 I/O 이벤트를 감지합니다.SocketProcessor: Poller가 감지한 I/O 이벤트에 대한 실제 작업을 처리합니다. 각 연결에 대해 SocketProcessor가 생성되며, 이는 Executor에 의해 관리되는 별도의 쓰레드에서 실행됩니다.Executor: SocketProcessor 작업을 실행하기 위한 쓰레드 풀입니다.NioSocketWrapper: 네트워크 소켓과 연관된 래퍼로, 이를 통해 데이터의 읽기/쓰기 작업이 이루어집니다.Http11Processor: SocketProcessor에 의해 호출되며, HTTP 요청을 처리합니다. HTTP 프로토콜에 따라 데이터를 분석하고, 해당 요청을 처리한 후 응답을 생성합니다.
2-1. LimitLatch
먼저 LimitLatch입니다. 이는 톰캣의 멀티 쓰레드 환경에서 동시 접근을 제어하기 위해 사용됩니다.
Shared latch that allows the latch to be acquired a limited number of times after which all subsequent requests to acquire the latch will be placed in a FIFO queue until one of the shares is returned.
여기서 limit는 application.yml의 max-connections의 값이며, 이를 설정하지 않은 경우 8,192개가 됩니다. 이를 통해 최대 커넥션 개수를 설정할 수 있습니다.
1
2
3
4
server:
port: 8090
tomcat:
max-connections: 1000
이는 동시 접근을 제어하기 위해 사용되기 때문에, 아래와 같이 사용자가 왔을 때 요청 가능한 자원을 증가/감소시키며 이를 관리합니다. Acceptor 클래스의 run( ) 메서드에서 이를 호출하는 것을 볼 수 있습니다.
2-2. Acceptor
다음은 Acceptor 클래스로, 이는 클라이언트 연결을 수락/처리 합니다.
새 연결이 들어오면 SocketChannel 객체가 되어 Poller에게 전달 돼 처리되며, 이는 run( ) 메서드를 통해 이루어집니다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
public class Acceptor<U> implements Runnable {
.....
@Override
public void run() {
......
try {
while (!stopCalled) {
while (endpoint.isPaused() && !stopCalled) {
......
try {
......
// 소켓 생성
U socket = null;
try {
socket = endpoint.serverSocketAccept();
} catch (Exception ioe) {
.....
}
......
if (!stopCalled && !endpoint.isPaused()) {
// 소켓 Option 설정
if (!endpoint.setSocketOptions(socket)) {
endpoint.closeSocket(socket);
}
} else {
endpoint.destroySocket(socket);
}
} catch (Throwable t) {
......
}
}
} finally {
stopLatch.countDown();
}
state = AcceptorState.ENDED;
}
}
......
}
setSocketOptions( ) 메서드를 보면 Channel, PollerEvent를 생성하고 PollerEvent를 큐에 등록하는 것을 볼 수 있습니다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
public class NioEndpoint extends AbstractJsseEndpoint<NioChannel,SocketChannel> {
......
@Override
protected boolean setSocketOptions(SocketChannel socket) {
NioSocketWrapper socketWrapper = null;
try {
NioChannel channel = null;
// 채널 생성
if (nioChannels != null) {
channel = nioChannels.pop();
}
if (channel == null) {
SocketBufferHandler bufhandler = new SocketBufferHandler(
socketProperties.getAppReadBufSize(),
socketProperties.getAppWriteBufSize(),
socketProperties.getDirectBuffer());
if (isSSLEnabled()) {
channel = new SecureNioChannel(bufhandler, this);
} else {
channel = new NioChannel(bufhandler);
}
}
// NioSocketWrapper 생성
NioSocketWrapper newWrapper = new NioSocketWrapper(channel, this);
// 채널 정보 등록
channel.reset(socket, newWrapper);
connections.put(socket, newWrapper);
socketWrapper = newWrapper;
socket.configureBlocking(false);
if (getUnixDomainSocketPath() == null) {
socketProperties.setProperties(socket.socket());
}
socketWrapper.setReadTimeout(getConnectionTimeout());
socketWrapper.setWriteTimeout(getConnectionTimeout());
socketWrapper.setKeepAliveLeft(NioEndpoint.this.getMaxKeepAliveRequests());
// 이벤트 등록
poller.register(socketWrapper);
return true;
} catch (Throwable t) {
......
}
return false;
}
......
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
public class Poller implements Runnable {
......
public void register(final NioSocketWrapper socketWrapper) {
socketWrapper.interestOps(SelectionKey.OP_READ);//this is what OP_REGISTER turns into.
PollerEvent pollerEvent = createPollerEvent(socketWrapper, OP_REGISTER);
addEvent(pollerEvent);
}
......
private void addEvent(PollerEvent event) {
events.offer(event);
if (wakeupCounter.incrementAndGet() == 0) {
selector.wakeup();
}
}
......
}
위에서 봤던 PollerEvent를 여기서 큐에 등록하는 것입니다.
참고로 NioSocketWrapper는 SocketChannel을 감싸는 래퍼 클래스로 소켓 상태, 읽기/쓰기 작업, 이벤트를 처리하는 데 필요한 다양한 메서드와 정보를 포함하고 있습니다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
public static class NioSocketWrapper extends SocketWrapperBase<NioChannel> {
private final SynchronizedStack<NioChannel> nioChannels;
private final Poller poller;
private int interestOps = 0;
private volatile SendfileData sendfileData = null;
private volatile long lastRead = System.currentTimeMillis();
private volatile long lastWrite = lastRead;
private final Object readLock;
private volatile boolean readBlocking = false;
private final Object writeLock;
private volatile boolean writeBlocking = false;
public NioSocketWrapper(NioChannel channel, NioEndpoint endpoint) {
super(channel, endpoint);
if (endpoint.getUnixDomainSocketPath() != null) {
localAddr = "127.0.0.1";
localName = "localhost";
localPort = 0;
remoteAddr = "127.0.0.1";
remoteHost = "localhost";
remotePort = 0;
}
nioChannels = endpoint.getNioChannels();
poller = endpoint.getPoller();
socketBufferHandler = channel.getBufHandler();
readLock = (readPending == null) ? new Object() : readPending;
writeLock = (writePending == null) ? new Object() : writePending;
}
......
}
대부분의 작업은 NioSocketWrapper를 통해 처리되는데, 이렇게 Wrapper 클래스를 사용하는 것은
필요한 상태와 정보를 캡슐화해서 논리 단위로 나누기 위함입니다.
2-3. Selector
다음은 Selector입니다. Selector는 여러 채널에 대한 입출력 이벤트를 감시하며, 각 채널에서 읽기, 쓰기, 연결, 수락 등의 이벤트가 발생했는지 비동기적으로 확인합니다. Selector에 의해 감지된 이벤트는 Poller에 의해 처리됩니다.
Selector는 SelectorProvider의 openSelector 메서드를 통해 열리게 됩니다.
1
2
3
4
5
6
7
8
9
10
11
public abstract class Selector implements Closeable {
protected Selector() { }
public static Selector open() throws IOException {
return SelectorProvider.provider().openSelector();
}
......
}
이후 사용자의 요청이 들어오면 채널이 생성되고, 이벤트를 등록합니다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
public class NioEndpoint extends AbstractJsseEndpoint<NioChannel,SocketChannel> {
......
@Override
protected boolean setSocketOptions(SocketChannel socket) {
NioSocketWrapper socketWrapper = null;
try {
......
// NioSocketWrapper 생성
NioSocketWrapper newWrapper = new NioSocketWrapper(channel, this);
// 채널 정보 등록
channel.reset(socket, newWrapper);
connections.put(socket, newWrapper);
socketWrapper = newWrapper;
socket.configureBlocking(false);
if (getUnixDomainSocketPath() == null) {
socketProperties.setProperties(socket.socket());
}
socketWrapper.setReadTimeout(getConnectionTimeout());
socketWrapper.setWriteTimeout(getConnectionTimeout());
socketWrapper.setKeepAliveLeft(NioEndpoint.this.getMaxKeepAliveRequests());
// 이벤트 등록
poller.register(socketWrapper);
return true;
......
}
return false;
}
......
}
이렇게 등록된 이벤트는 events( ) 메서드를 통해 처리됩니다. Selector는 여러 NIO 채널에서 발생하는 이벤트를 감시하며, 해당 이벤트에 어떻게 반응할지 결정하는데, 여기서 채널이 register( ) 메서드로 Selector에 등록됩니다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
public class Poller implements Runnable {
......
public boolean events() {
boolean result = false;
PollerEvent pe = null;
for (int i = 0, size = events.size(); i < size && (pe = events.poll()) != null; i++ ) {
......
if (sc == null) {
......
} else if (interestOps == OP_REGISTER) {
try {
// NIO 채널을 Selector에 등록하여 READ 이벤트가 발생할 때 알림을 받도록 함
sc.register(getSelector(), SelectionKey.OP_READ, socketWrapper);
} catch (Exception x) {
log.error(sm.getString("endpoint.nio.registerFail"), x);
}
} else {
// 이미 Selector에 등록된 채널의 SelectionKey를 검색
final SelectionKey key = sc.keyFor(getSelector());
......
}
......
}
return result;
}
......
}
위에서 봤던 그림과 같이요. 이를 통해 Selector는 Channel의 상태를 감시할 수 있습니다.
채널이 등록되면 Selector는 각 연결에 대한 읽기/쓰기 이벤트가 발생하면 적절한 작업을 수행할 수 있도록 이벤트를 감지합니다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
public class NioEndpoint extends AbstractJsseEndpoint<NioChannel,SocketChannel> {
......
@Override
public void run() {
// Loop until destroy() is called
while (true) {
boolean hasEvents = false;
try {
if (!close) {
hasEvents = events();
if (wakeupCounter.getAndSet(-1) > 0) {
// Selector가 이벤트 감지 - 깨어난 상태
keyCount = selector.selectNow();
} else {
// Selector가 이벤트 감지 - 블로킹 상태
keyCount = selector.select(selectorTimeout);
}
wakeupCounter.set(0);
}
......
} catch (Throwable x) {
......
}
......
}
getStopLatch().countDown();
}
......
}
이를 감시하는 것은 select( ) 메서드를 통해 동작하는데, 이를 따라가면 lockAndDoSelect( ), doSelect( ) 메서드를 거쳐 PollSelectorImpl 클래스에 도달하게 됩니다.
1
2
3
4
5
6
7
8
9
public abstract class Selector implements Closeable {
......
public abstract int select(long timeout) throws IOException;
......
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
public abstract class SelectorImpl
extends AbstractSelector
{
......
@Override
public final int select(long timeout) throws IOException {
if (timeout < 0)
throw new IllegalArgumentException("Negative timeout");
return lockAndDoSelect(null, (timeout == 0) ? -1 : timeout);
}
......
private int lockAndDoSelect(Consumer<SelectionKey> action, long timeout)
throws IOException
{
synchronized (this) {
ensureOpen();
if (inSelect)
throw new IllegalStateException("select in progress");
inSelect = true;
try {
synchronized (publicSelectedKeys) {
return doSelect(action, timeout);
}
} finally {
inSelect = false;
}
}
}
......
}
PollSelectorImpl 내부 구현을 보면 INTERRUPTED 상태를 체크하는 것을 볼 수 있습니다. 즉, 사용자 요청이 올 때까지 대기하고 있다가, 실제 요청이 오는 순간 이를 감지하고 요청을 처리하는 것입니다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class PollSelectorImpl extends SelectorImpl {
......
@Override
protected int doSelect(Consumer<SelectionKey> action, long timeout)
throws IOException
{
......
processUpdateQueue();
processDeregisterQueue();
try {
begin(blocking);
int numPolled;
do {
long startTime = timedPoll ? System.nanoTime() : 0;
numPolled = poll(pollArray.address(), pollArraySize, to);
// 운영체제의 Interrupt에 의한 감지.
if (numPolled == IOStatus.INTERRUPTED && timedPoll) {
long adjust = System.nanoTime() - startTime;
to -= TimeUnit.MILLISECONDS.convert(adjust, TimeUnit.NANOSECONDS);
if (to <= 0) {
numPolled = 0;
}
}
}
......
}
......
}
......
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
public final class IOStatus {
private IOStatus() { }
@Native public static final int EOF = -1; // End of file
@Native public static final int UNAVAILABLE = -2; // Nothing available (non-blocking)
@Native public static final int INTERRUPTED = -3; // System call interrupted
@Native public static final int UNSUPPORTED = -4; // Operation not supported
@Native public static final int THROWN = -5; // Exception thrown in JNI code
@Native public static final int UNSUPPORTED_CASE = -6; // This case not supported
......
}
감지된 이벤트가 있다면, 이벤트가 발생한 채널의 SelectionKey Iterator를 가져와 순회하며, 각각의 이벤트를 처리합니다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
public class Poller implements Runnable {
......
@Override
public void run() {
// Loop until destroy() is called
while (true) {
boolean hasEvents = false;
try {
if (!close) {
hasEvents = events();
......
}
......
// Selector에서 Key 목록 순회
Iterator<SelectionKey> iterator =
keyCount > 0 ? selector.selectedKeys().iterator() : null;
while (iterator != null && iterator.hasNext()) {
SelectionKey sk = iterator.next();
iterator.remove();
NioSocketWrapper socketWrapper = (NioSocketWrapper) sk.attachment();
// 작업 처리
if (socketWrapper != null) {
processKey(sk, socketWrapper);
}
}
// Process timeouts
timeout(keyCount,hasEvents);
}
getStopLatch().countDown();
}
......
}
processKey( ) 메서드를 보면 process( )와 같은 메서드를 볼 수 있습니다. 이를 통해 Executors를 호출하고 내부의 쓰레드에 요청을 하는 것입니다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
public class NioEndpoint extends AbstractJsseEndpoint<NioChannel,SocketChannel> {
......
protected void processKey(SelectionKey sk, NioSocketWrapper socketWrapper) {
try {
if (close) {
socketWrapper.close();
} else if (sk.isValid()) {
if (sk.isReadable() || sk.isWritable()) {
if (socketWrapper.getSendfileData() != null) {
processSendfile(sk, socketWrapper, false);
} else {
unreg(sk, socketWrapper, sk.readyOps());
boolean closeSocket = false;
if (sk.isReadable()) {
if (socketWrapper.readOperation != null) {
// 해당 메서드를 통해 Executors에 후속 요청 전달
if (!socketWrapper.readOperation.process()) {
closeSocket = true;
}
} else if (socketWrapper.readBlocking) {
synchronized (socketWrapper.readLock) {
socketWrapper.readBlocking = false;
socketWrapper.readLock.notify();
}
} else if (!processSocket(socketWrapper, SocketEvent.OPEN_READ, true)) {
closeSocket = true;
}
}
if (!closeSocket && sk.isWritable()) {
if (socketWrapper.writeOperation != null) {
// 해당 메서드를 통해 Executors에 후속 요청 전달
if (!socketWrapper.writeOperation.process()) {
closeSocket = true;
}
} else if (socketWrapper.writeBlocking) {
synchronized (socketWrapper.writeLock) {
socketWrapper.writeBlocking = false;
socketWrapper.writeLock.notify();
}
} else if (!processSocket(socketWrapper, SocketEvent.OPEN_WRITE, true)) {
closeSocket = true;
}
}
if (closeSocket) {
socketWrapper.close();
}
}
}
} else {
// Invalid key
socketWrapper.close();
}
......
}
}
......
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public abstract class SocketWrapperBase<E> {
......
protected boolean process() {
try {
// Executor에 처리
getEndpoint().getExecutor().execute(this);
return true;
} catch (RejectedExecutionException ree) {
log.warn(sm.getString("endpoint.executor.fail", SocketWrapperBase.this) , ree);
} catch (Throwable t) {
ExceptionUtils.handleThrowable(t);
log.error(sm.getString("endpoint.process.fail"), t);
}
return false;
}
......
}
즉, Executors 클래스에 사용자 요청을 비동기로 처리하도록 하는 것입니다.
2-4. Connector
그렇다면 위에서 봤던 Connector는 어디서 사용될까요? 이는 Http11Processor내부의 service( ) 메서드를 통해 서블릿이 처리할 수 있는 객체의 형태로 바꿔줍니다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
public class Http11Processor extends AbstractProcessor {
......
@Override
public SocketState service(SocketWrapperBase<?> socketWrapper)
throws IOException {
RequestInfo rp = request.getRequestProcessor();
rp.setStage(org.apache.coyote.Constants.STAGE_PARSE);
......
while (!getErrorState().isError() && keepAlive && !isAsync() && upgradeToken == null &&
sendfileState == SendfileState.DONE && !protocol.isPaused()) {
......
if (getErrorState().isIoAllowed()) {
try {
rp.setStage(org.apache.coyote.Constants.STAGE_SERVICE);
// Servlet 객체로 변환
getAdapter().service(request, response);
......
}
}
}
}
......
}
이는 내부적으로 CoyoteAdapter를 호출하는데, 여기서 Connector로부터 서블릿이 처리할 수 있는 객체로 변환해 주는 것입니다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
public class CoyoteAdapter implements Adapter {
......
@Override
public void service(org.apache.coyote.Request req, org.apache.coyote.Response res)
throws Exception {
Request request = (Request) req.getNote(ADAPTER_NOTES);
Response response = (Response) res.getNote(ADAPTER_NOTES);
if (request == null) {
// Create objects
request = connector.createRequest();
request.setCoyoteRequest(req);
response = connector.createResponse();
response.setCoyoteResponse(res);
// Link objects
request.setResponse(response);
response.setRequest(request);
// Set as notes
req.setNote(ADAPTER_NOTES, request);
res.setNote(ADAPTER_NOTES, response);
// Set query string encoding
req.getParameters().setQueryStringCharset(connector.getURICharset());
}
if (connector.getXpoweredBy()) {
response.addHeader("X-Powered-By", POWERED_BY);
}
......
}
즉, ServletContainer에 진입 전, Servlet이 처리할 수 있도록 객체를 변환하는 것입니다.
위에서 봤던, ServletContainer에 들어가기 전 거치는 Adapter가 CoyoteAdapter 입니다.
이를 통해 생성된 Request 객체를 보면 우리가 요청한 다양한 정보들이 파싱되어 담긴 것을 볼 수 있습니다.
여기서 정보가 파싱 돼 ServletRequest 객체가 생성되기 때문에, Filter나 Interceptor, Controller와 같은 뒷단에서 이를 사용할 수 있는 것입니다.
참고로 Connector로 부터 생성되는 Request는 ServletRequest의 자손입니다.
3. 정리
자바 NIO와 Endpoint부터 서블릿 컨테이너에 요청이 도달하기 전까지의 과정을 살펴보았습니다.
NIO에는 버퍼(Buffer), 채널(Channel), 커넥터(Connector), 셀렉터(Selector)와 같은 개념이 존재하며, 이들이 합쳐 아래와 같은 사용자 요청을 처리하게 됩니다.
이는 다음과 같은 과정을 거쳐 처리되며, 각 클래스의 역할은 다음과 같습니다. 이 외에도 NIO와 Selector에서 다루는 개념이 조금 더 있으므로, 이는 별도의 포스팅에서 살펴보겠습니다. 감사합니다.
Acceptor: 사용자 요청 수락. 이벤트 등록.Selector: 채널을 감시하며 알맞는 이벤트 처리. 실질적 처리는 Poller에서 관리.Http11Processor: 서블릿 컨테이너가 알 수 있는 형태로 데이터 파싱.