Netty笔记03-组件Channel
文章目录
- Channel概述
- Channel 的概念
- Channel 的主要功能
- Channel 的生命周期
- Channel 的状态
- Channel 的类型
- channel 的主要方法
- ChannelFuture
- CloseFuture
- 💡 netty异步提升的是什么
- 要点
- 总结
Channel概述
Channel 的概念
在 Netty 中，Channel 是一个非常重要的概念，它代表了网络连接的抽象(一个网络连接)，用于进行数据的读取和写入操作。Channel 是 Netty 中处理网络 I/O 的核心组件之一，它将网络 I/O 操作封装在一个统一的接口下，使得开发人员可以更容易地编写高性能的网络应用程序。
Channel 是 Netty 中的接口，它定义了一组基本的操作，如打开、关闭、读取、写入等。Netty 为不同的网络通信协议提供了不同的Channel 实现，比如 TCP 的 SocketChannel，UDP 的 DatagramChannel 等。
Channel 的主要功能
- 读取数据：从网络中读取数据。
- 写入数据：将数据写入到网络中。
- 关闭连接：关闭当前的 Channel。
- 绑定地址：将 Channel 绑定到特定的本地地址（IP 地址和端口号）。
注册到 EventLoop：将 Channel 注册到 EventLoop，以便处理 I/O 事件。获取ChannelPipeline：每个 Channel 都有一个 ChannelPipeline，用于处理入站和出站的数据流。
Channel 的生命周期
一个 Channel 的生命周期通常包括以下几个阶段：
- 创建：当一个新连接建立时，Netty 会创建一个新的 Channel。
- 注册：新创建的 Channel 会注册到一个 EventLoop 上。
- 激活：当 Channel 被绑定到一个 Socket 地址时，它会变为激活状态。
- 读写操作：Channel 可以进行读取和写入操作。
- 关闭：当连接关闭时，Channel 也会被关闭。
Channel 的状态
Channel 有几个重要的状态，包括但不限于：
- 注册状态：Channel 是否已经被注册到 EventLoop。
channel.isRegistered() // 检查是否已注册
- 活跃状态：Channel 是否已经绑定到一个 Socket 地址。
channel.isActive() // 检查是否活跃
- 打开状态：Channel 是否处于打开状态。
channel.isOpen() // 检查是否打开
Channel 的类型
Netty 为不同的网络通信协议提供了不同的 Channel 实现：
- TCP 通信：NioServerSocketChannel 和 NioSocketChannel。
- UDP 通信：NioDatagramChannel。
- 文件传输：FileRegion。
channel 的主要方法
- close() 可以用来关闭 channel
- closeFuture() 用来处理 channel 的关闭
- sync 方法作用是同步等待 channel 关闭
- 而 addListener 方法是异步等待 channel 关闭
- pipeline() 方法添加处理器
- write() 方法将数据写入(只会将数据写入到channel的缓冲区中，具体什么时候发送数据则有很多条件，如在执行flush()或缓冲区达到一定大小)
- writeAndFlush() 方法将数据写入并刷出(将数据写入到channel的缓冲区中并立刻从缓冲区中发出)
ChannelFuture
使用上一章的服务器端和客户端代码
服务器端
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import jdk.internal.org.objectweb.asm.Handle;
import lombok.extern.slf4j.Slf4j;
import java.nio.charset.Charset;@Slf4j
public class EventLoopServer {public static void main(String[] args) {// 创建一个独立的 EventLoopGroup，将耗时的代码放到一个额外的组中线程处理EventLoopGroup group = new DefaultEventLoopGroup();new ServerBootstrap()// boss 和 worker// boss 只负责 ServerSocketChannel 上 accept 事件// worker 只负责 socketChannel 上的读写//group参数1：boss不需要指定线程数，因为ServerSocketChannel只会跟一个EventLoop进行绑定，// 又因为服务器只有一个，所以只会占用一个线程，不用指定线程数。//group参数1：work线程指定为两个.group(new NioEventLoopGroup(), new NioEventLoopGroup(2)).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {//如果读操作耗费的时间很上，会影响其他客户端的读写操作，// 一个work管理多个channel，如果其中一个耗时过长则会影响其他channel的读写操作ch.pipeline().addLast("handler1", new ChannelInboundHandlerAdapter() {/*** @param ctx* @param msg ByteBuf类型* @throws Exception*/@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buf = (ByteBuf) msg;log.debug(buf.toString(Charset.defaultCharset()));ctx.fireChannelRead(msg); // 让消息传递给下一个handler,如果不添加则消息不会传递给handler2中}}).addLast(group, "handler2", new ChannelInboundHandlerAdapter() {@Override // ByteBufpublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buf = (ByteBuf) msg;log.debug(buf.toString(Charset.defaultCharset()));}});}}).bind(8080);}
}
客户端代码
new Bootstrap().group(new NioEventLoopGroup()).channel(NioSocketChannel.class).handler(new ChannelInitializer<Channel>() {@Overrideprotected void initChannel(Channel ch) {ch.pipeline().addLast(new StringEncoder());}}).connect("127.0.0.1", 8080).sync().channel().writeAndFlush(new Date() + ": hello world!");
拆分上面的代码
ChannelFuture channelFuture = new Bootstrap().group(new NioEventLoopGroup()).channel(NioSocketChannel.class).handler(new ChannelInitializer<Channel>() {@Overrideprotected void initChannel(Channel ch) {ch.pipeline().addLast(new StringEncoder());}}).connect("127.0.0.1", 8080); // 标记1channelFuture.sync()
.channel()
.writeAndFlush(new Date() + ": hello world!");
- 1 处返回的是 ChannelFuture 对象，它的作用是利用 channel() 方法来获取 Channel 对象
注意 connect 方法是异步的，意味着不等连接建立，方法执行就返回了。因此 channelFuture 对象中不能【立刻】获得到正确的 Channel 对象。
实验如下：
// 2. 类中带有 Future，Promise 的类型都是和异步方法配套使用，用来处理结果
ChannelFuture channelFuture = new Bootstrap().group(new NioEventLoopGroup()).channel(NioSocketChannel.class).handler(new ChannelInitializer<NioSocketChannel>() {@Override // 在连接建立后被调用protected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new StringEncoder());}})// 1. 连接到服务器// 异步非阻塞, main 发起了调用，真正执行 connect 是 nio 线程.connect(new InetSocketAddress("localhost", 8080)); // 1s 秒后// 2.1 使用 sync 方法同步处理结果
//1.
System.out.println("sync前："+channelFuture.channel());//[id: 0xe3489549]
//2.
channelFuture.sync(); // 阻塞住当前线程，直到nio线程连接建立完毕
//3.
System.out.println("sync后："+channelFuture.channel());//[id: 0xe3489549, L:/127.0.0.1:54398 - R:localhost/127.0.0.1:8080]
Channel channel = channelFuture.channel();
log.debug("{}", channel);
//[DEBUG] [main] c.i.n.c.EventLoopClient - [id: 0xf540a105, L:/127.0.0.1:20434 - R:localhost/127.0.0.1:8080]
channel.writeAndFlush("hello, world");
channel.writeAndFlush("hello, world");
channel.writeAndFlush("hello, world");
System.out.println();
- 执行到 1 时，连接未建立，打印
[id: 0xe3489549] - 执行到 2 时，sync 方法是同步等待连接建立完成
- 执行到 3 时，连接肯定建立了，打印
[id: 0xe3489549, L:/127.0.0.1:54398 - R:localhost/127.0.0.1:8080]
除了用 sync 方法可以让异步操作同步以外，还可以使用回调的方式：
// 2. 类中带有 Future，Promise 的类型都是和异步方法配套使用，用来处理结果ChannelFuture channelFuture = new Bootstrap().group(new NioEventLoopGroup()).channel(NioSocketChannel.class).handler(new ChannelInitializer<NioSocketChannel>() {@Override // 在连接建立后被调用protected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new StringEncoder());}})// 1. 连接到服务器// 异步非阻塞, main 发起了调用，真正执行 connect 是 nio 线程.connect(new InetSocketAddress("localhost", 8080)); // 1s 秒后// 2.2 使用 addListener(回调对象) 方法异步(将main线程处理的活交给其他线程)处理结果// 将等待连接建立、连接成功后处理结果全部交给其他的线程处理//1.System.out.println("sync前："+channelFuture.channel());//[id: 0x3baac75f]channelFuture.addListener(new ChannelFutureListener() {
/** ↑* ↑ ← ← ↑* ↑* @param future 该future对象就是channelFuture对象* @throws Exception*/@Override// 在 nio 线程连接建立好之后，会调用 operationCompletepublic void operationComplete(ChannelFuture future) throws Exception {//2.System.out.println("sync后："+channelFuture.channel());//[id: 0x3baac75f, L:/127.0.0.1:54744 - R:localhost/127.0.0.1:8080]Channel channel = future.channel();log.debug("{}", channel);//[DEBUG] [nioEventLoopGroup-2-1] c.i.n.c.EventLoopClient - [id: 0x2ac4448f, L:/127.0.0.1:20346 - R:localhost/127.0.0.1:8080]channel.writeAndFlush("hello, world");//调用以上三行代码的还是nio线程}});
- 执行到 1 时，连接未建立，打印
[id: 0x3baac75f] - ChannelFutureListener 会在连接建立时被调用（其中 operationComplete 方法），因此执行到 2 时，连接肯定建立了，打印
[id: 0x3baac75f, L:/127.0.0.1:54744 - R:localhost/127.0.0.1:8080]
CloseFuture
需求：客户端的控制台不断的接受用户的输入，将用户输入的信息发给服务器端， 当不想发送信息时，输入q退出。
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;
import java.net.InetSocketAddress;
import java.util.Scanner;@Slf4j
public class CloseFutureClient {public static void main(String[] args) throws InterruptedException {NioEventLoopGroup group = new NioEventLoopGroup();ChannelFuture channelFuture = new Bootstrap().group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<NioSocketChannel>() {@Override // 在连接建立后被调用protected void initChannel(NioSocketChannel ch) throws Exception {//LoggingHandler一般用于调试使用，会将channel运行流程、状态显示出来ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));ch.pipeline().addLast(new StringEncoder());}}).connect(new InetSocketAddress("localhost", 8080));System.out.println(channelFuture.getClass());Channel channel = channelFuture.sync().channel();log.debug("{}", channel);new Thread(()->{Scanner scanner = new Scanner(System.in);//不断拿到用户输入的信息while (true) {String line = scanner.nextLine();if ("q".equals(line)) {channel.close(); // 调用close()方法是交给其他线程执行异步操作
// log.debug("处理关闭之后的操作"); // 不能在这里善后，因为是交给其他线程执行可能出现1s之后执行close()的情况break;}//输入的不是q则发送给服务器channel.writeAndFlush(line);}}, "input").start();// 获取 CloseFuture 对象， 1) 同步处理关闭， 2) 异步处理关闭ChannelFuture closeFuture = channel.closeFuture();//1) 同步处理关闭
// log.debug("waiting close...");
// closeFuture.sync();//只有在调用了channel.close()后才会继续向下运行
// log.debug("处理关闭之后的操作");//2) 异步处理关闭System.out.println("closeFuture.getClass():"+closeFuture.getClass());// channelFuture.addListener(new ChannelFutureListener() {
// //关闭channel的线程调用operationComplete方法
// //也就是nio线程执行完close(),找到该回调对象调用该方法
// @Override
// public void operationComplete(ChannelFuture future) throws Exception {
// log.debug("处理关闭之后的操作");
// }
// });//以上代码会出现用户输入q时，程序没有结束的情况。// 原因是NioEventLoopGroup中还有一些线程在运行，需要进行关闭channelFuture.addListener(new ChannelFutureListener() {@Overridepublic void operationComplete(ChannelFuture future) throws Exception {log.debug("处理关闭之后的操作");group.shutdownGracefully();}});//或使用以下代码替换
// closeFuture.addListener((ChannelFutureListener) future -> {
// log.debug("处理关闭之后的操作");
// group.shutdownGracefully();
// });}
}
💡 netty异步提升的是什么
- 有些同学看到这里会有疑问：为什么不在一个线程中去执行建立连接、去执行关闭 channel，那样不是也可以吗？非要用这么复杂的异步方式：比如一个线程发起建立连接，另一个线程去真正建立连接。
- 还有同学会笼统地回答，因为 netty 异步方式用了多线程、多线程就效率高。其实这些认识都比较片面，多线程和异步所提升的效率并不是所认为的。(其实提高的是单位时间内处理请求的吞吐量)
思考：
思考下面的场景，4 个医生给人看病，每个病人花费 20 分钟，而且医生看病的过程中是以病人为单位的，一个病人看完了，才能看下一个病人。假设病人源源不断地来，可以计算一下 4 个医生一天工作 8 小时，处理的病人总数是：4 * 8 * 3 = 96
经研究发现，看病可以细分为四个步骤，经拆分后每个步骤需要 5 分钟，如下
因此可以做如下优化，只有一开始，医生 2、3、4 分别要等待 5、10、15 分钟才能执行工作，但只要后续病人源源不断地来，他们就能够满负荷工作，并且处理病人的能力提高到了 4 * 8 * 12 效率几乎是原来的四倍。
要点
- 单线程没法异步提高效率，必须配合多线程、多核 cpu 才能发挥异步的优势
- 异步并没有缩短响应时间，反而有所增加
- 合理进行任务拆分，也是利用异步的关键
总结
异步并没有缩短处理单位的时间，反而有所增加请求的处理和响应时间。
关键点是提高了单位时间内处理请求的(个数)吞吐量，单位时间内能过处理请求的速度。
全部文章：
Netty笔记01-Netty的基本概念与用法
Netty笔记02-组件EventLoop
Netty笔记03-组件Channel
Netty笔记04-组件Future & Promise
Netty笔记05-组件Handler & Pipeline