最近在研究Netty來構建SOA架構，其中也包括了前端接入的HTTP/WebSocket方面的接入響應，而WebSocket方面的接入響應對於移動端的訊息推送研發至關重要，這裡就將在這塊研發時的非同步socket響應服務例子程式筆記記錄下來，系統總共分為4個處理類，即：

HttpRequestHandler  --  HTTP請求處理類
TextWebSocketFrameHandler -- 對應Text訊息的處理類
WebSocketServer  -- 系統主類
WebSocketServerInitializer -- 服務主程式的初始化類

WebSocketServer 類程式碼：

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public final class WebSocketServer {

 private int port = 0;

 public WebSocketServer(int port) {
  this.port = port;
 }

 public void run() throws Exception {
  EventLoopGroup bossGroup = new NioEventLoopGroup();
  EventLoopGroup workerGroup = new NioEventLoopGroup();
  try {
   ServerBootstrap b = new ServerBootstrap();
   b.group(bossGroup, workerGroup)
     .channel(NioServerSocketChannel.class)
     .childHandler(new WebSocketServerInitializer())
     .option(ChannelOption.SO_BACKLOG, 128)
     .childOption(ChannelOption.SO_KEEPALIVE, true);
   
   System.out.println("WebsocketChatServer 啟動了");

   // 繫結埠，開始接收進來的連線
   ChannelFuture f = b.bind(port).sync();

   // 等待伺服器 socket 關閉 。在這個例子中，這不會發生，但你可以優雅地關閉你的伺服器。
   f.channel().closeFuture().sync();
  } finally {
   workerGroup.shutdownGracefully();
   bossGroup.shutdownGracefully();

   System.out.println("WebsocketChatServer 關閉了");
  }
 }
 
 public static void main(String[] args) throws Exception {
  int port = 0;
  
        if (args.length > 0) {
            port = Integer.parseInt(args[0]);
        } else {
            port = 8080;
        }
       
        // start current instance
        new WebSocketServer(port).run();
 }
 
}

HttpRequestHandler類程式碼：

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.DefaultFileRegion;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.DefaultHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpResponse;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.LastHttpContent;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.stream.ChunkedNioFile;

import java.io.File;
import java.io.RandomAccessFile;
import java.net.URISyntaxException;
import java.net.URL;

public class HttpRequestHandler extends SimpleChannelInboundHandler<FullHttpRequest> {

 private final String wsUri;
 private static final File INDEX;
 
 // static HTTP request handling operation.
 static {
  URL location = HttpRequestHandler.class.getProtectionDomain()
    .getCodeSource().getLocation();
  try {
   String path = location.toURI() + "WebSocketClient.html";
   path = !path.contains("file:") ? path : path.substring(5);
   INDEX = new File(path);
  } catch (URISyntaxException e) {
   throw new IllegalStateException("Unable to locate WebsocketChatClient.html", e);
  }
 }
 
 // constructor function call for current class
 public HttpRequestHandler(String wsUri) {
  this.wsUri = wsUri;
 }

 @Override
 public void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request)
   throws Exception {
  if (wsUri.equalsIgnoreCase(request.getUri())) {
   ctx.fireChannelRead(request.retain());
  } else {
   if (HttpHeaders.is100ContinueExpected(request)) {
    send100Continue(ctx);
   }
   
   RandomAccessFile file = new RandomAccessFile(INDEX, "r");
   HttpResponse response = new DefaultHttpResponse(
     request.getProtocolVersion(), HttpResponseStatus.OK);
   response.headers().set(HttpHeaders.Names.CONTENT_TYPE,
     "text/html; charset=UTF-8");

   boolean keepAlive = HttpHeaders.isKeepAlive(request);
   if (keepAlive) {
    response.headers().set(HttpHeaders.Names.CONTENT_LENGTH,  file.length());
    response.headers().set(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.KEEP_ALIVE);
   }
   ctx.write(response);

   if (ctx.pipeline().get(SslHandler.class) == null) {
    ctx.write(new DefaultFileRegion(file.getChannel(), 0, file.length()));
   } else {
    ctx.write(new ChunkedNioFile(file.getChannel()));
   }
   ChannelFuture future = ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
   if (!keepAlive) {
    future.addListener(ChannelFutureListener.CLOSE);
   }

   file.close();
  }
 }

 private static void send100Continue(ChannelHandlerContext ctx) {
  FullHttpResponse response = new DefaultFullHttpResponse(
    HttpVersion.HTTP_1_1, HttpResponseStatus.CONTINUE);
  ctx.writeAndFlush(response);
 }

 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
   throws Exception {
  Channel incoming = ctx.channel();
  System.out.println("Client:" + incoming.remoteAddress() + "異常");
  
  // 當出現異常就關閉連線
  cause.printStackTrace();
  ctx.close();
 }

}

TextWebSocketFrameHandler類程式碼：

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.concurrent.GlobalEventExecutor;

public class TextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
 
 public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
 
 @Override
 protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
  Channel incoming = ctx.channel();
  for (Channel channel : channels) {
   if (channel != incoming) {
    //channel.writeAndFlush(new TextWebSocketFrame("[" + incoming.remoteAddress() + "]" + msg.text()));
   } else {
    channel.writeAndFlush(new TextWebSocketFrame("[伺服器端返回]：" + msg.text()));
    
    //output current message to context.
    StringBuffer sb = new StringBuffer();
    sb.append(incoming.remoteAddress()).append("->").append(msg.text());
    System.out.println(sb.toString());
   }
  }
 }
 
 @Override
 public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
  Channel incoming = ctx.channel();
  for (Channel channel : channels) {
   channel.writeAndFlush(new TextWebSocketFrame("[SERVER] - "  + incoming.remoteAddress() + " 加入"));
  }
  channels.add(ctx.channel());
  System.out.println("Client:" + incoming.remoteAddress() + "加入");
 }
 
 @Override
 public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
  Channel incoming = ctx.channel();
  for (Channel channel : channels) {
   channel.writeAndFlush(new TextWebSocketFrame("[SERVER] - "  + incoming.remoteAddress() + " 離開"));
  }
  System.out.println("Client:" + incoming.remoteAddress() + "離開");
  channels.remove(ctx.channel());
 }
 
 @Override
 public void channelActive(ChannelHandlerContext ctx) throws Exception {
  Channel incoming = ctx.channel();
  System.out.println("Client:" + incoming.remoteAddress() + "線上");
 }
 
 @Override
 public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  Channel incoming = ctx.channel();
  System.out.println("Client:" + incoming.remoteAddress() + "掉線");
 }

 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
   throws Exception {
  Channel incoming = ctx.channel();
  System.out.println("Client:" + incoming.remoteAddress() + "異常");
  
  // 當出現異常就關閉連線
  cause.printStackTrace();
  ctx.close();
 }

}

WebSocketServerInitializer類程式碼：

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.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;

public class WebSocketServerInitializer extends ChannelInitializer<SocketChannel> {
 
 @Override
    public void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
       
        pipeline.addLast(new HttpServerCodec());
        pipeline.addLast(new HttpObjectAggregator(64*1024));
        pipeline.addLast(new ChunkedWriteHandler());
        pipeline.addLast(new HttpRequestHandler("/ws"));
        pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
        pipeline.addLast(new TextWebSocketFrameHandler());
       
    }
 
}

客戶端將通過HTML5的WebSocket來撰寫，特別注意WebSocket需要Firefox/Chrome/IE 比較高的版本才能夠支援，譬如IE11.

<!DOCTYPE html>
<html>
<head>
<meta charset="GBK">
<title>WebSocket聊天程式</title>
</head>
<body>
    <script type="text/javascript">
        var socket;
       
        function connectServer() {
         if (!window.WebSocket) {
             window.WebSocket = window.MozWebSocket;
         }
         if (window.WebSocket) {
             socket = new WebSocket("ws://localhost:8080/ws");
             socket.onmessage = function(event) {
                 var ta = document.getElementById('responseText');
                 ta.value = ta.value + '\n' + event.data
             };
             socket.onopen = function(event) {
                 var ta = document.getElementById('responseText');
                 ta.value = "連線開啟!";
             };
             socket.onclose = function(event) {
                 var ta = document.getElementById('responseText');
                 ta.value = ta.value + "連線被關閉";
             };
         } else {
             alert("你的瀏覽器不支援 WebSocket！");
         }
        }
       
        function send(message) {
            if (!window.WebSocket) {
                return;
            }
            if (socket.readyState == WebSocket.OPEN) {
                socket.send(message);
            } else {
                alert("連線沒有開啟.");
            }
        }
    </script>
    <form onsubmit="return false;">
     <input type="button" onclick="javascript:connectServer()" value="連線伺服器">
        <h3>WebSocket 聊天室：</h3>
        <textarea id="responseText" style="width: 500px; height: 300px;"></textarea>
        <br>
        <input type="text" name="message"  style="width: 300px" value="聊天文字在此....">
        <input type="button" value="傳送訊息" onclick="send(this.form.message.value)">
        <input type="button" onclick="javascript:document.getElementById('responseText').value=''" value="清空聊天記錄">
    </form>
    <br>
    <br>
</body>
</html>