Push Server 구현하기

서론

저번 포스팅에서 서버로부터 실시간으로 데이터를 받아오기 위한 기법에 대해서 설명을 하였고 이번에는 실제로 구현을 해본다. 구현을 위해 선택한 방법은 Streaming 기법을 선택 하였으며 Netty를 사용하였다. Netty를 사용한 이유는 이벤트 기반으로 동작하는 네트워크 프레임워크로써 요청에 대한 이벤트를 설정해서 항상 대기하는 서버가 아닌 이벤트 기반으로 동작하게 하여 효율성을 높이기 위해서 선택하였으며 통신기법은 WebSocket을 사용하였다. WebSocket 같은 경우는 HTML5에서 지원되므로 낮은 버전의 브라우저에서는 동작하지 않게 된다. 이 경우 SSE를 사용하여야 하며 이번 포스팅에서는 WebSocket기준으로만 구현을 하였다.

Netty에 대한 간략한 설명

Netty에 대해서는 깊이 있는 학습이 요구되지만 기본적으로 알아야하는 개념에 대해서 설명하고자한다. 첫번째로는 Netty로 서버를 만들기 위해서 사용하는 ServerBootstrap과 ChannelPipeLine에 대해서 설명을 하며 [자바 네트워크 소녀 Netty - 저 정경석]을 참조 하였다.

1. ServerBootStrap : Netty로 서버를 구현하기 위해서 사용하는 클래스로서 서버가 어떻게 동작 할 지에 대해서 설정한다. 설정하기 위해서 사용하는 종류는 7가지이며 아래와 같다.

1.1 group - 이벤트 루프 설정을 하기 위해서 사용되며 Server의 group은 클라이언트와 연결에 대해서 동작하는 bossGroup과 클라이언트가 보내오는 통신에 의해 발생하는 이벤트를 처리하는 workerGroup을 설정하게 된다.

1.2 channel - 소켓 입출력 모드를 설정하기 위해서 사용되며 이번 구현에서는 NioServerSocketChannel.class을 사용할 것이며 이 클래스는 논블로킹 모드의 서버 소켓 채널을 사용하기 위해 선언한다. NioServerSocketChannel.class를 제외하고도 Netty에서는 기본적으로 7개의 Channel의 종류를 더 지원해주고 있다.

1.3 channelFactory - 위의 channel이 하는 역할과 똑같지만 Factory단위로 소켓 입출력 모드를 설정하기 위해서 사용한다.

1.4 handler - 서버 소켓 채널의 이벤트 핸들러를 설정하기 위해 사용되며 대표적인 사용 예를 들면 서버에서 발생하는 로그를 남기기 위해서 LogginHandler를 사용하는 것이 있다. 

1.5 childHanler - 클라이언트와 연결된 채널에 이벤트가 발생하면 해당 이벤트 간에 발생한 데이터를 가공하기 위한 Handler를 정의하기 위한 부분이다. childHandler에 Handler를 등록하고 싶다면 ChannelInitializer<SocketChannel>을 구현 한 객체를 값으로 주어야한다.

1.6 option, childOption은 서버 소켓 채널, 소켓 채널 소켓 옵션 설정에 사용하며 예로 Nagle 알고리즘 비활성화 여부 지정과 같은 옵션을 설정한다. 기본적으로 총 7개의 option을 제공한다.

* 서버 소켓 채널과 소켓 채널의 차이는 서버 소켓 채널은 클라이언트와의 연결에 사용되는 채널이며 소켓 채널은 서버 내부에서 채널 파이프라인을 등록받고 데이터를 가공할 수 있는 연결고리이다.

 2. ChannelPipeLine : Netty의 채널 즉, 소켓 채널과 발생 이벤트에 맞는 상황에서의 데이터를 가공할 수 있게 해주는 이벤트핸들러 사이에 연결 통료 역할을 수행한다. EventHandler는 소켓 채널에서 발생하는 이벤트를 처리하는 채널 인바운드 이벤트, 소켓 채널에서 발생한 이벤트중에서 프로그래머가 명시한 이벤트를 처리하는 아웃바운트 이벤트가 있다.

1.1 채널 인바운드 이벤트 발생 순서

channelRegistred -> channelActive -> channelRead -> channelReadComplete -> channelInactive -> channelUnregistered 이며 만약 강제로 발생순서를 조작하고 싶다면 fire접두어가 붙은 메소드를 사용하여 발생순서를 조작 할 수 있다.

Push Server Main 구현

package com.netty.push.server; import com.netty.push.init.HttpServerInit; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.logging.LogLevel; import io.netty.handler.logging.LoggingHandler; public class PushServer { public static void main(String[] args) { EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(); ChannelFuture channelFuture = null; System.out.println("serverInit"); try { ServerBootstrap b = new ServerBootstrap(); b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class) .handler(new LoggingHandler(LogLevel.INFO)) .childHandler(new HttpServerInit()); Channel ch = b.bind(8989).sync().channel(); channelFuture = ch.closeFuture(); channelFuture.sync(); }catch(Exception e) { e.printStackTrace(); }finally{ bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } }

ChannelInitializer구현체인 HttpServerInit의 구현

package com.netty.push.init; import com.netty.handler.HttpHandler2; import com.netty.handler.WebSocketHandler2; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelPipeline; import io.netty.channel.socket.SocketChannel; import io.netty.handler.codec.http.HttpObjectAggregator; import io.netty.handler.codec.http.HttpRequestDecoder; import io.netty.handler.codec.http.HttpResponseEncoder; import io.netty.handler.codec.http.cors.CorsConfig; import io.netty.handler.codec.http.cors.CorsConfigBuilder; import io.netty.handler.codec.http.cors.CorsHandler; import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler; public class HttpServerInit extends ChannelInitializer<SocketChannel>{ private static final CorsConfig corsConfig = CorsConfigBuilder.forAnyOrigin().allowNullOrigin().allowCredentials().build(); @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeLine = ch.pipeline(); pipeLine.addLast(new HttpRequestDecoder()); pipeLine.addLast(new HttpObjectAggregator(65536)); pipeLine.addLast(new CorsHandler(corsConfig)); pipeLine.addLast(new HttpResponseEncoder()); pipeLine.addLast(new WebSocketServerProtocolHandler("/")); pipeLine.addLast(new HttpHandler()); pipeLine.addLast(new WebSocketHandler()); } }

위 소스코드를 보게 되면 pipeLine에 HttpRequestDecoder, HttpObjectAggergator, CorsHandler, HttpResponseEncoder, WebSocketServerProtocolHandler를 사용하였다.

HttpRequestDecoder, HttpResponseEncoder는 Netty서버가 http요청 및 응답을 할 수 있도록 요청 및 응답을 변환시켜주는 역할을 하며 CorsHandler는 도메인이 다른 서버들간의 호출을 위해서 사용하였다. 또한 WebSocketServerProtocolHandler는 구현할 Push Server가 WebSocket을 사용할 것인데 WebSocket을 사용하기 위한 프로토콜을 등록해 준 것이다.

이 후에 두개의 HttpServer, WebSocketHandler는 들어오는 요청에 대한 데이터 처리 및 응답을 하기 위해 직접 구현한 클래스이다.

HttpHandler의 구현

package com.netty.handler; import java.io.UnsupportedEncodingException; import java.net.URLDecoder; import java.util.HashMap; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentMap; import com.fasterxml.jackson.databind.ObjectMapper; import com.netty.singletone.UserWebSocketMultiMap; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; import io.netty.handler.codec.http.DefaultFullHttpResponse; import io.netty.handler.codec.http.FullHttpMessage; import io.netty.handler.codec.http.FullHttpResponse; import io.netty.handler.codec.http.HttpHeaders; import io.netty.handler.codec.http.HttpRequest; import io.netty.handler.codec.http.HttpResponseStatus; import io.netty.handler.codec.http.HttpVersion; import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; import io.netty.handler.codec.http.websocketx.WebSocketFrame; import io.netty.util.CharsetUtil; public class HttpHandler extends SimpleChannelInboundHandler<FullHttpMessage>{ @Override protected void channelRead0(ChannelHandlerContext ctx, FullHttpMessage msg) throws Exception { ConcurrentMap<String, ConcurrentMap<String, ChannelHandlerContext>> userChannelMap = UserWebSocketMultiMap.getInsatance(); ObjectMapper om = new ObjectMapper(); HttpRequest httpRequest = null; if(msg instanceof HttpRequest) { httpRequest = (HttpRequest) msg; } String uri = httpRequest.uri(); System.out.println(uri); Map<String, String> paramMap = paramGetter(uri); String userId = paramMap.get("userId"); String jsonResult = URLDecoder.decode(om.writeValueAsString(paramMap), "UTF-8"); if(userChannelMap.containsKey(userId)) { ConcurrentMap<String, ChannelHandlerContext> channelMap = userChannelMap.get(userId); Set<String> keySets = channelMap.keySet(); for(String key : keySets) { WebSocketFrame wsf = new TextWebSocketFrame(jsonResult); ChannelHandlerContext chc = channelMap.get(key); // Event send to next event handler chc.fireChannelActive(); chc.channel().writeAndFlush(wsf); } } FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.copiedBuffer("success", CharsetUtil.UTF_8)); response.headers().set(HttpHeaders.Names.CONTENT_TYPE, "application/json; charset=UTF-8"); response.headers().set("Access-Control-Allow-Origin", "null"); response.headers().set("Access-Control-Allow-Methods", "POST, GET, OPTIONS, DELETE"); response.headers().set("Access-Control-Max-Age", "3600"); response.headers().set("Access-Control-Allow-Headers", "Content-Type, Accept, X-Requested-With, remember-me"); response.headers().set("Access-Control-Allow-Credentials", "true"); response.headers().set("Access-Control-Expose-Headers", "Access-Control-Allow-Origin,Access-Control-Allow-Credentials"); //http 통신을 종료 하려면 ChannelFutureListener의 Close로 연결을 종료 시켜주어야 한다. ctx.write(response).addListener(ChannelFutureListener.CLOSE); } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ctx.flush(); } private Map<String, String> paramGetter(String uri) throws UnsupportedEncodingException { Map<String, String> paramMap = new HashMap<>(); String params = uri.substring(uri.indexOf("?") + 1, uri.length()); String[] paramsArray = params.split("&"); for(String param : paramsArray) { String[] paramArray = param.split("="); if(paramArray[0].equals("content")) { String content = URLDecoder.decode(paramArray[1], "UTF-8"); paramMap.put(paramArray[0], content); } paramMap.put(paramArray[0], paramArray[1]); } return paramMap; } }

HttpHandler를 구현하기 위해 SimpleChannelInboundHandler를 구현하였는데 뒤에 제네릭으로 준 Type을 보게 되면 FullHttpMessage를 사용한 것을 볼 수 있다. SimpleChannelInboundHandler를 구현할 때는 통신하고자 하는 방법에 정확히 맞는 Type을 제네릭타입으로 넘겨주어야이벤트에 의해 메소드가 동작한다. 또한 여기서 userChannelMap을 싱글턴으로 받아 오고 있는데 이유는 사용자계정으로 가지는 channel에 대해서 모두 Push를 주기 위함이다. 이렇게 한 이유는 Facebook의 경우 브라우저를 여러개 돌리더라도 다 같이 알림을 받기 때문이다.

그리고 Thread Safe하기 위해서 ConcurrentMap을 사용하였다.

주의 깊게 봐야하는 부분 49~64Line에 있는데 그것은 들어온 사용자 아이디가 있을때 Map에 등록된 모든 Channel에 메시지를 보내기 위해fireChnnelActive를 동작시켜 채널을 활성화 시키는 부분이다.

그리고 Http응답을 보낸 뒤 연결 종료를 위해 Listener를 등록을 한다.

WebSocketHandler의 구현

package com.netty.handler; import java.nio.charset.Charset; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import com.netty.singletone.UserWebSocketMultiMap; import com.netty.singletone.UserWithChannelIdMap; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; import io.netty.handler.codec.http.websocketx.WebSocketFrame; public class WebSocketHandler extends SimpleChannelInboundHandler<WebSocketFrame>{ @Override protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame msg) throws Exception { String initUserId = ((ByteBuf) msg.content()).toString(Charset.defaultCharset()); if(this.registryUserChannel(initUserId, ctx)) { System.out.println("registry"); }; System.out.println(initUserId + " : " + ctx); } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ctx.flush(); } @Override public void channelUnregistered(ChannelHandlerContext ctx) throws Exception { ConcurrentMap<String, String> channelIdMap = UserWithChannelIdMap.getInstance(); ConcurrentMap<String, ConcurrentMap<String, ChannelHandlerContext>> userChannelMap = UserWebSocketMultiMap.getInsatance(); String channelId = ctx.channel().id().toString(); String userId = ""; if(channelIdMap.containsKey(channelId)) { userId = channelIdMap.get(channelId); } if(userChannelMap.containsKey(userId)) { ConcurrentMap<String, ChannelHandlerContext> channelMap = userChannelMap.get(userId); channelMap.remove(channelId); System.out.println("remove channel"); } super.channelUnregistered(ctx); } /** * * @param userId * @param chc * @return boolean */ private boolean registryUserChannel(String userId, ChannelHandlerContext chc) { String channelId = chc.channel().id().toString(); ConcurrentMap<String, String> channelIdMap = UserWithChannelIdMap.getInstance(); ConcurrentMap<String, ConcurrentMap<String, ChannelHandlerContext>> userChannelMap = UserWebSocketMultiMap.getInsatance(); ConcurrentMap<String, ChannelHandlerContext> channelMap = null; if(userChannelMap.get(userId) == null) { channelMap = new ConcurrentHashMap<>(); } else { channelMap = userChannelMap.get(userId); } channelIdMap.put(channelId, userId); channelMap.put(channelId, chc); userChannelMap.put(userId, channelMap); return true; } @Override public void channelRegistered(ChannelHandlerContext ctx) throws Exception { return; } }

해당 클래스는 WebSocket 연결이 들어오는 채널에 대해서 Map에 등록하고 해제되는 채널에 대해서는 Map에서 제거하는 역할을 하게끔 구현을 하였다.

동작결과 

결론

간단하게 네티를 이용해서 PushServer를 구현해 보고 테스트로 만들고 있는 사이트에 적용해 보았다. 문제점은 WebSocket을 사용하여 구현을 하였지만 WebSocket이 지원되지 않는 브라우저가 존재하며 이 경우 PushServer가 요청을 받았을 때 브라우저 종류를 판별해 LongPolling 방식의 로직을 적용 해주면 될 것 같다는 생각이 든다.

또한 간단하게 만들어서 알림이 Json형태인것이며 알림을 어떻게 관리할 것인지에 대해서도 정책을 세워야 한다. 예를 들어 실시간으로 쌓이는 알림로그를 NoSql로 관리 할 것인가? RDB에 관리 할 것인가 부터 여러가지 고민을 해야한다.