public class MyServer { public static void main(String[] args) throws Exception{ // 负责链接的NioEventLoopGroup线程数为1 EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup wokerGroup = new NioEventLoopGroup(); try{ ServerBootstrap serverBootstrap = new ServerBootstrap(); serverBootstrap.group(bossGroup,wokerGroup).channel(NioServerSocketChannel.class) // 加入日志 .handler(new LoggingHandler(LogLevel.INFO)) // 自定义channel初始化器 .childHandler(new WebSocketChannelInitializer()); // 绑定本机的8005端口 ChannelFuture channelFuture = serverBootstrap.bind(new InetSocketAddress(8005)).sync(); // 异步回调-关闭事件 channelFuture.channel().closeFuture().sync(); }finally { bossGroup.shutdownGracefully(); wokerGroup.shutdownGracefully(); } }
channel初始化器javascript
public class WebSocketChannelInitializer extends ChannelInitializer<SocketChannel> { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); //websocket协议自己是基于http协议的,因此这边也要使用http解编码器 pipeline.addLast(new HttpServerCodec()); //以块的方式来写的处理器 pipeline.addLast(new ChunkedWriteHandler()); //netty是基于分段请求的,HttpObjectAggregator的做用是将请求分段再聚合,参数是聚合字节的最大长度 pipeline.addLast(new HttpObjectAggregator(8192)); // 使用websocket协议 //ws://server:port/context_path //参数指的是contex_path pipeline.addLast(new WebSocketServerProtocolHandler("/world")); //websocket定义了传递数据的中frame类型 pipeline.addLast(new TextWebSocketFrameHandler()); } }
WebSocketChannelInitializer
channel初始化器html
public class WebSocketChannelInitializer extends ChannelInitializer<SocketChannel> { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); //websocket协议自己是基于http协议的,因此这边也要使用http解编码器 pipeline.addLast(new HttpServerCodec()); //以块的方式来写的处理器 pipeline.addLast(new ChunkedWriteHandler()); //netty是基于分段请求的,HttpObjectAggregator的做用是将请求分段再聚合,参数是聚合字节的最大长度 pipeline.addLast(new HttpObjectAggregator(8192)); // 使用websocket协议 //ws://server:port/context_path //参数指的是contex_path pipeline.addLast(new WebSocketServerProtocolHandler("/world")); //websocket定义了传递数据的中frame类型,这里使用TextWebSocketFrame并自定义一个handler pipeline.addLast(new TextWebSocketFrameHandler()); } }
public class TextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> { /** * 通道列表 用于存放通道 */ public static CopyOnWriteArrayList<Channel> channelList = new CopyOnWriteArrayList<Channel>(); @Override protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception { /** * writeAndFlush接收的参数类型是Object类型,可是通常咱们都是要传入管道中传输数据的类型,好比咱们当前的demo * 传输的就是TextWebSocketFrame类型的数据 */ System.out.println(msg.text()); // 遍历通道list,向非当前通道发送消息 channelList.forEach(channel -> { if (channel != ctx.channel()){ channel.writeAndFlush(new TextWebSocketFrame(ctx.channel().id().asShortText()+":" +msg.text())); } else { channel.writeAndFlush(new TextWebSocketFrame("我:" +msg.text())); } } ); } /** * channel add后触发,向channelList添加新加入的通道 * * @param ctx ctx * @throws Exception 异常 */ @Override public void handlerAdded(ChannelHandlerContext ctx) throws Exception { channelList.forEach(channel -> channel.writeAndFlush(new TextWebSocketFrame(ctx.channel().id().asShortText()+"上线了"))); channelList.add(ctx.channel()); } /** * channel链接断开时触发,猜想是TCP断开时触发回调 发送链接断开事件 * * @param ctx ctx * @throws Exception 异常 */ @Override public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { channelList.forEach(channel -> channel.writeAndFlush(new TextWebSocketFrame(ctx.channel().id().asShortText()+"下线了"))); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { System.out.println("异常发生"); ctx.close(); }
TextWebSocketFrameHandler
继承了SimpleChannelInboundHandler
,SimpleChannelInboundHandler
其实是一个ChannelHandlerAdapter
,其方法会在入站的时候被调用。这里咱们经过重写handlerAdded
方法,在通道建立后将其加入当channelList中,并向其他通道发送成员上线的提示信息。重写channelRead0
方法,向全部非当前channel发送读取到消息。前端
前端
java
<!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <title>WebSocket客户端</title> </head> <body> <script type="text/javascript"> var socket; //若是浏览器支持WebSocket if(window.WebSocket){ //参数就是与服务器链接的地址 socket = new WebSocket("ws://localhost:8005/world"); //客户端收到服务器消息的时候就会执行这个回调方法 socket.onmessage = function (event) { var ta = document.getElementById("responseText"); ta.value = ta.value + "n"+event.data; } //链接创建的回调函数 socket.onopen = function(event){ var ta = document.getElementById("responseText"); ta.value = "链接开启"; } //链接断掉的回调函数 socket.onclose = function (event) { var ta = document.getElementById("responseText"); ta.value = ta.value +"n"+"链接关闭"; } }else{ alert("浏览器不支持WebSocket!"); } //发送数据 function send(message){ if(!window.WebSocket){ return; } //当websocket状态打开 if(socket.readyState == WebSocket.OPEN){ socket.send(message); }else{ alert("链接没有开启"); } } </script> <form onsubmit="return false"> <textarea name = "message" style="width: 400px;height: 200px"></textarea> <input type ="button" value="发送数据" onclick="send(this.form.message.value);"> <h3>服务器输出:</h3> <textarea id ="responseText" style="width: 400px;height: 300px;"></textarea> <input type="button" onclick="javascript:document.getElementById('responseText').value=''" value="清空数据"> </form> </body> </html>