Chapter .19-5 NIO 기반 입출력 및 네트워킹(TCP 비동기 채널)

19.8 TCP 비동기 채널

NIO는 TCP블로킹, 넌블로킹 채널 이외에 TCP 비동기 채널로, AsynchronousServerSocket Channel과 AsynchronousSocketChannel을 제공한다.

 

각각 ServerSocketChannel과 SocketChannel에 대응하는 클래스로 NIO의 보석같은 클래스 이다.

 

이 두 클래스의 사용 방법을 알아보기로 하자.

 

19.8.1 비동기 채널의 특징

TCP 비동기 채널은 연결 요청(connect()), 연결 수락(accept()), 읽기(read()), 쓰기(write())를 호출하면 즉시 리턴된다.

이것은 넌블로킹 방식과 동일하다.

 

차이점은 이 메소드들을 호출하면 스레드풀에게 작업 처리를 요청하고 이 메소드들은 즉시 리턴된다.

실질적인 작업 처리는 스레드풀의 작업 스레드가 담당한다.

 

작업 스레드가 작업을 완료하게 되면 콜백(callback) 메소드가 자동 호출되기 때문에 작업 완료 후 실행해야 할 코드가 있다면 콜백 메소드에서 작성하면 된다.

 

애플리케이션에서 read() 메소드를 호출하면 즉시 리턴되지만, 실질적으로 내부에서는 스레드풀의 작업 스레드가 read() 메소드를 실행한다.

 

작업 스레드가 read() 메소드를 모두 실행하고 나면 콜백메소드인 completed() 메소드가 자동 호출된다.

이때 completed() 메소드를 실행하는 스레드는 스레드풀의 작업 스레드이다.

 

19.8.2 비동기 채널 그룹

비동기 서버소켓 채널(AsynchronousServerSocketChannel)과 비동기 소켓 채널(AsynchronousSocketChannel)을 살펴보기 전에 우선 비동기 채널 그룹(AsynchronousChannelGroup)에 대해서 이해해보자.

 

비동기 채널 그룹은 같은 스레드풀을 공유하는 비동기 채널들의 묶음이라고 볼 수 있다.

하나의 스레드풀을 사용한다면 모든 비동기 채널은 같은 채널 그룹에 속해야 한다.

 

비동기 채널을 생성할 떄 채널 그룹을 지정하지 않으면 기본 비동기 채널 그룹이 생성된다.

기본 비동기 채널 그룹은 내부적으로 다음과 같이 스레드풀을 생성한다.

new ThreadPoolExecutor(
    0, Integer.MAX_VALUE,
    Long.MAX_VALUE, TimeUnit.MILLISECONDS,
    new SynchronousQueue<Runnable>(),
    threadFactory);

이론적으로 Integer.MAX_VALUE개만큼 스레드가 증가할 수 있도록 되어 있다.

하지만 스레드풀은 대부분 최대 스레드 수를 지정해서 사용하므로 다음과 같이 AnsynchrounousChannelGroup을 직접 생성하고 사용하는 것이 일반적이다.

AsynchronousChannelGroup channelGroup = AsynchronousChannelGroup.withFixedThreadPool(
    최대스레드수,
    Executors.defaultThreadFactory()
);

 

다음은 CPU 코어의 수만큼 스레드를 관리하는 스레드풀을 생성하고 이것을 이용하는 비동기 채널 그룹을 생성한다.

AsynchronousChannelGroup channelGroup = AsynchronousChannelGroup.withFixedThreadPool(
    Runtime.getRuntime().avaliableProcessors(),
    Executors.defaultThreadFactory()
);

이렇게 생성된 비동기 채널 그룹은 비동기 채널을 생성할 때 매개값으로 사용된다.

비동기 채널 그룹을 더 이상 사용하지 않고 종료할 경우에는 shutdown()과 shutdownNow() 메소드를 호출 할 수 있다.

channelGroup.shutdown();
channelGroup.shutdownNow();

 

shutdown()은 비동기 채널 그룹을 종료하겠다는 의사만 전달할 뿐 즉시 비동기 채널 그룹을 종료하지 않는다.

비동기 채널 그룹에 포함된 모든 비동기 채널이 닫히면 비로소 비동기 채널 그룹이 종료된다.

 

shutdown() 메소드를 호출한 이후에 새로운 비동기 채널을 비동기 채널 그룹에 포함시키면 ShutdownChannelGroupException이 발생한다.

 

shutdownNow()는 강제적으로 비동기 채널 그룹에 포함된 모든 비동기 채널을 닫아버리고 비동기 채널 그룹을 종료한다.

 

단 완료 콜백을 실행하고 있는 스레드는 종료되거나, 인터럽트되지 않는다.

 

 

19.8.3 비동기 서버소켓 채널

AsynchronousServerSocketChannel은 두 가지 정적 메소드인 open()을 호출해서 얻을 수 있다.

기본 비동기 채널 그룹에 포함되는 AsynchronousServerSocketChannel을 얻는 방법은 다음과 같이 매개값 없는 open() 메소드를 호출하는 것이다.

AsynchronousServerSocketChannel asynchronousServerSocketChannel =
AsynchronousServerSocketChannel.open();

별도로 비동기 채널 그룹을 생성하고 여기에 포함되는 AsynchronousServerSocketChannel을 얻고 싶다면 다음과 같이 비동기 채널 그룹ㅇ르 매개값으로 갖는 open() 메소드를 호출하면 된다.

AsynchronousChannelGroup channelGroup = AsynchronousChannelGroup.withFixedThreadPool(
    Runtime.getRuntime().avaliableProcessors(),
    Executors.defaultThreadFactory()
);
AsynchronousServerSocketChannel asynchronousServerSocketChannel = AsynchronousServerSocketChannel.open(channelGroup);

AsynchronousServerSocketChannel을 생성하고 나서는 포트 바인딩을 위해 다음과 같이 bind() 메소드를 호출해야 한다.

asynchronousServerSocketChannel.bind(new InetSocketAddress(5001));

AsynchronousServerSocketChannel을 더이상 사용하지 않을 경우에는 close() 메소드를 호출해서 서버가 사용한 포트를 언바인딩 해준다.

asynchronousServerSocketChannel.close();

 

AsynchronousServerSocketChannel은 연결 수락 작업을 스레드풀을 잉요해서 비동기로 처리한다.

다음은 accpt()메소드를 호출하는 코드이다.

accpt(A attachement, CompletionHandler<AsynchronousSocketChannel, A> handler);

첫번쨰 매개값은 콜백 메소드의 매개값으로 제공할 첨부 객체인데, 연결 수락 작업에는 별도의 첨부 객체가 필요하지 않기 때문에 null을 지정한다.

 

두번째 매개값은 콜백 메소드를 가지고 있는 CompletionHandler<AsynchronousSocketChannel, A> 구현 객체이다.

A는 첨부 객체 타입인데, 연결 수락 작업에는 별도의 첨부 객체가 필요하지 않기 때문에 Void로 지정한다.

다음은 accpt() 메소드를 호출하는 기본 뼈대를 보여준다.

asynchronousServerSocketChannel.accept(null,
    new CompletionHandler<AsynchronousSocketChannel, Void>(){
    @Override
    public void completed(AsynchronousSocketChannel asynchronousSocketChannel,
                                                              Void attachment){
        //연결 수락 후 실행할 코드
        asynchronousServerSocketChannel.accept(null,this);  //accept 재호출
    }
    @Override
    public void failed(Throwable exc, Void attachment){
        //연결 수락 실패 시 실행할 코드
    }
 });

completed() 메소드는 연결 수락이 완료 되었을 떄 스레드풀의 스레드가 호출된다.

첫 번째 매개값은 연결 수락 후 리턴된 AsynchronousSocketChannel이 대입되고, 두 번째 매개값은 첨부 객체인데 accept()의 첫 번쨰 매개값이 null 이므로 null이 대입된다.

 

만약 스레드풀의 스레드가 연결 수락에 문제가 생겨 에외를 발생시키면 failed()가 호출된다.

failed()의 첫 번째 매개값은 예외객체이고 두 번째 매개값은 첨부 객체인데 accept()의 첫 번째 매개값이 null 이므로 null이 대입된다.

 

주목할 점은 accpt()를 반복해서 호출하는 무한 루프가 없다는 것이다.

대신 completed() 메소드 끝에 accept()를 재호출해서 반복적으로 클라이언트의 연결 수락 작업을 수행한다.

 

19.8.4 비동기 소켓 채널

AsynchronousSocketChannel은 서버와 클라이언트에 각각 존재하는데, 클라이언트가 AsynchronousSocketChannel을 생성해서 서버로 연결을 요청하면 서버의 AsynchronousServerSocketChannel 은 연결 수락 후 AsynchronousSocketChannel을 생성해서 서로 통신할 수 있도록 만들어준다.

 

AsynchronousServerSocketChannel이 생성하는 AsynchronousSocketChannel은 자동적으로 AsynchronousServerSocketChannel과 같은 비동기 채널 그룹에 속하게 된다.

 

클라이언트에서 AsynchronousSocketChannel을 생성하려면 open() 메소드를 사용할 수 있다.

기본 비동기 채널에 포함되는 AsynchronousSocketChannel을 얻고 싶다면 매개값 없는 open() 메소드를 호출하면 된다.

AsynchronousSocketChannel asynchronousSocketChannel = AsynchronousSocketChannel.open();

 

별도로 비동기 채널 그룹을 생성하고 여기에 포함되는 AsynchronousSocketChannel을 얻고 싶다면 다음과 같이 비동기 채널 그룹을 매개값으로 갖는 open() 메소드를 호출하면 된다.

AsynchronousChannelGroup channelGroup = AsynchronousChannelGroup.withFixedThreadPool(
    Runtime.getRuntime().avaliableProcessors(),
    Executors.defaultThreadFactory()
);

AsynchronousSocketChannel asynchronousSocketChannel = AsynchronousSocketChannel.open(channelGroup);

 

AsynchronousSocketChannel을 더이상 사용하지 않을 경우에는 close() 메소드를 호출해서 연결을 끊어준다.

asynchronousSocketChannel.close();

 

클라이언트가 생성하는 AsynchronousSocketChannel은 서버 연결 요청 작업을 스레드풀을 이용해서 비동기로 처리한다.

다음은 connect() 메소드를 호출하는 코드이다.

connect(SocketAddress remote, A attachment, CompletionHandler<Void, A> handler);

첫 번째 매개값은 서버 IP와 연결 포트 정보를 가진 InetSocketAddress 객체이다.

두 번째 매개값은 콜백 메소드의 매개값으로 제공할 첨부 객체인데, 연결 요청 작업에는 별도의 첨부 객체가 필요하지 않기 때문에 null을 지정한다.

 

세 번째 매개값은 CompletionHandler<Void, A> 구현 객체이다.

A는 첨부 객체 타입인데, 연결 요청 작업에는 별도의 첨부 객체가 필요하지 않기 때문에 Void로 지정한다.

 

다음은 connect() 메소드를 호출하는 기본 뼈대를 보여준다.

asynchronousSocketChannel.connect(new InetSocketAddress("localhost", 5001), null,
    new CompletionHandler<Void, Void>(){
    @Override
    public void completed(Void result, Void attachment){
        //연결 성공 후 실행할 코드
    }
    @Override
    public void failed(Throwable e, Void attachment){
        //연결 실패후 실행할 코드
    }
});

completed() 메소드는 연결이 성공했을 때 스레드풀의 스레드를 호출한다.

첫번째 매개값은 null이 대입되고, 두 번째 매개값은 첨부 객체인데 connect()의 두 번째 매개값이 null 이므로 null이 대입된다.

 

completed() 메소드에서는 서버가 보낸 데이터를 받기 위한 코드가 일반적으로 작성된다.

 

만약 스레드풀의 스레드가 연결 요청에 문제가 생겨 예외가 발생되면 failed()가 호출된다.

 

첫 번째 매개값은 예외 객체이고, 두 번쨰 매개값은 첨부 객체인데 connect()의 두 번째 매개값이 null 이므로 null이 대입된다.

 

19.8.5 비동기 소켓 채널 데이터 통신

클라이언트와 서버가 연결되면 양쪽 AsynchronousSocketChannel의 read()와 write() 메소드로 데이터 통신을 할 수 있다.

이 메소드들은 호출하는 즉시 리턴되고, 실질적인 작업은 스레드풀의 스레드가 담당한다.

 

다음은 read() 와 write()를 호출하는 코드이다.

read(ByteBuffer dst, A attachment, CompletionHandler<Integer, A> handler);
write(ByteBuffer src, A attachment, CompletionHandler<Integer, A> handler);

첫 번째 매개값은 읽고 쓰기 위한 ByteBuffer 객체이다.

두 번째 매개값은 콜백 메소드의 매개값으로 제공할 첨부 객체이다.

세 번쨰 매개값은 CompletionHandler<Integer, A> 구현 객체이다.

Integer는 읽고 쓴 바이트 수이고, A는 첨부 객체 타입이다.

 

다음은 read() 메소드를 호출하는 기본 뼈대를 보여준다.

asynchronousSocketChannel.read(byteBuffer, attachment,
    new CompletionHandler<Integer, A>() {
    @Override
    public void completed(Integer result, A attachment){
      //받은 데이터를 처리하는 코드
      asynchronousSocketChannel.read(byteBuffer, attachment, this);   // this : read() 호출임
    }
    @Override
    public void failed(Throwable exc, A attachment){
       //실패된 경우 실행할 코드
    }
});

스레드풀의 작업 스레드가 읽기 작업을 완료하면 completed() 메소드를 호출한다.

첫 번째 매개값은 읽은 바이트수이고,

두 번쨰 매개값은 첨부 객체인데 read() 메소드 호출 시 사용된 두 번째 매개값이 대입된다.

만약 스레드풀의 작업 스레드에서 읽기 작업 도중 문제가 생겨 예외가 발생하면 failed() 메소드를 호출한다.

 

첫 번째 매개값은 예외 객체이고, 두 번째 매개값은 첨부 객체인데 read() 메소드 호출 시 사용된 두 번째 매개값이 대입된다.

 

주목할 점은 read()를 반복해서 호출하는 무한루프가 없다는 점이다.

 

대신 completed() 메소드 끝에 read() 메소드를 재호출해서 반복적으로 데이터를 받는 작업을 수행한다.

 

다음은 write() 메소드를 호출하는 기본 뼈대를 보여준다.

asynchronousSocketChannel.write(byteBuffer, attachment,
    new CompletionHandler<Integer, A>(){
    @Override
    public void completed(Integer result, A attachement){
        //성공한 경우 실행할 코드
    }
    @Override
    public void failed(Throwable exc, A attachment){
        //실패된 경우 실행할 코드
    }
});

스레드풀의 작업 스레드가 쓰기 작업을 완료하면 completed() 메소드를 호출한다.

첫 번째 매개값은 사용한 바이트 수이고, 

두 번쨰 매개값은 첨부 객체인데 write() 메소드 호출 시 사용된 두번쨰 매개값이 대입된다.

 

만약 스레드풀의 작업 스레드에서 쓰기 작업 도중에 문제가 생겨 예외가 발생하면 failed() 메소드를 호출한다.

 

첫 번째 매개값은 예외 객체이고

두 번쨰 매개값은 첨부 객체인데 write() 메소드 호출 시 사용된 두 번째 매개값이 대입된다.

 

19.8.6 채팅 서버 구현

 

■채팅 서버, 클라이언트와 UI 구현

- P.1204~1218

 

- 비동기 서버소켓채널과 비동기 소켓채널 사용법 이해

- 서버에는 다수의 클라이언트가 붙는다는 것을 전제로 작업

 

- UI의 경우는 TCP 동기채널의 작동과 동일

 

 

 

19.9 UDP 채널

NIO에서 UDP 채널은 DatagramChannel이다.

datagramChannel도 TCP 채널과 마찬가지로 블로킹과 넌블로킹 방식으로 사용할 수 있지만, 책에서는 블로킹 방식만 설명하도록 하겠다.

 

19.9.1 발신자 만들기

발신자 프로그램을 구현해 보면서 DatagramChannel을 사용하는 방법에 대해 알아보기로 하자.

DatagramChannel을 생성하려면 open() 메소드를 호출해야 한다.

 

open() 메소드는 ProtocalFamily 인터페이스 타입의 매개값을 갖는데, 이 객체의 역할을 IPv4와 IPv6를 구분하기 위해서이다.

 

구현 객체는 StandardProtocolFamily 열거 상수를 사용하면 된다.

 

다음은 IPv4를 사용하는 DatagramChannel을 생성하는 코드이다.

DatagramChannel datagramChannel = DatagramChannel.open(StandardProtocolFamily.INET);

DatagramChannel을 이용해서 데이터를 보내기 위해서는 send() 메소드를 이용한다.

send()메소드의 첫 번째 매개값은 보낼 데이터를 가지고 있는 ByteBuffer이고, 두 번째 매개값은 수신자 IP와 포트 정볼르 가지고 있는 InetSocketAddress이다.

 

SocketAddress는 추상 클래스이므로 하위 클래스인 InetSocketAddress 객체를 생성하고 대입하면 된다.

 

send() 메소드의 리턴값은 실제로 보낸 바이트 수이다.

다음은 로컬PC 5001번을 수신자로 하고 데이터를 보낸다.

int byteCount = datagramChannel.send(byteBuffer, new InetSocketAddress("localhost", 5001));

더 이상 보낼 데이터가 없을 경우에는 DatagramChannel을 닫기 위해 close() 메소드를 호출한다.

datagramChannel.close();

 

다음 예제는 UDP 발신 프로그램이다.

for문은 두 번 반복하는데 각각 "메세지1", "메세지2" 문자열을 전송한다.

import java.net.InetSocketAddress;
import java.net.StandardProtocolFamily;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.charset.Charset;

public class UdpSendExample {
	public static void main(String[] args) throws Exception {
		DatagramChannel datagramChannel = DatagramChannel.open(StandardProtocolFamily.INET);
		
		System.out.println("[발신 시작]");
		
		for(int i=1; i<3; i++) {
			String data = "메시지" + i;
			Charset charset = Charset.forName("UTF-8");
			ByteBuffer byteBuffer = charset.encode(data);
			
			int byteCount = datagramChannel.send(
				byteBuffer,  
				new InetSocketAddress("localhost", 5001)
			);
			System.out.println("[보낸 바이트 수] " + byteCount + " bytes");
		}
		
		System.out.println("[발신 종료]");
		
		datagramChannel.close();
	}
}

 

 

19.9.2 수신자 만들기

이번에는 DatagramChannel로 수신자 프로그램을 구현하는 방법에 대해 알아보기로 하자.

 

DatagramChannel을 이용해서 데이테를 받기 위해서는 bind() 메소드를 호출해서 포트와 먼저 바인딩을 해야한다.

매개값은 SocketAddress 타입으로 InetSocketAddress 객체를 대입하면 된다.

DatagramChannel datagramChannel = DatagramChannel.open(StandardProtocolFamily.INET);
datagramChannel.bind(new InetSocketAddress(5001));

포트와 바인딩이 되었다면 다음과 같이 receive() 메소드로 데이터를 받을 수 있다.

receive() 메소드의 매개값은 받은 데이터를 저장할 ByteBuffer이고, 리턴 타입은 원격 클라이언트의 IP와 포트 정볼르 가지고 있는 SocketAddress이다.

 

실제로는 InetSocketAddress가 리턴된다.

SocketAddress socketAddress = datagramChannel.receive(ByteBuffer dst);

데이터를 받기 전까지 receive() 메소드는 블로킹되고, 데이터를 받으면 리턴된다.

 

수신자는 항상 데이터를 받을 준비를 해야 하므로 작업 스레드를 생성해서 receive() 메소드를 반복적으로 호출해야 한다.

 

작업 스레드를 종료시키는 방법은 두 가지 인데, receive() 메소드가 블로킹 되어 있는 상태에서 작업 스레드의 interrupt()를 호출시켜 ClosedByInterruptException 예외를 발생시키거나,

다음과 같이 DatagramChannel의 close()를 호출시켜 AsynchronousCloseException 예외를 발생시키는 것이다.

 

그리고 예외 처리 코드에서 작업 스레드를 종료시키면 된다.

datagramChannel.close();

 

다음은 수신자 프로그램의 전체 코드이다.

실행 후 10초가 지나면 수신자를 종료하도록 하였다.

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.StandardProtocolFamily;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.charset.Charset;

public class UdpReceiveExample extends Thread {
	public static void main(String[] args) throws Exception {
		DatagramChannel datagramChannel = DatagramChannel.open(StandardProtocolFamily.INET);
		datagramChannel.bind(new InetSocketAddress(5001));
		
		Thread thread = new Thread() {
			@Override
			public void run() {
				System.out.println("[수신 시작]");
				try {
					while(true) {
						ByteBuffer byteBuffer = ByteBuffer.allocateDirect(100);
						SocketAddress socketAddress = datagramChannel.receive(byteBuffer);
						byteBuffer.flip();
						
						Charset charset = Charset.forName("UTF-8");
						String data = charset.decode(byteBuffer).toString();
						System.out.println("[받은 내용: "  + socketAddress.toString() + "] " + data);
					}
				} catch (Exception e) {
					System.out.println("[수신 종료]");
				}
			}			
		};
		thread.start();
		
		Thread.sleep(10000);
		datagramChannel.close();
	}
}

 

 

19.9.3 수신자와 발신자 실행

실행순서는 상관 없지만, 수신자를 먼저 실행하고 발신자를 싱행해야만 발신자가 보낸 데이터를 수신자가 모두 받을 수 있다.

 

발신자를 먼저 실행하면 수시자가 실행하기 전에 보낸 데이터는 받을 수 없다.