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 效率几乎是原来的四倍。
在这里插入图片描述