与建设通相关的网站,网站排名诊断,分销商城,学校网站的系统建设方式文章目录 前言一、ChannelPipeline 接口1.1 创建 ChannelPipeline1.2 ChannelPipeline 事件传输机制1.2.1 处理出站事件1.2.2 处理入站事件 二、ChannelPipeline 中的 ChannelHandler三、ChannelHandlerContext 接口3.1 ChannelHandlerContext 与其他组件的关系3.2 跳过某些 Ch… 文章目录 前言一、ChannelPipeline 接口1.1 创建 ChannelPipeline1.2 ChannelPipeline 事件传输机制1.2.1 处理出站事件1.2.2 处理入站事件 二、ChannelPipeline 中的 ChannelHandler三、ChannelHandlerContext 接口3.1 ChannelHandlerContext 与其他组件的关系3.2 跳过某些 ChannelHandler 总结 前言
我们在前面的文章中也对ChannelPipeline接口做了初步的介绍。
Netty 概述一Netty 架构设计二Netty Channel 概述三Netty ChannelHandler四
一、ChannelPipeline 接口
ChannelPipeline接口采用了责任链设计模式底层采用双向链表的数据结构将链上的各个处理器串联起来。客户端每一个请求的到来ChannelPipeline中所有的处理器都有机会处理它。 每一个新创建的Channel都将会被分配一个新的ChannelPipeline。这项关联是永久性的Channel既不能附加另一个ChannelPipeline也不能分离其当前的。 1.1 创建 ChannelPipeline ChannelPipeline数据管道是与Channel管道绑定的一个Channel通道对应一个ChannelPipelineChannelPipeline是在Channel初始化时被创建。 观察下面这个实例
public void run() throws Exception {EventLoopGroup bossGroup new NioEventLoopGroup(); // (1)EventLoopGroup workerGroup new NioEventLoopGroup();try {ServerBootstrap b new ServerBootstrap(); // (2)b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class) // (3).childHandler(new ChannelInitializerSocketChannel() { // (4)Overridepublic void initChannel(SocketChannel ch) throws Exception {// 添加ChannelHandler到ChannelPipelinech.pipeline().addLast(new DiscardServerHandler());}}).option(ChannelOption.SO_BACKLOG, 128) // (5).childOption(ChannelOption.SO_KEEPALIVE, true); // (6)// 绑定端口开始接收进来的连接ChannelFuture f b.bind(port).sync(); // (7)System.out.println(DiscardServer已启动端口 port);// 等待服务器 socket 关闭 。// 在这个例子中这不会发生但你可以优雅地关闭你的服务器。f.channel().closeFuture().sync();} finally {workerGroup.shutdownGracefully();bossGroup.shutdownGracefully();}
}从上述代码中可以看到当ServerBootstrap初始化后直接就可以获取到SocketChannel上的ChannelPipeline而无需手动实例化因为 Netty 会为每个Channel连接创建一个ChannelPipeline。
Channel的大部分子类都继承了AbstractChannel在创建实例时也会调用AbstractChannel构造器。在AbstractChannel构造器中会创建ChannelPipeline管道实例核心代码如下
protected AbstractChannel(Channel parent) {this.parent parent;this.id this.newId();this.unsafe this.newUnsafe();this.pipeline this.newChannelPipeline();
}protected DefaultChannelPipeline newChannelPipeline() {return new DefaultChannelPipeline(this);
}从上述代码中可以看出在创建Channel时会由Channel创建DefaultChannelPipeline类的实例。DefaultChannelPipeline是ChannelPipeline的默认实现。
pipeline是AbstractChannel的属性内部维护着一个以AbstractChannelHandlerContext为节点的双向链表创建的head和tail节点分别指向链表头尾源码如下
public class DefaultChannelPipeline implements ChannelPipeline { protected DefaultChannelPipeline(Channel channel) {this.channel (Channel)ObjectUtil.checkNotNull(channel, channel);this.succeededFuture new SucceededChannelFuture(channel, (EventExecutor)null);this.voidPromise new VoidChannelPromise(channel, true);this.tail new DefaultChannelPipeline.TailContext(this);this.head new DefaultChannelPipeline.HeadContext(this);this.head.next this.tail;this.tail.prev this.head;}...final class TailContext extends AbstractChannelHandlerContext implements ChannelInboundHandler {TailContext(DefaultChannelPipeline pipeline) {super(pipeline, (EventExecutor)null, DefaultChannelPipeline.TAIL_NAME, DefaultChannelPipeline.TailContext.class);this.setAddComplete();}...}final class HeadContext extends AbstractChannelHandlerContext implements ChannelOutboundHandler, ChannelInboundHandler {private final Unsafe unsafe;HeadContext(DefaultChannelPipeline pipeline) {super(pipeline, (EventExecutor)null, DefaultChannelPipeline.HEAD_NAME, DefaultChannelPipeline.HeadContext.class);this.unsafe pipeline.channel().unsafe();this.setAddComplete();}...}...
}从上述源码可以看到TailContext和HeadContext都继承了AbstractChannelHandlerContext并实现了ChannelHandler接口。AbstractChannelHandlerContext内部维护着next、prev链表指针和入站、出站节点方向等。其中TailContext实现了ChannelInboundHandlerHeadContext实现了ChannelOutboundHandler和ChannelInboundHandler。
1.2 ChannelPipeline 事件传输机制
通过ChannelPipeline的addFirst()方法来添加ChannelHandler并为这个ChannelHandler创建一个对应的DefaultChannelHandlerContext实例。
public class DefaultChannelPipeline implements ChannelPipeline { //...public final ChannelPipeline addFirst(EventExecutorGroup group, String name, ChannelHandler handler) {AbstractChannelHandlerContext newCtx;synchronized(this) {checkMultiplicity(handler);name this.filterName(name, handler);newCtx this.newContext(group, name, handler);this.addFirst0(newCtx);if (!this.registered) {newCtx.setAddPending();this.callHandlerCallbackLater(newCtx, true);return this;}EventExecutor executor newCtx.executor();if (!executor.inEventLoop()) {this.callHandlerAddedInEventLoop(newCtx, executor);return this;}}this.callHandlerAdded0(newCtx);return this;}//...private AbstractChannelHandlerContext newContext(EventExecutorGroup group, String name, ChannelHandler handler) {return new DefaultChannelHandlerContext(this, this.childExecutor(group), name, handler);}//...}1.2.1 处理出站事件
当处理出站事件时channelRead()方法的示例如下
public class EchoServerHandler extends ChannelInboundHandlerAdapter {Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {System.out.println(ctx.channel().remoteAddress() - Server : msg);// 写消息到管道ctx.write(msg);// 写消息}//...
}上述代码中的write()方法会触发一个出站事件该方法会调用DefaultChannelPipeline上的write()方法。
public final ChannelFuture write(Object msg) {return this.tail.write(msg);
}从上述源码可以看到调用的是DefaultChannelPipeline上尾部节点tail的write方法。 上述方法最终会调用到DefaultChannelHandlerContext的write()方法。
private void write(Object msg, boolean flush, ChannelPromise promise) {ObjectUtil.checkNotNull(msg, msg);try {if (this.isNotValidPromise(promise, true)) {ReferenceCountUtil.release(msg);return;}} catch (RuntimeException var8) {ReferenceCountUtil.release(msg);throw var8;}AbstractChannelHandlerContext next this.findContextOutbound(flush ? 98304 : 耀);Object m this.pipeline.touch(msg, next);EventExecutor executor next.executor();if (executor.inEventLoop()) {if (flush) {next.invokeWriteAndFlush(m, promise);} else {next.invokeWrite(m, promise);}} else {AbstractChannelHandlerContext.WriteTask task AbstractChannelHandlerContext.WriteTask.newInstance(next, m, promise, flush);if (!safeExecute(executor, task, promise, m, !flush)) {task.cancel();}}}上述的write()方法会查找下一个出站的节点也就是当前ChannelHandler后的一个出站类型的ChannelHandler并调用下一个节点的invokeWrite()方法。
void invokeWrite(Object msg, ChannelPromise promise) {if (this.invokeHandler()) {this.invokeWrite0(msg, promise);} else {this.write(msg, promise);}}接着调用invokeWrite0()方法该方法最终调用ChannelOutboundHandler的write方法。
private void invokeWrite0(Object msg, ChannelPromise promise) {try {((ChannelOutboundHandler)this.handler()).write(this, msg, promise);} catch (Throwable var4) {notifyOutboundHandlerException(var4, promise);}}至此处理完成了第一个节点的处理开始执行下一个节点并不断循环。 所以处理出站事件时数据传输的方向是从尾部节点tail到头部节点head。
1.2.2 处理入站事件
入站事件处理的起点是触发ChannelPipeline fire方法例如fireChannelActive()方法的示例如下
public class DefaultChannelPipeline implements ChannelPipeline { //...public final ChannelPipeline fireChannelActive() {AbstractChannelHandlerContext.invokeChannelActive(this.head);return this;}//...
}从上述源码可以看到处理的节点是头部节点head。AbstractChannelHandlerContext.invokeChannelActive方法定义如下
static void invokeChannelActive(final AbstractChannelHandlerContext next) {EventExecutor executor next.executor();if (executor.inEventLoop()) {next.invokeChannelActive();} else {executor.execute(new Runnable() {public void run() {next.invokeChannelActive();}});}}该方法最终调用ChannelInboundHandler的channelActive方法。
private void invokeChannelActive() {if (this.invokeHandler()) {try {((ChannelInboundHandler)this.handler()).channelActive(this);} catch (Throwable var2) {this.invokeExceptionCaught(var2);}} else {this.fireChannelActive();}}至此完成了第一个节点的处理开始执行下一个节点的不断循环。 所以处理入站事件时数据传输的方向是从头部节点head到尾部节点tail。
二、ChannelPipeline 中的 ChannelHandler
从上述的ChannelPipeline 接口源码可以看出ChannelPipeline 是通过addXxx或者removeXxx方法来将ChannelHandler动态的添加到ChannelPipeline中或者从ChannelPipeline移除ChannelHandler的。那么ChannelPipeline是如何保障并发访问时的安全呢
以addLast方法为例DefaultChannelPipeline的源码如下
public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {AbstractChannelHandlerContext newCtx;//synchronized 保障线程安全synchronized(this) {checkMultiplicity(handler);newCtx this.newContext(group, this.filterName(name, handler), handler);this.addLast0(newCtx);if (!this.registered) {newCtx.setAddPending();this.callHandlerCallbackLater(newCtx, true);return this;}EventExecutor executor newCtx.executor();if (!executor.inEventLoop()) {this.callHandlerAddedInEventLoop(newCtx, executor);return this;}}this.callHandlerAdded0(newCtx);return this;
}从上述源码可以看到使用synchronized关键字保障了线程的安全访问。其他方法的实现方式也是类似。
三、ChannelHandlerContext 接口
ChannelHandlerContext 接口是联系ChannelHandler和ChannelPipeline 之间的纽带。
每当有ChannelHandler添加到ChannelPipeline 中时都会创建ChannelHandlerContext 。 ChannelHandlerContext 的主要功能是管理它所关联的ChannelHandler和在同一个ChannelPipeline 中的其他ChannelHandler之间的交互。
例如ChannelHandlerContext 可以通知ChannelPipeline 中的下一个ChannelHandler开始执行及动态修改其所属的ChannelPipeline 。
ChannelHandlerContext 中包含了许多方法其中一些方法也出现在Channel和ChannelPipeline 中。如果通过Channel或ChannelPipeline 的实例来调用这些方法它们就会在整个ChannelPipeline 中传播。相比之下一样的方法在ChannelHandlerContext 的实例上调用就只会从当前ChannelHandler开始并传播到相关管道中的下一个有处理事件能力的ChannelHandler中。因此ChannelHandlerContext 所包含的事件流比其他类中同样的方法都要短利用这一点可以尽可能提高性能。
3.1 ChannelHandlerContext 与其他组件的关系
下图展示了ChannelPipeline 、Channel、ChannelHandler和ChannelHandlerContext 之间的关系做了如下说明
Channel被绑定到ChannelPipeline 上。和Channel绑定的ChannelPipeline 包含了所有的ChannelHandler。ChannelHandler。当添加ChannelHandler到ChannelPipeline 时ChannelHandlerContext 被创建。
3.2 跳过某些 ChannelHandler
下面的代码展示了从ChannelHandlerContext 获取到Channel的引用并通过调用Channel上的write()方法来触发一个写事件到流中。
ChannelHandlerContext ctx context;
Channel channel ctx.channel(); //获取ChannelHandlerContext上的Channel
channel.write(msg);以下代码展示了从ChannelHandlerContext 获取到ChannelPipeline 。
ChannelHandlerContext ctx context;
ChannelPipeline pipeline ctx.pipeline(); //获取ChannelHandlerContext上的ChannelPipeline
pipeline.write(msg);上述的两个示例事件流是一样的。虽然被调用的Channel和ChannelPipeline 上的write()方法将一直传播事件通过整个ChannelPipeline 但是在ChannelHandler的级别上事件从一个ChannelHandler到下一个ChannelHandler的移动是由ChannelHandlerContext 上的调用完成的。
下图展示了Channel或者ChannelPipeline 进行的事件传播机制。
在上图中可以看出
事件传递给ChannelPipeline 的第一个ChannelHandlerChannelHandler通过关联的ChannelHandlerContext 传递事件给ChannelPipeline 中的下一个ChannelHandler。ChannelHandler通过关联的ChannelHandlerContext 传递事件给ChannelPipeline 中的下一个ChannelHandler。
从上面的流程可以看出如果通过Channel或ChannelPipeline 的实例来调用这些方法它们肯定会在整个ChannelPipeline 中传播。 那么是否可以跳过某些处理器呢答案是肯定的。 通过减少ChannelHandler不感兴趣的事件的传递减少开销并排除掉特定的对此事件感兴趣的处理器的处理以提升性能。想要实现从一个特定的ChannelHandler开始处理必须引用与此ChannelHandler的前一个ChannelHandler关联的ChannelHandlerContext 。这个ChannelHandlerContext 将会调用与自身关联的ChannelHandler的下一个ChannelHandler代码如下
ChannelHandlerContext ctx context;
ctx.write(msg);直接调用ChannelHandlerContext 的write()方法将会把缓冲区发送到下一个ChannelHandler。
如下图消息会将从下一个ChannelHandler开始流过ChannelPipeline 绕过所有在它之前的ChannelHandler。
执行ChannelHandlerContext 方法调用。事件发送到了下一个ChannelHandler。经过最后一个ChannelHandler后事件从ChannelPipeline 中移除。
当调用某个特定的ChannelHandler操作时它尤为有用。 例如
public class EchoServerHandler extends ChannelInboundHandlerAdapter {Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {System.out.println(ctx.channel().remoteAddress() - Server : msg);// 写消息到管道ctx.write(msg);// 写消息ctx.flush(); // 冲刷消息// 上面两个方法等同于 ctx.writeAndFlush(msg);}Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {// 当出现异常就关闭连接cause.printStackTrace();ctx.close();}
}总结
以上就是关于ChannelPipeline 的源码分析相信认真看完了你就明白ChannelPipeline 、Channel、ChannelHandler和ChannelHandlerContext 之间的关系。下节我们继续来剖析 Netty 的源码。
