/** * 该方法用于客户端,用来设置一个 EventLoop */ public ServerBootstrap group(EventLoopGroup group); /** * 该方法用于服务器端,用来设置两个 EventLoop */ public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup); /** * 用来给接收到的通道添加配置 */ public <T> ServerBootstrap childOption(ChannelOption<T> childOption, T value); /** * 用来给接收到的通道添加属性 */ public <T> ServerBootstrap childAttr(AttributeKey<T> childKey, T value); /** * 该方法用来设置业务处理类(自定义的 handler) */ public ServerBootstrap childHandler(ChannelHandler childHandler); /** * 该方法用于服务器端,用来设置占用的端口号 */ public ChannelFuture bind(int inetPort); /** * 该方法用于客户端,用来连接服务器 */ public ChannelFuture connect(String inetHost, int inetPort) ;
/** * 返回当前正在进行 IO 操作的通道 */ Channel channel(); /** * 给通道添加监听器 */ ChannelFuture addListener(GenericFutureListener<? extends Future<? super Void>> var1); /** * 给通道添加监听器 */ ChannelFuture addListeners(GenericFutureListener<? extends Future<? super Void>>... var1); /** * 给通道移除监听器 */ ChannelFuture removeListener(GenericFutureListener<? extends Future<? super Void>> var1); /** * 给通道移除监听器 */ ChannelFuture removeListeners(GenericFutureListener<? extends Future<? super Void>>... var1); /** * 等待任务结束,如果任务产生异常或被中断则抛出异常,否则返回Future自身 */ ChannelFuture sync() throws InterruptedException; /** * 等待任务结束,任务本身不可中断,如果产生异常则抛出异常,否则返回Future自身 */ ChannelFuture syncUninterruptibly(); /** * 等待任务结束,如果任务被中断则抛出中断异常,与sync不同的是只抛出中断异常,不抛出任务产生的异常 */ ChannelFuture await() throws InterruptedException; /** * 等待任务结束,任务不可中断 */ ChannelFuture awaitUninterruptibly();
适配器
Netty 提供一个专门用来操作缓冲区(即 Netty 的数据容器)的工具类,常用的一个方法如下
public static ByteBuf copiedBuffer(CharSequence string, Charset charset);
ByteBuf 原理图:
BEFORE clear() +-------------------+------------------+------------------+ | discardable bytes | readable bytes | writable bytes | +-------------------+------------------+------------------+ | | | | 0 <= readerIndex <= writerIndex <= capacity AFTER clear() +---------------------------------------------------------+ | writable bytes (got more space) | +---------------------------------------------------------+ | | 0 = readerIndex = writerIndex <= capacity
Bytebuf Demo 代码如下:
package bin.netty.bytebuf; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.util.CharsetUtil; /** * @author liyibin * @date 2021-06-26 */ public class ByteBufDemo { public static void main(String[] args) { ByteBuf byteBuf = Unpooled.copiedBuffer("hello, world", CharsetUtil.UTF_8); // buf index System.out.println("readerIndex: " + byteBuf.readerIndex() + ", writerIndex: " + byteBuf.writerIndex() + ", capacity: " + byteBuf.capacity()); for (int i = 0; i < byteBuf.writerIndex(); i++) { System.out.println((char) byteBuf.readByte()); } // buf index System.out.println("readerIndex: " + byteBuf.readerIndex() + ", writerIndex: " + byteBuf.writerIndex() + ", capacity: " + byteBuf.capacity()); // 读取部分 // param1: 起始索引 // param2: 长度 System.out.println(byteBuf.getCharSequence(0, 3, CharsetUtil.UTF_8)); } }
服务端:
package bin.netty.groupchat; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelPipeline; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.codec.string.StringDecoder; import io.netty.handler.codec.string.StringEncoder; import io.netty.handler.logging.LogLevel; import io.netty.handler.logging.LoggingHandler; /** * @author liyibin * @date 2021-06-26 */ public class NettyGroupChatServer { public static void main(String[] args) throws Exception { NioEventLoopGroup bossGroup = new NioEventLoopGroup(1); NioEventLoopGroup workerGroup = new NioEventLoopGroup(); try { ServerBootstrap serverBootstrap = new ServerBootstrap() .group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) // 给 bossGroup 添加一个日志处理器 .handler(new LoggingHandler(LogLevel.INFO)) .childHandler(new ChannelInitializer<SocketChannel>() { // 初始化通道,给 workerGroup 的通道添加处理器 @Override protected void initChannel(SocketChannel socketChannel) throws Exception { ChannelPipeline pipeline = socketChannel.pipeline(); // 字符类型解码器 pipeline.addLast("decoder", new StringDecoder()); // 字符类型编码器 pipeline.addLast("encoder", new StringEncoder()); // 业务处理器 pipeline.addLast(new GroupChatServerHandler()); } }); ChannelFuture channelFuture = serverBootstrap.bind(9999).sync(); channelFuture.addListener(cf -> { if (cf.isSuccess()) { System.out.println("listen on port 9999"); } }); channelFuture.channel().closeFuture().sync(); } finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } }
GroupChatServerHandler:
package bin.netty.groupchat; import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; import io.netty.channel.group.ChannelGroup; import io.netty.channel.group.DefaultChannelGroup; import io.netty.util.concurrent.GlobalEventExecutor; /** * @author liyibin * @date 2021-06-26 */ public class GroupChatServerHandler extends SimpleChannelInboundHandler<String> { /** * 通道组,用于管理当前连接的通道,全局唯一 */ private final static ChannelGroup CHANNELS = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); /** * 接收客户端发送的消息 */ @Override protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { Channel channel = ctx.channel(); // 服务端打印消息 System.out.printf("[%s]: %s\n", channel.remoteAddress().toString(), msg); // 给其他通道发送消息 CHANNELS.forEach(ch -> { // 不是当前通讯的通道,就发送消息 if (ch != channel) { ch.writeAndFlush(String.format("[%s]: %s\n", channel.remoteAddress().toString(), msg)); } }); } @Override public void handlerAdded(ChannelHandlerContext ctx) throws Exception { System.out.printf("%s 加入了群聊\n", ctx.channel().remoteAddress()); CHANNELS.add(ctx.channel()); System.out.println("当前群聊用户数:" + CHANNELS.size()); } /** * 上线 */ @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.printf("%s 上线了\n", ctx.channel().remoteAddress()); } @Override public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { System.out.printf("%s 退出了群聊", ctx.channel().remoteAddress()); // 会自动移除 System.out.println("当前群聊用户数:" + CHANNELS.size()); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ctx.close(); } }
客户端:
package bin.netty.groupchat; import io.netty.bootstrap.Bootstrap; import io.netty.buffer.Unpooled; import io.netty.channel.*; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.codec.string.StringDecoder; import io.netty.handler.codec.string.StringEncoder; import io.netty.util.CharsetUtil; import java.util.Scanner; /** * @author liyibin * @date 2021-06-26 */ public class NettyGroupChatClient { public static void main(String[] args) throws Exception { NioEventLoopGroup group = new NioEventLoopGroup(); try { Bootstrap bootstrap = new Bootstrap() .group(group) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel socketChannel) throws Exception { ChannelPipeline pipeline = socketChannel.pipeline(); // 字符类型解码器 pipeline.addLast("decoder", new StringDecoder()); // 字符类型编码器 pipeline.addLast("encoder", new StringEncoder()); pipeline.addLast(new SimpleChannelInboundHandler<String>() { @Override protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { // 打印接收到的消息 System.out.printf("%s\n", msg); } }); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ctx.close(); } }); ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 9999).sync(); channelFuture.addListener(cf -> { if (cf.isSuccess()) { System.out.println("connect to server"); } }); Channel channel = channelFuture.channel(); // 处理用户输入 Scanner scanner = new Scanner(System.in); while (scanner.hasNextLine()) { String msg = scanner.nextLine(); channel.writeAndFlush(Unpooled.copiedBuffer(msg + "\r\n", CharsetUtil.UTF_8)); } } finally { group.shutdownGracefully(); } } }
服务端:
package bin.netty.heartcheck; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelPipeline; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.logging.LogLevel; import io.netty.handler.logging.LoggingHandler; import io.netty.handler.timeout.IdleStateHandler; import java.util.concurrent.TimeUnit; /** * 心跳检测机制 * * @author liyibin * @date 2021-06-27 */ public class NettyHeartCheckServer { public static void main(String[] args) throws Exception { NioEventLoopGroup bossGroup = new NioEventLoopGroup(1); NioEventLoopGroup workerGroup = new NioEventLoopGroup(); try { ServerBootstrap serverBootstrap = new ServerBootstrap() .group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .handler(new LoggingHandler(LogLevel.INFO)) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel socketChannel) throws Exception { ChannelPipeline pipeline = socketChannel.pipeline(); // 加入处理空闲状态的处理器 // readerIdleTime: 表示多长时间没读,就发送一个心跳检测包 // writerIdleTime: 表示多长时间没写,就发送一个心跳检测包 // allIdleTime: 表示多长时间没读写,就发送一个心跳检测包 // Triggers an {@link IdleStateEvent} when a {@link Channel} has not performed read, write, or both operation for a while. // 当触发 IdleStateEvent 事件时,就会传递给下一个的 handler 的 useEventTriggered 方法处理 pipeline.addLast(new IdleStateHandler(3, 5, 7, TimeUnit.SECONDS)); // 业务处理器 pipeline.addLast(new MyServerHandler()); } }); ChannelFuture channelFuture = serverBootstrap.bind(9999).sync(); channelFuture.addListener(cf -> { if (cf.isSuccess()) { System.out.println("listen on 9999"); } }); channelFuture.channel().closeFuture().sync(); } finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } }
MyServerHandler:
package bin.netty.heartcheck; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.handler.timeout.IdleStateEvent; /** * @author liyibin * @date 2021-06-27 */ public class MyServerHandler extends ChannelInboundHandlerAdapter { @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { if (evt instanceof IdleStateEvent) { IdleStateEvent idleStateEvent = (IdleStateEvent) evt; String eventType; switch (idleStateEvent.state()) { case READER_IDLE: eventType = "读空闲"; break; case WRITER_IDLE: eventType = "写空闲"; break; case ALL_IDLE: eventType = "读写空闲"; break; default: eventType = null; break; } System.out.println(ctx.channel().remoteAddress() + "---空闲类型---" + eventType); } } }
客户端:
package bin.netty.heartcheck; import io.netty.bootstrap.Bootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; /** * @author liyibin * @date 2021-06-27 */ public class NettyHeartCheckClient { public static void main(String[] args) throws Exception { NioEventLoopGroup group = new NioEventLoopGroup(); try { Bootstrap bootstrap = new Bootstrap() .group(group) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { } }); ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 9999).sync(); channelFuture.channel().closeFuture().sync(); } finally { group.shutdownGracefully(); } } }
服务端:
package bin.netty.websocket; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelPipeline; 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.codec.http.websocketx.WebSocketServerProtocolHandler; import io.netty.handler.logging.LogLevel; import io.netty.handler.logging.LoggingHandler; import io.netty.handler.stream.ChunkedWriteHandler; /** * @author liyibin * @date 2021-06-27 */ public class NettyWebSocketServer { public static void main(String[] args) throws Exception { NioEventLoopGroup bossGroup = new NioEventLoopGroup(1); NioEventLoopGroup workerGroup = new NioEventLoopGroup(); try { ServerBootstrap serverBootstrap = new ServerBootstrap() .group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .handler(new LoggingHandler(LogLevel.INFO)) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); // http 编解码器 pipeline.addLast(new HttpServerCodec()); // 以块方式读写 pipeline.addLast(new ChunkedWriteHandler()); // http 数据传输是分段的,当发送大量数据时,需要将其聚合在一起 pipeline.addLast(new HttpObjectAggregator(8192)); // 1. websocket 处理器,数据以帧形式传输,netty 中对应 WebSocketFrame类,其下有 6 个子类。 // 2. 请求 ws://localhost:9999/chat 进行通讯 // 3. WebSocketServerProtocolHandler 核心是将 http 协议提升未 ws 协议,通过响应状态码 101 升级 pipeline.addLast(new WebSocketServerProtocolHandler("/chat")); // 业务处理器 pipeline.addLast(new MyTextWebSocketFrameHandler()); } }); ChannelFuture channelFuture = serverBootstrap.bind(9999).sync(); channelFuture.addListener(cf -> { if (cf.isSuccess()) { System.out.println("websocket listen on 9999"); } }); channelFuture.channel().closeFuture().sync(); } finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } }
MyTextWebSocketFrameHandler:
package bin.netty.websocket; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; import java.time.LocalDateTime; /** * @author liyibin * @date 2021-06-27 */ public class MyTextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> { @Override protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception { System.out.printf("[%s]: %s", ctx.channel().remoteAddress(), msg.text()); // 回复消息 ctx.writeAndFlush(new TextWebSocketFrame("服务器收到消息: " + LocalDateTime.now())); } @Override public void handlerAdded(ChannelHandlerContext ctx) throws Exception { System.out.println("handlerAdded 被" + ctx.channel().id().asShortText() + " 调用"); System.out.println("handlerAdded 被" + ctx.channel().id().asLongText() + " 调用"); } @Override public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { System.out.println("handlerRemoved 被" + ctx.channel().id().asShortText() + " 调用"); System.out.println("handlerRemoved 被" + ctx.channel().id().asLongText() + " 调用"); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ctx.close(); } }
客户端:
<!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <title>Title</title> </head> <body> <script> let socket; if (window.WebSocket) { socket = new WebSocket("ws://localhost:9999/chat"); socket.onopen = function (evt) { let res = document.getElementById("responseText"); res.value = "连接开启...\n"; } socket.onmessage = function (evt) { let res = document.getElementById("responseText"); res.value = res.value + evt.data; } socket.onclose = function (evt) { let res = document.getElementById("responseText"); res.value = res.value + "连接关闭...\n"; } } else { alert("当前浏览器不支持 WebSocket") } function send(msg) { console.log(msg); /*if (!window.socket) { console.log("no open"); return ; }*/ if (socket.readyState === WebSocket.OPEN) { socket.send(msg); } else { alert("连接没有开启") } } </script> <form onsubmit="return false"> <textarea name="message" style="width: 300px; height: 300px"></textarea> <input type="button" value="发送消息" onclick="send(this.form.message.value)"> <textarea id="responseText" style="width: 300px; height: 300px"></textarea> <input type="button" value="清空消息" onclick="document.getElementById('responseText').value=''"> </form> </body> </html>