[JAVA / 네트워크] 비동기 통신 프로그램 샘플

저번 포스팅에 이어서 JAVA MessagePacker 및 이를 활용한 비동기 통신 프로그램의 예이다.

/*
 * Author : Gompang
 * Desc : MessagePacker를 활용한 비동기 통신 프로그램 서버
 * Blog : http://gompangs.tistory.com/
 */

package Chat;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import java.util.Iterator;

import MsgPacker.MessagePacker;
import MsgPacker.MessageProtocol;

public class ChatServer implements Runnable {

	private Selector selector;private HashMap<SocketChannel,String>dataMapper;
	private InetSocketAddress socketAddress;
	private MessagePacker msg; // 여기서 MessagePacker를 써보자

	public ChatServer(String address, int port) {
		socketAddress = new InetSocketAddress(address, port); // 소켓 주소 설정
		dataMapper = new HashMap<SocketChannel, String>();
	}

	public static void main(String args[]) {
		ChatServer server = new ChatServer("localhost", 31203); // 서버 객체 생성
		server.run(); // 서버 실행
	}

	@Override
	public void run() { // 서버가 실행되면 호출된다.
		// TODO Auto-generated method stub
		try {
			selector = Selector.open(); // 소켓 셀렉터 열기
			ServerSocketChannel socketChannel = ServerSocketChannel.open(); // 서버소켓채널 열기
			socketChannel.configureBlocking(false); // 블럭킹 모드를 False로 설정한다.

			socketChannel.socket().bind(socketAddress); // 서버 주소로 소켓을 설정한다.
			socketChannel.register(selector, SelectionKey.OP_ACCEPT); // 서버셀렉터를 등록한다.

		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		System.out.println("Server Started!");

		while (true) {
			try {
				selector.select(); // 셀럭터로 소켓을 선택한다. 여기서 Blocking 됨.

				Iterator<?> keys = selector.selectedKeys().iterator();

				while (keys.hasNext()) { // 셀렉터가 가지고 있는 정보와 비교해봄

					SelectionKey key = (SelectionKey) keys.next();
					keys.remove();

					if (!key.isValid()) { // 사용가능한 상태가 아니면 그냥 넘어감.
						continue;
					}

					if (key.isAcceptable()) { // Accept가 가능한 상태라면
						accept(key);
					}

					else if (key.isReadable()) { // 데이터를 읽을 수 있는 상태라면
						readData(key);
					}
				}

			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}

	}

	private void accept(SelectionKey key) { // 전달받은 SelectionKey로 Accept를 진행
		ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel();
		SocketChannel channel;

		try {
			channel = serverChannel.accept();
			channel.configureBlocking(false);
			Socket socket = channel.socket();
			SocketAddress remoteAddr = socket.getRemoteSocketAddress();
			System.out.println("Connected to: " + remoteAddr);

			// register channel with selector for further IO
			dataMapper.put(channel, remoteAddr.toString());
			channel.register(this.selector, SelectionKey.OP_READ);

		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	}

	private void readData(SelectionKey key) { // 전달받은 SelectionKey에서 데이터를 읽는다.

		SocketChannel channel = (SocketChannel) key.channel();
		ByteBuffer buffer = ByteBuffer.allocate(1024);
		int numRead = -1;

		try {
			numRead = channel.read(buffer);

			if (numRead == -1) { // 아직 읽지 않았다면 읽는다.
				this.dataMapper.remove(channel);
				Socket socket = channel.socket();
				SocketAddress remoteAddr = socket.getRemoteSocketAddress();
				channel.close();
				key.cancel();
				return;
			}

		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		System.out.println(buffer.array().length);

		msg = new MessagePacker(buffer.array()); // Byte Data를 다 받아왔다.
		byte protocol = msg.getProtocol();

		switch (protocol) {

		case MessageProtocol.CHAT: {
			System.out.println("CHAT");
			System.out.println(msg.getString());
			System.out.println(msg.getInt());
			break;
		}

		case MessageProtocol.BATTLE_START: {
			System.out.println("BATTLE_START");
			break;
		}

		case MessageProtocol.BATTLE_END: {
			System.out.println("BATTLE_END");
			break;
		}

		case MessageProtocol.WHISPHER: {
			System.out.println("WHISPHER");
			break;
		}
		}

	}
}

/*
 * Author : Gompang
 * Desc : MessagePacker를 활용한 비동기 통신 프로그램 클라이언트
 * Blog : http://gompangs.tistory.com/
 */
package Chat;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;

import MsgPacker.MessagePacker;
import MsgPacker.MessageProtocol;

public class ChatClient {
 
    public void startClient() throws IOException, InterruptedException {
 
        InetSocketAddress hostAddress = new InetSocketAddress("localhost", 31203);
        SocketChannel client = SocketChannel.open(hostAddress);
 
        System.out.println("Client Started!");
        
        MessagePacker msg = new MessagePacker(); // MessagePacker 사용해보자
        
        msg.SetProtocol(MessageProtocol.CHAT);
        msg.add("채팅 메세지 전송 테스트입니다~~ This is test message for CHAT");
        msg.add(1020302);
        msg.Finish();
        
        client.write(msg.getBuffer());
        client.close();
    }
    
    public static void main(String args[]) throws IOException, InterruptedException{
    	
    	ChatClient client = new ChatClient();
    	client.startClient();
    }
}

Server부분의 코드가 꽤 길고,

그에 비해 Client단의 코드가 짧은 편이다.

---

Server

MessagePacker를 사용하기 위해 import가 필요하다

import MsgPacker.MessagePacker;

import MsgPacker.MessageProtocol;

를 상단에 추가해준다.

호출되는 순서는 main -> CharServer -> run() -> selctor.select() -> accept or read() 순이다.

설명하기 앞서, 사용된 코드는 JAVA NIO(New/Non-blocking I/O) 를 사용한 단순한 네트워크 프로그램임을 알림.

본 코드는 클라이언트의 메세지를 읽고 화면에 출력해주는 역할을 하고 있음.

(채팅과 관련된 처리 되어 있지 않음. MessagePacker 기능 테스트를 위함)

#1. 최초 ChatServer 클래스를 생성하면서 함께 연결할 주소와 포트를 함께 넘겨주게 된다.

ChatServer server = new ChatServer("localhost", 31203); // 서버 객체 생성

server.run(); // 서버 실행

이후 서버를 실행시킨다.

#2. 이후 서버에서는 Selector(NIO의 다중접속 연결관리를 위한 클래스)를 사용해서 소켓에 Selector를 사용하여 연결을 받기 시작한다.

Selector는 C / JAVA에서 사용되는 가장 기초적인 개념 중 하나이며, 여러개의 소켓에서 발생된 이벤트(Accept , Read, Write 등)를 파악하고

각 소켓을 선택해주는 역할을 담당하고 있다.

selector = Selector.open(); // 소켓 셀렉터 열기

ServerSocketChannel socketChannel = ServerSocketChannel.open(); // 서버소켓채널 열기

socketChannel.configureBlocking(false); // 블럭킹 모드를 False로 설정한다.

socketChannel.socket().bind(socketAddress); // 서버 주소로 소켓을 설정한다.

socketChannel.register(selector, SelectionKey.OP_ACCEPT); // 서버셀렉터를 등록한다.

위의 코드가 진행되면, CharServer를 생성할때 넘겨준 포트와 주소로 Server를 Bind하게 된다.

#3. 그 다음은 Selector로 생성된 소켓에 이벤트가 발생했는지를 체크해야 하는 부분이다.

while문을 돌면서 Selector의 이벤트 발생 시 select() 함수로 받아온다.

while (true) {

try {

selector.select(); // 셀럭터로 소켓을 선택한다. 여기서 Blocking 됨.

Iterator<?> keys = selector.selectedKeys().iterator();

while (keys.hasNext()) { // 셀렉터가 가지고 있는 정보와 비교해봄

SelectionKey key = (SelectionKey) keys.next();

keys.remove();

if (!key.isValid()) { // 사용가능한 상태가 아니면 그냥 넘어감.

continue;

}

if (key.isAcceptable()) { // Accept가 가능한 상태라면

accept(key);

}

else if (key.isReadable()) { // 데이터를 읽을 수 있는 상태라면

readData(key);

}

}

} catch (IOException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

}

이부분인데, 계속 루프가 돌지는 않고 이벤트가 있을때만 돌고, 아닐때는 selector.select();에서 루프문이 멈춰있게 된다.

이후 SelectionKey의 상태를 확인해서 accept를 할 것인지, Read를 할 것인지 확인한다.

그리고 각각에 맞는 동작을 하는 함수를 호출해준다.

Accept는 따로 설명하지 않고, readData를 보도록 하자.

#4. 

SocketChannel channel = (SocketChannel) key.channel();

ByteBuffer buffer = ByteBuffer.allocate(1024);

int numRead = -1;

try {

numRead = channel.read(buffer);

if (numRead == -1) { // 아직 읽지 않았다면 읽는다.

this.dataMapper.remove(channel);

Socket socket = channel.socket();

SocketAddress remoteAddr = socket.getRemoteSocketAddress();

channel.close();

key.cancel();

return;

}

} catch (IOException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

위와 같은 코드로 데이터를 읽게 되며, 

numRead = channel.read(buffer); 의 코드로 buffer에는 데이터가, numRead에는 읽은 바이트수가 들어가게 된다.

msg는 MessagePacker로 선언되어 있음 ( MessagePacker msg;)

여기서부터는 클라이언트로부터 받은 데이터를 읽어오는 부분이다

byte protocol = msg.getProtocol();

switch (protocol) {

case MessageProtocol.CHAT: {

System.out.println("CHAT");

System.out.println(msg.getString());

System.out.println(msg.getInt());

break;

}

case MessageProtocol.BATTLE_START: {

System.out.println("BATTLE_START");

break;

}

case MessageProtocol.BATTLE_END: {

System.out.println("BATTLE_END");

break;

}

case MessageProtocol.WHISPHER: {

System.out.println("WHISPHER");

break;

}

}

직관적이라 바로 알 수 있을것이다. getProtocol()로 맨 앞 byte를 읽어서 프로토콜에 따라 분기한다

예시로 CHAT, BATTLE_START, BATTLE_END, WHISPHER로 잡았는데, 이 부분이 자유롭게 변경될 수 있다는 것(MessageProtocol 수정해야함)

프로토콜을 읽은 뒤 각각에 맞는 행동을 하면 된다.

단, 넣은 순서대로 읽어야 한다

무슨 소리냐하면

Client :

msg.add("한글 테스트 스트링~");

msg.add(1234);

msg.add(132.23142f);

msg.add(1.234123);

를 넣어서 전송했다면 받는 측에서는

msg.getString();

msg.getInt();

msg.getFloat();

msg.getDouble();

로 읽어야한다는 소리다.

---

Client

설명할 코드가 거의 없다

        InetSocketAddress hostAddress = new InetSocketAddress("localhost", 31203);

        SocketChannel client = SocketChannel.open(hostAddress);

 

        System.out.println("Client Started!");

        

        MessagePacker msg = new MessagePacker(); // MessagePacker 사용해보자

        

        msg.SetProtocol(MessageProtocol.CHAT);

        msg.add("채팅 메세지 전송 테스트입니다~~ This is test message for CHAT");

        msg.add(1020302);

        msg.Finish();

        

        client.write(msg.getBuffer());

        client.close();

1. 서버 주소/포트를 지정

2. SocketChannel을 open시켜준다(위 주소로)

3. MessagePacker를 생성

4. MessagePacker에 데이터를 넣는다.

5. Finish() 꼭 호출해주자

6. msg.getBuffer()로 ByteBuffer를 가져온다

7. SocketChannel로 write한다

8. close() -> 현재 한번 보내고 종료하는데, 이 부분을 바로 종료하지 말고 원하는 로직을 구성하면 된다

---

앞서 구현했던 MessagePacker를 이용한 데이터 송/수신을 살펴보았음

아무래도 간단한 데이터가 아니라, 여러 데이터형 및 프로토콜을 활용하고자 한다면

이렇게 사용할 수도 있다는 것을 예시로 들고 싶었다

아직 클래스 최적화 및 사용성을 좋게하는 함수들은 구현이 완벽하진 않지만, 그래도 충분히 사용가능할만하다.

(JAVA <-> JAVA 간 통신)

만약, 이종 프레임워크 간(JAVA <-> C#, C++ <-> C# 등) 데이터 통신에 필요한 메세지 형식을 만들어야 한다면

MessagePack(오픈소스), Protobuf(오픈소스), FlatBuffer(오픈소스), ProudNet(유료) 를 사용하는 것을 고려해봐야 한다.

IDL이라는 방식으로 타겟 플랫폼에 맞게 클래스 및 헤더를 생성해주는 라이브러리들이다.

이상으로 일단 JAVA에서의 네트워크 통신을 구성하는 방법에 대해 알아봤다.

다음 포스팅주제는... 흠 C# SuperSocket을 활용한 서버 혹은 관련 게임서버 주제로 포스팅하고자 한다.