Netty 系列八（基於 WebSocket 的簡單聊天室）.
摘要：
一、前言
之前寫過一篇 Spring 整合 WebSocket 協議的文章 ——Spring訊息之WebSocket ，所以對於 WebSocket 協議的介紹就不多說了，可以參考這篇文章。這裡只做一些補充說明。另外，Netty 對 WebSocket 協議...
一、前言
之前寫過一篇 Spring 整合 WebSocket 協議的文章 —— Spring訊息之WebSocket ， 所以對於 WebSocket 協議的介紹就不多說了，可以參考這篇文章。這裡只做一些補充說明。另外，Netty 對 WebSocket 協議的支援要比 Spring 好太多了，用起來舒服的多。
WebSocket 以幀的方式傳輸資料，每一幀代表訊息的一部分。一個完整的訊息可能會包含許多幀。
由 IETF 釋出的 WebSocket RFC，定義了 6 種幀， Netty 為它們每種都提供了一個 POJO 實現。下表列出了這些幀型別，並描述了它們的用法。
二、聊天室功能說明
1、A、B、C 等所有使用者都可以加入同一個聊天室。
2、A 傳送的訊息，B、C 可以同時收到，但是 A 收不到自己傳送的訊息。
3、當用戶長時間沒有傳送訊息，系統將把他踢出聊天室。
三、聊天室功能實現
1、Netty 版本
<dependency> <groupId>io.netty</groupId> <artifactId>netty-all</artifactId> <version>5.0.0.Alpha2</version> </dependency>
2、處理 HTTP 協議的 ChannelHandler —— 非 WebSocket 協議的請求，返回 index.html 頁面
public class HttpRequestHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
private final String wsUri;
private static File INDEX;
static {
URL location = HttpRequestHandler.class.getProtectionDomain().getCodeSource().getLocation();
try {
String path = location.toURI() + "index.html";
path = !path.contains("file:") ? path : path.substring(5);
INDEX = new File(path);
} catch (URISyntaxException e) {
e.printStackTrace();
}
}
public HttpRequestHandler(String wsUri) {
this.wsUri = wsUri;
}
@Override
protected void messageReceived(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception {
// 如果請求了Websocket，協議升級，增加引用計數(呼叫retain())，並將他傳遞給下一個 ChannelHandler
// 之所以需要呼叫 retain() 方法，是因為呼叫 channelRead() 之後，資源會被 release() 方法釋放掉，需要呼叫 retain() 保留資源
if (wsUri.equalsIgnoreCase(request.uri())) {
ctx.fireChannelRead(request.retain());
} else {
//處理 100 Continue 請求以符合 HTTP 1.1 規範
if (HttpHeaderUtil.is100ContinueExpected(request)) {
send100Continue(ctx);
}
// 讀取 index.html
RandomAccessFile randomAccessFile = new RandomAccessFile(INDEX, "r");
HttpResponse response = new DefaultHttpResponse(request.protocolVersion(), HttpResponseStatus.OK);
HttpHeaders headers = response.headers();
//在該 HTTP 頭資訊被設定以後，HttpRequestHandler 將會寫回一個 HttpResponse 給客戶端
headers.set(HttpHeaderNames.CONTENT_TYPE, "text/html; charset=UTF-8");
boolean keepAlive = HttpHeaderUtil.isKeepAlive(request);
if (keepAlive) {
headers.setLong(HttpHeaderNames.CONTENT_LENGTH, randomAccessFile.length());
headers.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
}
ctx.write(response);
//將 index.html 寫給客戶端
if (ctx.pipeline().get(SslHandler.class) == null) {
ctx.write(new DefaultFileRegion(randomAccessFile.getChannel(), 0, randomAccessFile.length()));
} else {
ctx.write(new ChunkedNioFile(randomAccessFile.getChannel()));
}
//寫 LastHttpContent 並沖刷至客戶端，標記響應的結束
ChannelFuture channelFuture = ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
if (!keepAlive) {
channelFuture.addListener(ChannelFutureListener.CLOSE);
}
}
}
private 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 {
cause.printStackTrace();
ctx.close();
}
3、處理 WebSocket 協議的 ChannelHandler —— 處理 TextWebSocketFrame 的訊息幀
/**
* WebSocket 幀：WebSocket 以幀的方式傳輸資料，每一幀代表訊息的一部分。一個完整的訊息可能會包含許多幀
*/
public class TextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
private final ChannelGroup group;
public TextWebSocketFrameHandler(ChannelGroup group) {
this.group = group;
}
@Override
protected void messageReceived(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
//增加訊息的引用計數（保留訊息），並將他寫到 ChannelGroup 中所有已經連線的客戶端
Channel channel = ctx.channel();
//自己傳送的訊息不返回給自己
group.remove(channel);
group.writeAndFlush(msg.retain());
group.add(channel);
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
//是否握手成功，升級為 Websocket 協議
if (evt == WebSocketServerProtocolHandler.ServerHandshakeStateEvent.HANDSHAKE_COMPLETE) {
// 握手成功，移除 HttpRequestHandler，因此將不會接收到任何訊息
// 並把握手成功的 Channel 加入到 ChannelGroup 中
ctx.pipeline().remove(HttpRequestHandler.class);
group.writeAndFlush(new TextWebSocketFrame("Client " + ctx.channel() + " joined"));
group.add(ctx.channel());
} else if (evt instanceof IdleStateEvent) {
IdleStateEvent stateEvent = (IdleStateEvent) evt;
if (stateEvent.state() == IdleState.READER_IDLE) {
group.remove(ctx.channel());
ctx.writeAndFlush(new TextWebSocketFrame("由於您長時間不線上，系統已自動把你踢下線！")).addListener(ChannelFutureListener.CLOSE);
}
} else {
super.userEventTriggered(ctx, evt);
}
}
}
WebSocket 協議升級完成之後， WebSocketServerProtocolHandler 將會把 HttpRequestDecoder 替換為 WebSocketFrameDecoder，把 HttpResponseEncoder 替換為WebSocketFrameEncoder。為了效能最大化，它將移除任何不再被 WebSocket 連線所需要的 ChannelHandler。這也包括了 HttpObjectAggregator 和 HttpRequestHandler 。
4、ChatServerInitializer —— 多個 ChannelHandler 合併成 ChannelPipeline 鏈
public class ChatServerInitializer extends ChannelInitializer<Channel> {
private final ChannelGroup group;
private static final int READ_IDLE_TIME_OUT = 60; // 讀超時
private static final int WRITE_IDLE_TIME_OUT = 0;// 寫超時
private static final int ALL_IDLE_TIME_OUT = 0; // 所有超時
public ChatServerInitializer(ChannelGroup group) {
this.group = group;
}
@Override
protected void initChannel(Channel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new HttpServerCodec());
pipeline.addLast(new ChunkedWriteHandler());
pipeline.addLast(new HttpObjectAggregator(64 * 1024));
// 處理那些不傳送到 /ws URI的請求
pipeline.addLast(new HttpRequestHandler("/ws"));
// 如果被請求的端點是 "/ws",則處理該升級握手
pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
// //當連線在60秒內沒有接收到訊息時，進會觸發一個 IdleStateEvent 事件，被 HeartbeatHandler 的 userEventTriggered 方法處理
pipeline.addLast(new IdleStateHandler(READ_IDLE_TIME_OUT, WRITE_IDLE_TIME_OUT, ALL_IDLE_TIME_OUT, TimeUnit.SECONDS));
pipeline.addLast(new TextWebSocketFrameHandler(group));
}
}
ChatServerInitializer.java
tips： 上面這些開箱即用 ChannelHandler 的作用，我就不一一介紹了，可以參考 上一篇文章 。
5、引導類 ChatServer
public class ChatServer {
private final ChannelGroup channelGroup = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE);
private final EventLoopGroup group = new NioEventLoopGroup();
private Channel channel;
public ChannelFuture start(InetSocketAddress address) {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(group)
.channel(NioServerSocketChannel.class)
.childHandler(new ChatServerInitializer(channelGroup));
ChannelFuture channelFuture = bootstrap.bind(address);
channelFuture.syncUninterruptibly();
channel = channelFuture.channel();
return channelFuture;
}
public void destroy() {
if (channel != null) {
channel.close();
}
channelGroup.close();
group.shutdownGracefully();
}
public static void main(String[] args) {
final ChatServer chatServer = new ChatServer();
ChannelFuture channelFuture = chatServer.start(new InetSocketAddress(9999));
// 返回與當前Java應用程式關聯的執行時物件
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
chatServer.destroy();
}
});
channelFuture.channel().closeFuture().syncUninterruptibly();
}
}
ChatServer.java
三、效果展示
在瀏覽器中輸入 http://127.0.0.1:9999 即可看到預先準備好的 index.html 頁面；訪問 ws://127.0.0.1:9999/ws （可隨意找一個 WebSocket 測試工具測試）即可加入聊天室。
有點 low 的聊天室總算是完成了，算是 Netty 對 HTTP 協議和 WebSocket 協議的一次實踐吧！雖然功能欠缺，但千里之行，始於足下！不積矽步，無以至千里；不積小流，無以成江海！
參考資料：《Netty IN ACTION》
演示原始碼： https://github.com/JMCuixy/NettyDemo/tree/master/src/main/java/org/netty/demo/chatroom