netty websocket

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); ???}}

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(); ???}}

<!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="width:500px;height:300px;"></textarea></form></body></html>