netty的channelPipeline添加header处理器与eventLoop添加task任务过程

原创

spbreak

修改于 2023-09-25 02:59:24

28700

代码可运行

文章被收录于专栏：nettynetty

运行总次数：0

代码可运行

一. ChannelPipeline管道添加header处理器

channelPipeline管道是channel通道与socket套接字进行IO操作时, 对其数据作数据加工的流水线
一个pipeline包含一个hander Linked处理器链表, 以实现对数据的分步加工
hander Linked的基础单元为 hander处理器

// channelPipeline添加handler与删除handler实现
public static void main(String[] args) {
    ChannelFactory<NioServerSocketChannel> channelFactory = new ReflectiveChannelFactory<NioServerSocketChannel>(
            ObjectUtil.checkNotNull(NioServerSocketChannel.class, "channelClass")
    );

    Channel channel = channelFactory.newChannel();
    ChannelPipeline p = channel.pipeline();

    ChannelInitializer<Channel> handler = new ChannelInitializer<Channel>() {
        @Override
        public void initChannel(final Channel ch) {
            
        }
        @Override
        public String toString() {
            return "自定义hander处理器";
        }
    };
    
    p.addLast(handler);
    System.out.println("添加handler后第一个handler = " + p.first());
    p.remove(handler);
    System.out.println("移除Handler后第一个handler = " + p.first());
}

Connected to the target VM, address: '127.0.0.1:5103', transport: 'socket'
添加handler后第一个handler = 自定义hander处理器
移除Handler后第一个handler = null
Disconnected from the target VM, address: '127.0.0.1:5103', transport: 'socket'

Process finished with exit code 0

二. EventLoop事件循环器添加runnable可执行任务

eventLoop从本质上来讲是继承executor的执行器, 可接收继承runnable接口的异步编程任务
eventLoop的taskQueue任务队列属性是用于存储多个runnable可执行任务
每一个eventLoop都有一个所属于自己的运行线程, 保证每次都能在同一个线程中执行任务, 避免出现切换线程所带来的资源损耗
eventLoop#execute(runnable) 方法会先判断当前运行线程是否是eventLoop的所属线程
- 如果不是, 新创建一个线程startThread()来run()跑taskQueue里的任务
eventLoop的run()方法会遍历taskQueue任务队列中的所有task, 进行执行, 直到taskQueue为空

// EventLoop添加和删除runnable可执行任务
public static void main(String[] args) throws InterruptedException {
    EventLoopGroup bossGroup = new NioEventLoopGroup(1);

    EventLoop eventLoop = bossGroup.next();
    Runnable runnable = new Runnable() {
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName() + " " + Thread.currentThread().getStackTrace()[1].getMethodName());
        }
    };
    eventLoop.execute(runnable); // eventLoop.taskQueue.size = 1
    Thread.sleep(2*1000);
    // eventLoop.taskQueue.size = 0. 被执行的任务自动移除
    System.out.println(Thread.currentThread().getName() + " " + Thread.currentThread().getStackTrace()[1].getMethodName());
}