基于netty的websocket例子

package com.atguigu.netty.websocket;import javax.annotation.PostConstruct;import org.springframework.stereotype.Service;import io.netty.bootstrap.ServerBootstrap;import io.netty.channel.Channel;import io.netty.channel.ChannelInitializer;import io.netty.channel.EventLoopGroup;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;/*** ClassName:NettyServer 注解式随spring启动* Function: TODO ADD FUNCTION.* @author fwz*/@Servicepublic class NettyServer {public static void main(String[] args) {new NettyServer().run();}@PostConstructpublic void initNetty(){new Thread(){public void run() {new NettyServer().run();}}.start();}public void run(){System.out.println("===========================Netty端口启动========");// Boss线程：由这个线程池提供的线程是boss种类的，用于创建、连接、绑定socket， （有点像门卫）然后把这些socket传给worker线程池。// 在服务器端每个监听的socket都有一个boss线程来处理。在客户端，只有一个boss线程来处理所有的socket。EventLoopGroup bossGroup = new NioEventLoopGroup();// Worker线程：Worker线程执行所有的异步I/O，即处理操作EventLoopGroup workGroup = new NioEventLoopGroup();try {// ServerBootstrap 启动NIO服务的辅助启动类,负责初始话netty服务器，并且开始监听端口的socket请求ServerBootstrap b = new ServerBootstrap();b.group(bossGroup, workGroup);// 设置非阻塞,用它来建立新accept的连接,用于构造serversocketchannel的工厂类b.channel(NioServerSocketChannel.class);// ChildChannelHandler 对出入的数据进行的业务操作,其继承ChannelInitializerb.childHandler(new ChildChannelHandler());System.out.println("服务端开启等待客户端连接 ... ...");Channel ch = b.bind(8081).sync().channel();ch.closeFuture().sync();} catch (Exception e) {e.printStackTrace();}finally{bossGroup.shutdownGracefully();workGroup.shutdownGracefully();}}private class ChildChannelHandler extends ChannelInitializer<SocketChannel>{@Overrideprotected void initChannel(SocketChannel e) throws Exception {// 设置30秒没有读到数据，则触发一个READER_IDLE事件。// pipeline.addLast(new IdleStateHandler(30, 0, 0));// HttpServerCodec：将请求和应答消息解码为HTTP消息e.pipeline().addLast("http-codec",new HttpServerCodec());// HttpObjectAggregator：将HTTP消息的多个部分合成一条完整的HTTP消息e.pipeline().addLast("aggregator",new HttpObjectAggregator(65536));// ChunkedWriteHandler：向客户端发送HTML5文件e.pipeline().addLast("http-chunked",new ChunkedWriteHandler());// 在管道中添加我们自己的接收数据实现方法e.pipeline().addLast("handler",new MyWebSocketServerHandler());}}}

package com.atguigu.netty.websocket;import java.util.Date;import java.util.List;import java.util.Map;import java.util.logging.Level;import java.util.logging.Logger;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.HttpHeaderNames;import io.netty.handler.codec.http.HttpResponseStatus;import io.netty.handler.codec.http.HttpUtil;import io.netty.handler.codec.http.HttpVersion;import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;import io.netty.handler.codec.http.websocketx.WebSocketFrame;import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;import io.netty.util.AttributeKey;import io.netty.util.CharsetUtil;/** * ClassName:MyWebSocketServerHandler Function: TODO ADD FUNCTION. * * @author fwz */public class MyWebSocketServerHandler extends SimpleChannelInboundHandler<Object> { ???private static final Logger logger = Logger.getLogger(WebSocketServerHandshaker.class.getName()); ???private WebSocketServerHandshaker handshaker; ???/** ????* channel 通道 action 活跃的 当客户端主动链接服务端的链接后，这个通道就是活跃的了。也就是客户端与服务端建立了通信通道并且可以传输数据 ????*/ ???@Override ???public void channelActive(ChannelHandlerContext ctx) throws Exception {// 添加 ???????Global.group.add(ctx.channel()); ???????System.out.println("客户端与服务端连接开启：" + ctx.channel().remoteAddress().toString()); ???} ???/** ????* channel 通道 Inactive 不活跃的 ????* 当客户端主动断开服务端的链接后，这个通道就是不活跃的。也就是说客户端与服务端关闭了通信通道并且不可以传输数据 ????*/ ???@Override ???public void channelInactive(ChannelHandlerContext ctx) throws Exception {// 移除 ???????Global.group.remove(ctx.channel()); ???????System.out.println("客户端与服务端连接关闭：" + ctx.channel().remoteAddress().toString()); ???} ???/** ????* 接收客户端发送的消息 channel 通道 Read 读 ????* 简而言之就是从通道中读取数据，也就是服务端接收客户端发来的数据。但是这个数据在不进行解码时它是ByteBuf类型的 ????*/ ???@Override ???public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {// 传统的HTTP接入 ???????if (msg instanceof FullHttpRequest) { ???????????handleHttpRequest(ctx, ((FullHttpRequest) msg));// WebSocket接入 ???????} else if (msg instanceof WebSocketFrame) { ???????????System.out.println(handshaker.uri()); ???????????if ("anzhuo".equals(ctx.channel().attr(AttributeKey.valueOf("type")).get())) { ???????????????handlerWebSocketFrame(ctx, (WebSocketFrame) msg); ???????????} else { ???????????????handlerWebSocketFrame2(ctx, (WebSocketFrame) msg); ???????????} ???????} ???} ???/** ????* channel 通道 Read 读取 Complete 完成 在通道读取完成后会在这个方法里通知，对应可以做刷新操作 ctx.flush() ????*/ ???@Override ???public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ???????ctx.flush(); ???} ???private void handlerWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {// 判断是否关闭链路的指令 ???????if (frame instanceof CloseWebSocketFrame) { ???????????System.out.println(1); ???????????handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain()); ???????????return; ???????}// 判断是否ping消息 ???????if (frame instanceof PingWebSocketFrame) { ???????????ctx.channel().write(new PongWebSocketFrame(frame.content().retain())); ???????????return; ???????}// 本例程仅支持文本消息，不支持二进制消息 ???????if (!(frame instanceof TextWebSocketFrame)) { ???????????System.out.println("本例程仅支持文本消息，不支持二进制消息"); ???????????throw new UnsupportedOperationException( ???????????????????String.format("%s frame types not supported", frame.getClass().getName())); ???????}// 返回应答消息 ???????String request = ((TextWebSocketFrame) frame).text(); ???????System.out.println("服务端收到：" + request); ???????if (logger.isLoggable(Level.FINE)) { ???????????logger.fine(String.format("%s received %s", ctx.channel(), request)); ???????} ???????TextWebSocketFrame tws = new TextWebSocketFrame(new Date().toString() + ctx.channel().id() + "：" + request);// 群发 ???????Global.group.writeAndFlush(tws);// 返回【谁发的发给谁】// ctx.channel().writeAndFlush(tws); ???} ???private void handlerWebSocketFrame2(ChannelHandlerContext ctx, WebSocketFrame frame) {// 判断是否关闭链路的指令 ???????if (frame instanceof CloseWebSocketFrame) { ???????????handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain()); ???????????return; ???????}// 判断是否ping消息 ???????if (frame instanceof PingWebSocketFrame) { ???????????ctx.channel().write(new PongWebSocketFrame(frame.content().retain())); ???????????return; ???????}// 本例程仅支持文本消息，不支持二进制消息 ???????if (!(frame instanceof TextWebSocketFrame)) { ???????????System.out.println("本例程仅支持文本消息，不支持二进制消息"); ???????????throw new UnsupportedOperationException( ???????????????????String.format("%s frame types not supported", frame.getClass().getName())); ???????}// 返回应答消息 ???????String request = ((TextWebSocketFrame) frame).text(); ???????System.out.println("服务端2收到：" + request); ???????if (logger.isLoggable(Level.FINE)) { ???????????logger.fine(String.format("%s received %s", ctx.channel(), request)); ???????} ???????TextWebSocketFrame tws = new TextWebSocketFrame(new Date().toString() + ctx.channel().id() + "：" + request);// 群发 ???????Global.group.writeAndFlush(tws);// 返回【谁发的发给谁】// ctx.channel().writeAndFlush(tws); ???} ???private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {// 如果HTTP解码失败，返回HHTP异常 ???????if (!req.decoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))) { ???????????sendHttpResponse(ctx, req, ???????????????????new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST)); ???????????return; ???????}//获取url后置参数 ???????String uri = req.uri(); ???// 构造握手响应返回，本机测试 ???????WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory( ???????????????"ws://localhost:8081/websocket" , null, false); ???????handshaker = wsFactory.newHandshaker(req); ???????if (handshaker == null) { ???????????WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel()); ???????} else { ???????????handshaker.handshake(ctx.channel(), req); ???????} ???} ???private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse res) {// 返回应答给客户端 ???????if (res.status().code() != 200) { ???????????ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8); ???????????res.content().writeBytes(buf); ???????????buf.release(); ???????}// 如果是非Keep-Alive，关闭连接 ???????ChannelFuture f = ctx.channel().writeAndFlush(res); ???????if (!HttpUtil.isKeepAlive(req) || res.status().code() != 200) { ???????????f.addListener(ChannelFutureListener.CLOSE); ???????} ???} ???/** ????* exception 异常 Caught 抓住 抓住异常，当发生异常的时候，可以做一些相应的处理，比如打印日志、关闭链接 ????*/ ???@Override ???public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ???????cause.printStackTrace(); ???????ctx.close(); ???} ???@Override ???protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception { ???????// TODO Auto-generated method stub ???}}

package com.atguigu.netty.websocket;import io.netty.channel.group.ChannelGroup;import io.netty.channel.group.DefaultChannelGroup;import io.netty.util.concurrent.GlobalEventExecutor;/*** ClassName:Global* Function: TODO ADD FUNCTION.* @author fwz*/public class Global {public static ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);}

package com.atguigu.netty.websocket;import org.springframework.stereotype.Controller;import org.springframework.web.bind.annotation.RequestMapping;@Controllerpublic class DeptController_Consumer{ ???????????@RequestMapping(value = "/socketHtml") ???public String html() ???{ ???????return"WebSocketServer"; ???} ???}

<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:8081/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=""; ???????????????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("The socket is not open."); ???????????} ???????} ???</script> ???<form onsubmit="return false;"> ???????<input type="text" name="message" value="Hello,Netty-WebSocket" /> ???????<br> ???????<br> ????????<input ???????????type="button" value="发送 WebSocket 请求消息" ???????????onclick="send(this.form.message.value)" /> ???????????<hr color="blue"/> ???????????<h3>服务端返回的应答消息</h3> ???????????<textarea style="width: 500px;height: 300px;" id="responseText" ></textarea> ???</form></body></html>