Unsafe是托委访问socket，那么Channel是直接提供给开发者使用的
Channel 主要有两个实现 NioServerSocketChannel同NioSocketChannel 致于其它不常用不在研究范围内
NioServerSocketChannel 是给server用的，程序由始至终只有一个NioServerSocketChannel
NioSocketChannel 是给客户端用的，每个连接生成一个NioSocketChannel 对象
NioSocketChannel同NioSocketChannel的继承关系
NioSocketChannel -> AbstractNioByteChannel -> AbstractNioChannel -> AbstractChannel
NioServerSocketChannel -> AbstractNioMessageChannel-> AbstractNioChannel -> AbstractChannel
小提示：如果看文字不够直观可以在eclipse里按快捷键 选择类 ctrl+t
channel有unsafe相应的实现类，反之亦是。其实功能是很简单的，划分太多对象目的是对某部分功能重用，有时也可能因过渡设计造成
对于channel我们主要分析 I/O read/write操作
NioServerSocketChannel AbstractNioMessageChannel SelectorProvider DEFAULT_SELECTOR_PROVIDER = (= NioServerSocketChannelConfig( doReadMessages(List<Object> buf) = (ch != buf.add( NioSocketChannel( 1 0 doWrite(ChannelOutboundBuffer in)
public class NioSocketChannel extends AbstractNioByteChannel implements io.netty.channel.socket.SocketChannel { public NioSocketChannel(Channel parent, SocketChannel socket) { super(parent, socket);
config = new NioSocketChannelConfig(this, socket.socket());
}
//////////////////////////////这部分是unsafe底层调用上层的实现////////////////////////////////////////////// @Override protected int doReadBytes(ByteBuf byteBuf) throws Exception { final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle(); //这里设置byteBuf写入数据坐标 allocHandle.attemptedBytesRead(byteBuf.writableBytes()); return byteBuf.writeBytes(javaChannel(), allocHandle.attemptedBytesRead());
}
@Override protected int doWriteBytes(ByteBuf buf) throws Exception { final int expectedWrittenBytes = buf.readableBytes(); return buf.readBytes(javaChannel(), expectedWrittenBytes);
}
@Override protected void doWrite(ChannelOutboundBuffer in) throws Exception { for (;;) { int size = in.size(); //没有数据退出
if (size == 0) {
clearOpWrite(); break;
}
long writtenBytes = 0; //记录写数据size
boolean done = false; //是否完成
boolean setOpWrite = false;
ByteBuffer[] nioBuffers = in.nioBuffers(); int nioBufferCnt = in.nioBufferCount(); long expectedWrittenBytes = in.nioBufferSize();
SocketChannel ch = javaChannel(); //这里有三种分支处理 //如果没有ByteBuffer 有可能只发送几个byte //1跟default逻辑其实是一样的
switch (nioBufferCnt) { case 0: //调用父类 AbstractNioByteChannel doWrite，逻辑基本相同，不同的是AbstractNioByteChannel处理的是byte 实现调用的是 doWriteBytes(ByteBuf buf)方法。。。
super.doWrite(in); return; case 1: //这里只循环16次，可以看出是复制下面代码的哈。。。
ByteBuffer nioBuffer = nioBuffers[0]; for (int i = config().getWriteSpinCount() - 1; i >= 0; i --) { final int localWrittenBytes = ch.write(nioBuffer); if (localWrittenBytes == 0) {
setOpWrite = true; break;
}
expectedWrittenBytes -= localWrittenBytes;
writtenBytes += localWrittenBytes; if (expectedWrittenBytes == 0) {
done = true; break;
}
} break; default: //多个ByteBuffer时跟上面逻辑一样
for (int i = config().getWriteSpinCount() - 1; i >= 0; i --) { final long localWrittenBytes = ch.write(nioBuffers, 0, nioBufferCnt); if (localWrittenBytes == 0) {
setOpWrite = true; break;
}
expectedWrittenBytes -= localWrittenBytes;
writtenBytes += localWrittenBytes; if (expectedWrittenBytes == 0) {
done = true; break;
}
} break;
} // Release the fully written buffers, and update the indexes of the partially written buffer. in.removeBytes(writtenBytes); if (!done) { // Did not write all buffers completely. incompleteWrite(setOpWrite); break;
}
}
}
}
public abstract class AbstractChannel extends DefaultAttributeMap implements Channel { //生成NioSocketChannel时就绑定 unsafe pipeline
protected AbstractChannel(Channel parent) { this.parent = parent;
id = newId();
unsafe = newUnsafe();
pipeline = newChannelPipeline();
}
}protected abstract class AbstractUnsafe implements Unsafe { private void register0(ChannelPromise promise) { try { if (!promise.setUncancellable() || !ensureOpen(promise)) { return;
} boolean firstRegistration = neverRegistered;
doRegister(); // doRegister 是调用 AbstractNioChannel selectionKey = javaChannel().register(eventLoop().selector, 0, this);
neverRegistered = false;
registered = true; //这里是添加 Handler 每个Handler会生成一个Context pipeline.invokeHandlerAddedIfNeeded();
safeSetSuccess(promise); //通知Handler Registered pipeline.fireChannelRegistered(); if (isActive()) { if (firstRegistration) { //通知Handler Active pipeline.fireChannelActive();
} else if (config().isAutoRead()) {
beginRead();
}
}
} catch (Throwable t) { //....... }
}
}
小结：看似很复杂的Channel实现其实没想象难，大多数读写坐标记录交给ByteBuf处理掉了
1.server每个client连接转换成NioSocketChannel对象
2.构建NioSocketChannel时就已经生成 unsafe、pipeline
| 作者: | solq |
| 博客地址： | http://www.cnblogs.com/solq111 |
| 博客版权： | 本文以学习、研究和分享为主，欢迎转载，但必须在文章页面明显位置给出原文连接。 如果文中有不妥或者错误的地方还望高手的你指出，以免误人子弟。如果觉得本文对你有所帮助不如【推荐】一下！如果你有更好的建议，不如留言一起讨论，共同进步！ 再次感谢您耐心的读完本篇文章。 QQ群：9547527  |
http://www.cnblogs.com/solq111/p/7066208.html
延伸阅读
- 告别“老顽固” 2023-04-18 11:15:46.437
- 时光飞逝 2024-03-26 11:25:53.823
- 会当凌绝顶，一览众山小 2023-04-18 10:38:34.08
- 自豪 2023-04-18 10:31:27.963
- 绽放 2023-04-18 10:30:19.243
- 毕业有感 2023-04-18 10:27:54.643
- 投入六个月的经历，收获了整个人生 2023-04-18 10:18:55.813
- 非常响亮的名字 2023-04-18 10:10:46.923
- 不可磨灭的力量 2023-04-18 10:06:02.717
- 困难是把双刃剑 2023-04-18 10:04:30.883
告别“老顽固”