package com.imooc.netty; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelFuture; 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.FullHttpRequest; import io.netty.handler.codec.http.HttpResponseStatus; import io.netty.handler.codec.http.HttpVersion; import io.netty.handler.codec.http.websocketx.*; import io.netty.util.CharsetUtil; import java.util.Date; /** * 接收/处理/响应客户端websocket请求的核心业务处理类 * * @author liuyazhuang */ public class MyWebSocketHandler extends SimpleChannelInboundHandler<Object> { private WebSocketServerHandshaker handshaker; private static final String WEB_SOCKET_URL = "ws://localhost:8888/websocket"; //客户端与服务端创建连接的时候调用 @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { NettyConfig.group.add(ctx.channel()); System.out.println("客户端与服务端连接开启..."); } //客户端与服务端断开连接的时候调用 @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { NettyConfig.group.remove(ctx.channel()); System.out.println("客户端与服务端连接关闭..."); } //服务端接收客户端发送过来的数据结束之后调用 @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ctx.flush(); } //工程出现异常的时候调用 @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { cause.printStackTrace(); ctx.close(); } //服务端处理客户端websocket请求的核心方法 @Override protected void messageReceived(ChannelHandlerContext context, Object msg) throws Exception { // 传统的HTTP接入 //第一次握手请求消息由HTTP协议承载,所以它是一个HTTP消息,执行handleHttpRequest方法来处理WebSocket握手请求。 if (msg instanceof FullHttpRequest) { handHttpRequest(context, (FullHttpRequest) msg); } // WebSocket接入 // 客户端通过文本框提交请求消息给服务端,WebSocketServerHandler接收到的是已经解码后的WebSocketFrame消息。 else if (msg instanceof WebSocketFrame) { handWebsocketFrame(context, (WebSocketFrame) msg); } } /** * 处理客户端向服务端发起http握手请求的业务 * * @param ctx * @param req */ private void handHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) { //如果不是WebSocket握手请求消息,那么就返回 HTTP 400 BAD REQUEST 响应给客户端。 if (!req.getDecoderResult().isSuccess() || !("websocket".equals(req.headers().get("Upgrade")))) { sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST)); return; } //如果是握手请求,那么就进行握手 WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory( WEB_SOCKET_URL, null, false); handshaker = wsFactory.newHandshaker(req); if (handshaker == null) { WebSocketServerHandshakerFactory.sendUnsupportedWebSocketVersionResponse(ctx.channel()); } else { // 通过它构造握手响应消息返回给客户端, // 同时将WebSocket相关的编码和解码类动态添加到ChannelPipeline中,用于WebSocket消息的编解码, // 添加WebSocketEncoder和WebSocketDecoder之后,服务端就可以自动对WebSocket消息进行编解码了 handshaker.handshake(ctx.channel(), req); } } /** * 处理客户端与服务端之前的websocket业务 * * @param ctx * @param frame */ private void handWebsocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) { //判断是否是关闭websocket的指令 if (frame instanceof CloseWebSocketFrame) { handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain()); } //判断是否是ping消息 if (frame instanceof PingWebSocketFrame) { ctx.channel().write(new PongWebSocketFrame(frame.content().retain())); return; } //判断是否是二进制消息,如果是二进制消息,抛出异常 if (!(frame instanceof TextWebSocketFrame)) { System.out.println("目前我们不支持二进制消息"); throw new RuntimeException("【" + this.getClass().getName() + "】不支持消息"); } //返回应答消息 //获取客户端向服务端发送的消息 String request = ((TextWebSocketFrame) frame).text(); System.out.println("服务端收到客户端的消息====>>>" + request); TextWebSocketFrame tws = new TextWebSocketFrame(new Date().toString() + ctx.channel().id() + " ===>>> " + request); //群发,服务端向每个连接上来的客户端群发消息 NettyConfig.group.writeAndFlush(tws); } /** * 服务端向客户端响应消息 * * @param ctx * @param req * @param res */ private void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse res) { // 返回应答给客户端 if (res.getStatus().code() != 200) { ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8); res.content().writeBytes(buf); buf.release(); } // 如果是非Keep-Alive,关闭连接 ChannelFuture f = ctx.channel().writeAndFlush(res); if (res.getStatus().code() != 200) { f.addListener(ChannelFutureListener.CLOSE); } } }
最后,编写入口程序:启动WebSocket服务
package com.imooc.netty; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.Channel; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.nio.NioServerSocketChannel; /** * 程序的入口,负责启动应用 * @author liuyazhuang * */ public class Main { public static void main(String[] args) { EventLoopGroup bossGroup = new NioEventLoopGroup(); EventLoopGroup workGroup = new NioEventLoopGroup(); try { ServerBootstrap b = new ServerBootstrap(); b.group(bossGroup, workGroup); b.channel(NioServerSocketChannel.class); b.childHandler(new MyWebSocketChannelHandler()); System.out.println("服务端开启等待客户端连接...."); Channel ch = b.bind(8888).sync().channel(); ch.closeFuture().sync(); } catch (Exception e) { e.printStackTrace(); }finally{ //优雅的退出程序 bossGroup.shutdownGracefully(); workGroup.shutdownGracefully(); } } }
客户端代码:
<html> <head> <meta http-equiv="Content-Type" content="text/html; charset = utf-8"/> <title>WebSocket客户端</title> <script type="text/javascript"> var socket; if(!window.WebSocket){ window.WebSocket = window.MozWebSocket; } if(window.WebSocket){ socket = new WebSocket("ws://localhost:8888/websocket"); socket.onmessage = function(event){ var ta = document.getElementById('responseContent'); ta.value += event.data + "\r\n"; }; socket.onopen = function(event){ var ta = document.getElementById('responseContent'); ta.value = "你当前的浏览器支持WebSocket,请进行后续操作\r\n"; }; socket.onclose = function(event){ var ta = document.getElementById('responseContent'); ta.value = ""; ta.value = "WebSocket连接已经关闭\r\n"; }; }else{ alert("您的浏览器不支持WebSocket"); } function send(message){ if(!window.WebSocket){ return; } if(socket.readyState == WebSocket.OPEN){ socket.send(message); }else{ alert("WebSocket连接没有建立成功!!"); } } </script> </head> <body> <form onSubmit="return false;"> <input type = "text" name = "message" value = ""/> <br/><br/> <input type = "button" value = "发送WebSocket请求消息" onClick = "send(this.form.message.value)"/> <hr color="red"/> <h2>客户端接收到服务端返回的应答消息</h2> <textarea id = "responseContent" style = "width:1024px; height:300px"></textarea> </form> </body> </html>
实际使用WebSocket
上面的例子讲解了Netty实现WebSocket,一般我们使用WebSocket不会自己来实现,都是用现成的工具包来进行实现。
我查到的常用的方式有两种:
- Tomcat实现WebSocket
- 整合Spring实现WebSocket
这一部分我就不再赘述了,等我用到的时候再补教程吧,先mark下相关的博客:
- Tomcat实现
- Spring整合和Tomcat实现
- 基于Java实现
总结
WebSocket最大的特点就是长连接,能够实时推送数据。