Netty-ChannelPipeline

EventLoop可以说是 Netty 的调度中心,负责监听多种事件类型:I/O 事件、信号事件、定时事件等,然而实际的业务处理逻辑则是由 ChannelPipeline 中所定义的 ChannelHandler 完成的,ChannelPipeline 和 ChannelHandler应用开发的过程中打交道最多的组件,为用户提供了 I/O 事件的全部控制权。

文章目录

  • 一、ChannelPipeline 是什么?🤔️
  • 二、ChannelPipeline 的内部结构🔍
    • 1、HeadContext
    • 2、TailContext
    • 3、addLiast() 方法🔍
  • 三、ChannelPipeline 事件传播机制
  • 四、ChannelPipeline 异常传播机制
  • 五、统一的异常处理器

一、ChannelPipeline 是什么?🤔️

pipeline 有管道,流水线的意思,最早使用在 Unix 操作系统中,可以让不同功能的程序相互通讯,使软件更加”高内聚,低耦合”,它以一种”链式模型”来串起不同的程序或组件,使它们组成一条直线的工作流。

ChannelPipeline 也是 Netty 中的一个比较重要的组件,从上面的 Channel 实例化过程可以看出,每一个 Channel 实例中都会包含一个对应的 ChannelPipeline 属性。ChannelPipeline维护着处理或拦截channel的进站事件和出站事件的双向链表,事件在ChannelPipeline中流动和传递,可以增加或删除ChannelHandler来实现对不同业务逻辑的处理。通俗的说,ChannelPipeline是工厂里的流水线,ChannelHandler是流水线上的工人。

二、ChannelPipeline 的内部结构🔍

final AbstractChannelHandlerContext head;
final AbstractChannelHandlerContext tail;

private final Channel channel;
private final ChannelFuture succeededFuture;
private final VoidChannelPromise voidPromise;

protected DefaultChannelPipeline(Channel channel) {
    this.channel = ObjectUtil.checkNotNull(channel, "channel");
    succeededFuture = new SucceededChannelFuture(channel, null);
    voidPromise =  new VoidChannelPromise(channel, true);

    tail = new TailContext(this);
    head = new HeadContext(this);

    head.next = tail;
    tail.prev = head;
}

从 ChannelPipeline 的构造函数可以看出,ChannelPipeline 维护了一组 ChannelHandlerContext 实例组成双向链表。默认会包含 head 和 tail 头尾节点,用来进行一些默认的逻辑处理。我们自定义的ChannelHandler会插入到 head 和 tail 之间,这两个节点在 Netty 中已经默认实现了,它们在ChannelPipeline 中起到了至关重要的作用。

那么你可能会有疑问,为什么这里会多一层 ChannelHandlerContext 的封装呢?

其实这是一种比较常用的编程思想。ChannelHandlerContext用于保存ChannelHandler。ChannelHandlerContext包含了ChannelHandler生命周期的所有事件,如 connect、bind、read、 flush、write、close 等。

可以试想一下,如果没有ChannelHandlerContext 的这层封装,那么我们在做 ChannelHandler 之间传递的时候。前置后置的通用逻辑就要在每个 ChannelHandler 里都实现一份。

首先我们看下 HeadContext 和 TailContext 的继承关系
在这里插入图片描述

1、HeadContext

通过集成关系我们发现 HeadContext 分别实现了ChannelInboundHandler 和 ChannelOutboundHandler,即 HeadContext 既是 入站处理器,也是出站处理器。

HeadContext是入站第一站出站最后一站。对于1个请求先由HeadContext处理入栈,经过一系列的入栈处理器然后传递到TailContext,由TailContext往下传递经过一系列的出栈处理器,最后再经过HeadContext返回。

2、TailContext

TailContext 只实现了 ChannelInboundHandler 接口。它会在 ChannelInboundHandler 调用链路的最后一步执行,主要用于终止 入站事件传播,例如释放 Message 数据资源等。

TailContext是入站最后一站出站第一站。TailContext节点作为出站事件传播的第一站,仅仅是将出站事件传递给下一个节点。

从整个 ChannelPipeline 调用链路来看,如果由 Channel 直接触发事件传播,那么调用链路将贯穿整个 ChannelPipeline。然而也可以在其中某一个 ChannelHandlerContext 触发同样的方法,这样只会从当前的 ChannelHandler 开始执行事件传播,该过程不会从头贯穿到尾,在一定场景下,可以提高程序性能。

3、addLiast() 方法🔍

addLast() 方法是向 ChannelPipeline 中添加 ChannelHandler 用来进行业务处理,关于ChannelHandler将会在下文中详细讲解!
在这里插入图片描述

三、ChannelPipeline 事件传播机制

入站事件是由I/O线程被动触发,由入站处理器按自下而上的方向处理,在中途可以被拦截丢弃,出站事件由用户handler主动触发,由出站处理器按自上而下的方向处理。
在这里插入图片描述
接下来用一个示例来讲解~

服务端代码,

public class PipelineServer {
    public static void main(String[] args) throws InterruptedException {
        NioEventLoopGroup boss = new NioEventLoopGroup(1);
        NioEventLoopGroup worker = new NioEventLoopGroup(2);
        new ServerBootstrap()
                .group(boss, worker)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                System.out.println(1);
                                super.channelRead(ctx, msg);
                            }
                        });
                        ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                System.out.println(2);
                                super.channelRead(ctx, msg);
                            }
                        });
                        ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                System.out.println(3);
                                super.channelRead(ctx, msg);
                            }
                        });
                        ch.pipeline().addLast(new ChannelOutboundHandlerAdapter() {
                            @Override
                            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                                System.out.println(4);
                                super.write(ctx, msg, promise);
                            }
                        });
                        ch.pipeline().addLast(new ChannelOutboundHandlerAdapter() {
                            @Override
                            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                                System.out.println(5);
                                super.write(ctx, msg, promise);
                            }
                        });
                        ch.pipeline().addLast(new ChannelOutboundHandlerAdapter() {
                            @Override
                            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                                System.out.println(6);
                                super.write(ctx, msg, promise);
                            }
                        });
                    }
                })
                .bind(8080);
    }
}

客户端代码,

public class PipelineClient {
    public static void main(String[] args) throws InterruptedException {
        new Bootstrap()
                .group( new NioEventLoopGroup())
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel ch) throws Exception {
                        ch.pipeline().addLast(new StringDecoder());
                        ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                System.out.println(msg);
                                super.channelRead(ctx, msg);
                            }
                        });
                        ch.pipeline().addLast(new StringEncoder());
                    }
                }).connect("127.0.0.1", 8080)
                .sync()
                .channel()
                .writeAndFlush("Hello,server!");
    }
}

依次启动服务端和客户端,服务端打印如下:

1
2
3

以上我们通过 Pipeline 的 addLast 方法分别添加了三个 ChannelInboundHandlerAdapter 和 ChannelOutboundHandlerAdapter,添加顺序分别是 1 -> 2 -> 3,4 -> 5 -> 6。

此时为什么没有打印 4、5、6呢,即没有触发出站的操作❓

出站处理器只有向channel中写入数据才会触发,我们在第三个 ChannelInboundHandlerAdapter 实现类中加入以下代码!

在这里插入图片描述
通过依次点入,我们发现最终是调用了 tail节点 的writeAndFlush 方法,即TailContext节点作为出站事件传播的第一站!
在这里插入图片描述

最终服务端打印如下:

1
2
3
6
5
4

可以看到,ChannelInboundHandlerAdapter 是按照 addLast 的顺序执行的,而 ChannelOutboundHandlerAdapter 是按照 addLast 的逆序执行的。ChannelPipeline 的实现是一个 ChannelHandlerContext(包装了 ChannelHandler) 组成的双向链表

在这里插入图片描述

  • 入站处理器中,ctx.fireChannelRead(msg) 是 调用下一个入站处理器
    • 如果注释掉 1 处代码,则仅会打印 1
    • 如果注释掉 2 处代码,则仅会打印 1 2
  • 3 处的 ctx.channel().write(msg) 会 从尾部开始触发 后续出站处理器的执行
    • 如果注释掉 3 处代码,则仅会打印 1 2 3
  • 类似的,出站处理器中,ctx.write(msg, promise) 的调用也会 触发上一个出站处理器
    • 如果注释掉 6 处代码,则仅会打印 1 2 3 6
  • ctx.channel().write(msg) vs ctx.write(msg)
    • 都是触发出站处理器的执行
    • ctx.channel().write(msg) 从尾部开始查找出站处理器
    • ctx.write(msg) 是从当前节点找上一个出站处理器
    • 3 处的 ctx.channel().write(msg) 如果改为 ctx.write(msg) 仅会打印 1 2 3,因为节点3 之前没有其它出站处理器了
    • 6 处的 ctx.write(msg, promise) 如果改为 ctx.channel().write(msg) 会打印 1 2 3 6 6 6… 因为 ctx.channel().write() 是从尾部开始查找,结果又是节点6 自己

如图,服务端 pipeline 触发的原始流程,图中数字代表了处理步骤的先后次序

在这里插入图片描述

四、ChannelPipeline 异常传播机制

ChannelPipeline 事件传播的实现采用了经典的责任链模式,调用链路环环相扣。那么如果有一个节点处理逻辑异常会出现什么现象呢?我们通过修改 第二个 ChannelInboundHandlerAdapter 实现类 的实现来模拟业务逻辑异常:
在这里插入图片描述
由输出结果可以看出 ctx.fireExceptionCaugh 会将异常按顺序从 Head 节点传播到 Tail 节点
在这里插入图片描述
如果用户没有对异常进行拦截处理,最后将由 Tail 节点统一处理,在 TailContext 源码中可以找到具体实现:
在这里插入图片描述

五、统一的异常处理器

在 Netty 应用开发的过程中,良好的异常处理机制会让开发在排查问题的时候事半功倍。虽然 Netty 中 TailContext 提供了兜底的异常处理逻辑,但是在很多场景下,并不能满足我们的需求。假如你需要拦截指定的异常类型,并做出相应的异常处理,应该如何实现呢?

小编个人推荐用户对异常进行统一拦截,然后根据实际业务场景实现更加完善的异常处理机制。

通过异常传播机制的学习,我们应该可以想到最好的方法是在 ChannelPipeline 自定义处理器的末端添加统一的异常处理器!

/**
 * 自定义异常处理器
 */
public static class ExceptionHandler extends ChannelDuplexHandler {
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        if (cause instanceof RuntimeException) {
            System.out.println();
            log.error("业务异常处理,异常信息:{}", cause.getMessage());
        }
    }
}

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:/a/100959.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

[dasctf]misc05

盲水印 png里藏jpg&#xff0c;bwm.py可以提取含flag的图片

git学习笔记 | 版本管理 - 分支管理

文章目录 git学习笔记Git是什么仓库常见的命令commit 备注规范在文件下设置git忽略文件 .gitignore 版本管理git log | git reflog 查看提交日志/历史版本版本穿梭 git resetgit reset HEAD <file> git checkout -- fileName 丢弃工作区的修改git revertGit恢复之前版本的…

论文阅读_扩散模型_SDXL

英文名称: SDXL: Improving Latent Diffusion Models for High-Resolution Image Synthesis 中文名称: SDXL&#xff1a;改进潜在扩散模型的高分辨率图像合成 论文地址: http://arxiv.org/abs/2307.01952 代码: https://github.com/Stability-AI/generative-models 时间: 2023-…

3D视觉测量:形位公差 面对面平行度(附源码)

文章目录 0. 测试效果1. 基本内容1.1 面对面平行度概述2. 代码实现文章目录:3D视觉测量目录微信:dhlddxB站: Non-Stop_0. 测试效果 1. 基本内容 “平行度” 是用来描述两个或多个对象或方向之间的平行关系的度量。在几何和工程学中,平行度通常用于衡量物体、表面、线条或方向…

mac帧 arp

1.分片 2.MSS max segment size 3.跨网络的本质 就是经历很多的子网或者局域网 4.将数据从A主机跨网络送到B主机的能力 IP和mac IP解决的是路径选择的问题 5.数据链路层 用于两个设备&#xff08;同一种数据链路节点&#xff09;之间进行传递 6.以太网ether 7.局域网通…

圆圈加数字的css

方式一 .circle { width: 50px; height: 50px; border-radius: 50%; background-color: #f00; color: #fff; text-align: center; line-height: 50px; } .circle::before { content: attr(data-number); display: block; } <div class"circle" data-number"…

ChatGPT总结(持续更新)

目录 体验渠道 weTab CSDN-AI助手 其他插件 ChatGPT简介 ChatGPT主要用途 ChatGPT发展历程 GPT-4架构的特点和优势 ChatGPT的工作原理 神经网络和自然语言处理技术 Transformer模型 模型训练优化技巧 ChatGPT对程序员的帮助 与ChatGPT交互和提问技巧 ChatGPT未来…

50ETF期权开户平台(0门槛期权开户指南)

50ETF期权开户平台比较好的有&#xff1a;期权馆&#xff0c;期权科普馆&#xff0c;小熊期权&#xff0c;期权酱&#xff0c;财顺财经&#xff0c;财顺期权等&#xff0c;都是国内前十的期权分仓平台&#xff0c;下文为大家结算50ETF期权开户平台&#xff08;0门槛期权开户指南…

SoC 总线结构学习记录之系统存储总线(System Memory Bus)与私有设备总线

蜂鸟 E203 SOC总线结构&#xff1a;  蜂鸟 E203 内核 BIU 的系统存储接口 ICB 连接系统存储总线&#xff0c;通过其访问 SoC 中的若干存储组件&#xff0c;譬如 ROM&#xff0c;Flash 的只读区间等。  蜂鸟 E203 内核 BIU 的私有设备接口 ICB 连接私有设备总线&#xff0c…

mac电脑屏幕录制Berrycast Mac屏幕录制软件

Berrycast是一款为Mac设计的优秀屏幕录制软件&#xff0c;它让屏幕录制变得简单而高效。以下是Berrycast的一些主要特点&#xff1a; 简单的用户界面&#xff1a;Berrycast拥有直观和简洁的用户界面&#xff0c;使得用户可以轻松上手。高质量的视频输出&#xff1a;Berrycast能…

Sharding-JDBC介绍及分库分表实践

1.1 ShardingSphere简介 最早是当当网内部使用的一款分库分表框架&#xff0c;名字叫Sharding-JDBC&#xff0c;定位为轻量级 Java 框架&#xff0c;在 Java 的 JDBC 层提供的额外服务。 它使用客户端直连数据库&#xff0c;以 jar 包形式提供服务&#xff0c;无需额外部署和依…

机器人中的数值优化(六)—— 线搜索最速下降法

本系列文章主要是我在学习《数值优化》过程中的一些笔记和相关思考&#xff0c;主要的学习资料是深蓝学院的课程《机器人中的数值优化》和高立编著的《数值最优化方法》等&#xff0c;本系列文章篇数较多&#xff0c;不定期更新&#xff0c;上半部分介绍无约束优化&#xff0c;…

C语言每日一练--Day(16)

本专栏为c语言练习专栏&#xff0c;适合刚刚学完c语言的初学者。本专栏每天会不定时更新&#xff0c;通过每天练习&#xff0c;进一步对c语言的重难点知识进行更深入的学习。 今日练习题关键字&#xff1a;寻找奇数 峰值 二分查找 &#x1f493;博主csdn个人主页&#xff1a;小…

《数字图像处理-OpenCV/Python》连载(4)图像的读取与保存

《数字图像处理-OpenCV/Python》连载&#xff08;4&#xff09;图像的读取与保存 本书京东优惠购书链接&#xff1a;https://item.jd.com/14098452.html 本书CSDN独家连载专栏&#xff1a;https://blog.csdn.net/youcans/category_12418787.html 第1章 图像的基本操作 为了方…

【MyBatisⅡ】动态 SQL

目录 &#x1f392;1 if 标签 &#x1fad6;2 trim 标签 &#x1f460;3 where 标签 &#x1f9ba;4 set 标签 &#x1f3a8;5 foreach 标签 动态 sql 是Mybatis的强⼤特性之⼀&#xff0c;能够完成不同条件下不同的 sql 拼接。 在 xml 里面写判断条件。 动态SQL 在数据库里…

vr智慧党建主题展厅赋予企业数字化内涵

现如今&#xff0c;VR全景技术的发展让我们动动手指就能在线上参观博物馆、纪念馆&#xff0c;不仅不用受时间和空间的限制&#xff0c;还能拥有身临其境般的体验&#xff0c;使得我们足不出户就能随时随地学习、传承红色文化。 很多党建展厅都是比较传统的&#xff0c;没有运用…

多级缓存 架构设计

说在前面 在40岁老架构师 尼恩的读者社区(50)中&#xff0c;很多小伙伴拿到一线互联网企业如阿里、网易、有赞、希音、百度、网易、滴滴的面试资格&#xff0c;多次遇到一个很重要的面试题&#xff1a; 20w的QPS的场景下&#xff0c;服务端架构应如何设计&#xff1f;10w的QPS…

国际版阿里云/腾讯云:弹性高性能计算E-HPC入门概述

入门概述 本文介绍E-HPC的运用流程&#xff0c;帮助您快速上手运用弹性高性能核算。 下文以创立集群&#xff0c;在集群中安装GROMACS软件并运转水分子算例进行高性能核算为例&#xff0c;介绍弹性高性能核算的运用流程&#xff0c;帮助您快速上手运用弹性高性能核算。运用流程…

企业架构LNMP学习笔记7

PHP介绍&#xff1a; HTML&#xff1a;超文本标记语言 http: 超文本传输协议 端口80 浏览器将html代码解析成web页面。 PHP&#xff1a;超文本预处理器。后端语言开发&#xff0c;页面上需要动态改变修改的&#xff0c;需要连接数据库查询数据&#xff0c;转为html。 主要…

电脑每次开机杀毒软件报iusb3mon.exe病毒已清除,电脑中病毒iusbmon杀毒办法,工具杀毒

不知道什么时候开始&#xff0c;我电脑C盘的系统数据存储文件夹programdata 不知不觉就没了&#xff0c;找不到了 programdata文件夹为存储系统数据文件的&#xff0c;这个文件不见了&#xff0c;而且我打开了显示隐藏文件和文件夹还是没有显示 然后我重启电脑&#xff0c;杀毒…