基于netty的websocket例子 -云博客

基于netty的websocket例子

# 前端 2024-05-05 02:19 0 43 来源：云博客

nettyServer

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

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 }}

Global

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

controller

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"; } }

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>