WebSocketServer
package com.zhaowb.netty.ch11; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.*; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.codec.http.HttpObjectAggregator; import io.netty.handler.codec.http.HttpServerCodec; import io.netty.handler.stream.ChunkedWriteHandler; public class WebSocketServer { public void run(int port) throws Exception { EventLoopGroup bossGroup = new NioEventLoopGroup(); EventLoopGroup workerGroup = new NioEventLoopGroup(); try { ServerBootstrap b = new ServerBootstrap(); b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); pipeline.addLast("http-codec", new HttpServerCodec());// 添加 HttpServerCodec 将请求和应答消息编码或解码为HTTP 消息 pipeline.addLast("aggregator", new HttpObjectAggregator(65536)); // 增加HttpObjectAggregator ，将HTTP 消息的多个部分组合成一条完整的HTTP消息 pipeline.addLast("http-chunked", new ChunkedWriteHandler());// 添加ChunkedWriteHandler，来向客户端发送HTML5文件，主要用于支持浏览器和服务端进行WebSocket 通信。 pipeline.addLast("handle", new WebSocketServerHandler()); // 增加WebSocket 服务端的handler } }); Channel ch = b.bind(port).sync().channel(); System.out.println("Web socket server started at port : " + port + '.'); System.out.println("Open your browser and navigate to http://localhost:" + port + '/'); ch.closeFuture().sync(); } catch (Exception e) { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } public static void main(String[] args) throws Exception { int port = 8080; if (args.length > 0) { try { port = Integer.parseInt(args[0]); } catch (NumberFormatException e) { e.printStackTrace(); } } new WebSocketServer().run(port); } }
WebSocketServerHandler
package com.zhaowb.netty.ch11; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.*; import io.netty.handler.codec.http.*; import io.netty.handler.codec.http.websocketx.*; import io.netty.util.CharsetUtil; import java.util.Date; import java.util.logging.Level; import java.util.logging.Logger; import static io.netty.handler.codec.http.HttpHeaders.isKeepAlive; import static io.netty.handler.codec.http.HttpHeaders.setContentLength; import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST; public class WebSocketServerHandler extends SimpleChannelInboundHandler<Object> { private static final Logger logger = Logger.getLogger(WebSocketServerHandler.class.getName()); private WebSocketServerHandshaker handshaker; /** * 第一次握手请求消息由HTTP协议承载，所有它是一个HTTP 消息，执行handleHttpRequest方法来处理WebSocket 握手请求。 * * @param ctx * @param msg * @throws Exception */ @Override protected void messageReceived(ChannelHandlerContext ctx, Object msg) throws Exception { // 传统HTTP 接入 if (msg instanceof FullHttpRequest) { handleHttpRequest(ctx, (FullHttpRequest) msg); } else if (msg instanceof WebSocketFrame) { handleWebSocketFrame(ctx, (WebSocketFrame) msg); } } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ctx.flush(); } private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) { /** * 对握手请求消息进行判断，如果消息头中没有包含Upgrade 字段或者它的值不是websocket。则返回HTTP 400响应 */ // 如果HTTP解码失败，返回HTTP异常 if (!req.getDecoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))) { sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, BAD_REQUEST)); return; } // 构造握手响应返回，本机测试。 /** * 握手请求简单校验通过之后，开始构造握手工厂，创建握手处理类 WebSocketServerHandshaker，通过它构造握手响应消息 * 返回给客户端，同时将WebSocket相关的编码和解码类动态添加到ChannelPipeline中，用于WebSocket消息的编解码 */ WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory("ws:localhost:8080/websocket", null, false); handshaker = wsFactory.newHandshaker(req); if (handshaker == null) { WebSocketServerHandshakerFactory.sendUnsupportedWebSocketVersionResponse(ctx.channel()); } else { handshaker.handshake(ctx.channel(), req); } } /** * 对WebSocket 请求消息进行处理，首先需要对控制帧进行判断，如果是关闭链路的控制信息，就调用WebSocketServerHandshaker * 的close 方法关闭WebSocket连接，如果是维持链路的Ping消息，则构造Pong消息返回。 * @param ctx * @param frame */ private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) { // 判断链路是否关闭 if (frame instanceof CloseWebSocketFrame) { handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain()); return; } // 支持文本信息，不支持二进制 if (!(frame instanceof TextWebSocketFrame)) { throw new UnsupportedOperationException(String.format("%s frame types not supported", frame.getClass().getName())); } // 返回应答消息 // 从TextWebSocketFrame中获取请求消息字符串，对它处理通过后构造新的TextWebSocketFrame消息，返回给客户端， // 由于握手应答时动态增加了TextWebSocketFrame的编码类，所以直接发送TextWebSocketFrame对象。 String request = ((TextWebSocketFrame) frame).text(); if (logger.isLoggable(Level.FINE)) { logger.fine(String.format("%s received %s", ctx.channel())); } ctx.channel().write(new TextWebSocketFrame(request + " , 欢迎使用 Netty WebScoket服务，现在时刻:" + new Date().toString())); } private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) { // 返回应答给客户端 if (res.getStatus().code() != 200) { ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8); res.content().writeBytes(buf); buf.release(); setContentLength(res, res.content().readableBytes()); } // 如果是非 Keep-Alive ,关闭连接 ChannelFuture f = ctx.channel().writeAndFlush(res); if (!isKeepAlive(req) || res.getStatus().code() != 200) { f.addListener(ChannelFutureListener.CLOSE); } } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { cause.printStackTrace(); ctx.close(); } }
test.html
<!DOCTYPE html> <html> <head> <meta charset="UTF-8"> Netty WebSocket 时间服务器 </head> <br> <body> <br> <script type="text/javascript"> var socket; if (!window.WebSocket) { window.WebSocket = window.MozWebSocket; } if (window.WebSocket) { socket = new WebSocket("ws://localhost:8080/websocket"); socket.onmessage = function (event) { var ta = document.getElementById('responseText'); ta.value = ""; ta.value = event.data }; socket.onopen = function (event) { var ta = document.getElementById('responseText'); ta.value = "打开WebSocket服务正常，浏览器支持WebSocket!"; }; socket.onclose = function (event) { var ta = document.getElementById('responseText'); ta.value = ""; ta.value = "WebSocket 关闭!"; }; } else { alert("抱歉，您的浏览器不支持WebSocket协议!"); } function send(message) { if (!window.WebSocket) { return; } if (socket.readyState == WebSocket.OPEN) { socket.send(message); } else { alert("WebSocket连接没有建立成功!"); } } </script> <form onsubmit="return false;"> <input type="text" name="message" value="Netty最佳实践"/> <br><br> <input type="button" value="发送WebSocket请求消息" onclick="send(this.form.message.value)"/> <hr color="blue"/> <h3>服务端返回的应答消息</h3> <textarea id="responseText" style="500px;height:300px;"></textarea> </form> </body> </html>
启动的时候先启动 WebSocketServer，然后用浏览器打开test.html,点击 “发送WebSocket请求消息” 就会返回