netty channel
Posted benjious
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了netty channel相关的知识,希望对你有一定的参考价值。
问题
- channel 是如何处理发送一半中断后继续重发的
- channel 具体作用是什么
概述
这一节我们将介绍 Channel 和内部接口 Unsafe .其中Unsafe 是内部接口,聚合在Channel 中协助网络读写操作相关的操作,设计初衷就是 Channel 的内部辅助类,不应该被用户使用。
继承类分析
继承关系链 :
AbstractChannel -> AbstractNioChannel -> AbstractNioByteChannel -> NiosocketChannel 如下图 从以上的类结构我们也要学习一下类的构建,各个类实现应该实现的功能,最后生成的具体类具有不同的功能。 AbstractChannel ,保存以下重要的字段 ,主要 - EventLoop - localAddress - remoteAddress - unsafe - DefaultChannelPipleline - Future类 和 Promise类 等
AbstractNioChannel,从类名可以看出和nio 中 Channel 相关,注册,监听
private final SelectableChannel ch; protected final int readInterestOp; private volatile SelectionKey selectionKey; private volatile boolean inputShutdown; /** * The future of the current connection attempt. If not null, subsequent * connection attempts will fail. */ private ChannelPromise connectPromise; private ScheduledFuture<?> connectTimeoutFuture; private SocketAddress requestedRemoteAddress;
AbstractNioByteChannel 这个类是Channel对Byte进行操作,对ByteBuff的读写。
源码分析
AbstractChannel
AbstractChannel 的读写方法都是交由 ChannelPiple 来解决的
@Override public Channel read() pipeline.read(); return this; @Override public ChannelFuture write(Object msg) return pipeline.write(msg);
eventLoop方法,直接返回持有的 eventloop对象
@Override public EventLoop eventLoop() return eventLoop;
AbstractNioChannel
public abstract class AbstractNioChannel extends AbstractChannel private static final InternalLogger logger = InternalLoggerFactory.getInstance(AbstractNioChannel.class); // No.1 注册监听相关的字段 private final SelectableChannel ch; protected final int readInterestOp; private volatile SelectionKey selectionKey; private volatile boolean inputShutdown; // No.2 异步执行的字段,或是回调相关的字段 /** * The future of the current connection attempt. If not null, subsequent * connection attempts will fail. */ private ChannelPromise connectPromise; private ScheduledFuture<?> connectTimeoutFuture; private SocketAddress requestedRemoteAddress; ... //核心方法 @Override protected void doRegister() throws Exception boolean selected = false; for (;;) try //拿父类的channel对象(父类的channel对象是java原生channel 对象) selectionKey = javaChannel().register(eventLoop().selector, 0, this); return; catch (CancelledKeyException e) if (!selected) // Force the Selector to select now as the "canceled" SelectionKey may still be // cached and not removed because no Select.select(..) operation was called yet. eventLoop().selectNow(); selected = true; else // We forced a select operation on the selector before but the SelectionKey is still cached // for whatever reason. JDK bug ? throw e; //开始read的操作 @Override protected void doBeginRead() throws Exception if (inputShutdown) return; final SelectionKey selectionKey = this.selectionKey; if (!selectionKey.isValid()) return; //就是改变监听的事件 final int interestOps = selectionKey.interestOps(); if ((interestOps & readInterestOp) == 0) selectionKey.interestOps(interestOps | readInterestOp);
AbstractNioByteChannel
@Override protected void doWrite(ChannelOutboundBuffer in) throws Exception int writeSpinCount = -1; //循环 for (;;) Object msg = in.current(true); if (msg == null) // Wrote all messages.写完了(发送完了)所有的消息,清除标志,结束 clearOpWrite(); break; if (msg instanceof ByteBuf) //加入是ByteBuf类型 ByteBuf buf = (ByteBuf) msg; int readableBytes = buf.readableBytes(); //判断当前的可读字节是否为 0 ,为 0 丢弃掉 if (readableBytes == 0) in.remove(); continue; boolean setOpWrite = false; boolean done = false; long flushedAmount = 0; //循环发送次数 if (writeSpinCount == -1) writeSpinCount = config().getWriteSpinCount(); for (int i = writeSpinCount - 1; i >= 0; i --) //doWriteBytes 子类实现 int localFlushedAmount = doWriteBytes(buf); if (localFlushedAmount == 0) setOpWrite = true; break; flushedAmount += localFlushedAmount; //一直到不可读 if (!buf.isReadable()) done = true; break; //发送完,更新发送的进度(有可能没发完) in.progress(flushedAmount); if (done) in.remove(); else //没发完,设置写半包标识,启动刷新线程继续发送之前没有发送完成的半包消息 incompleteWrite(setOpWrite); break; else if (msg instanceof FileRegion) FileRegion region = (FileRegion) msg; boolean setOpWrite = false; boolean done = false; long flushedAmount = 0; if (writeSpinCount == -1) writeSpinCount = config().getWriteSpinCount(); //循环发送 for (int i = writeSpinCount - 1; i >= 0; i --) long localFlushedAmount = doWriteFileRegion(region); if (localFlushedAmount == 0) setOpWrite = true; break; flushedAmount += localFlushedAmount; if (region.transfered() >= region.count()) done = true; break; //发送完(有可能发送了一半)更新进度 in.progress(flushedAmount); if (done) in.remove(); else //没法完,创建一个任务扔到EventLoop incompleteWrite(setOpWrite); break; else throw new UnsupportedOperationException("unsupported message type: " + StringUtil.simpleClassName(msg)); //没写完(没发送完) protected final void incompleteWrite(boolean setOpWrite) // Did not write completely. if (setOpWrite) setOpWrite(); else // Schedule flush again later so other tasks can be picked up in the meantime //创建任务扔到 eventLoop执行 Runnable flushTask = this.flushTask; if (flushTask == null) flushTask = this.flushTask = new Runnable() @Override public void run() flush(); ; eventLoop().execute(flushTask);
循环发送次数是指一次发送没有完成时(写半包),程序就继续尝试循环写操作,此时IO线程是不能处理其他事件的,例如读新的消息或者执行定时任务和 NioTask 等, 如果网络IO阻塞或者对方接收消息太慢,可能会导致线程假死,于是就要循环发送。
AbstractNioMessageChannel
我们再来看一下AbstractNioChannel 的另外一个子类 AbstractNioMessageChannel,直接看doWrite方法
@Override protected void doWrite(ChannelOutboundBuffer in) throws Exception final SelectionKey key = selectionKey(); final int interestOps = key.interestOps(); for (;;) Object msg = in.current(); if (msg == null) // Wrote all messages. if ((interestOps & SelectionKey.OP_WRITE) != 0) key.interestOps(interestOps & ~SelectionKey.OP_WRITE); break; boolean done = false; for (int i = config().getWriteSpinCount() - 1; i >= 0; i --) if (doWriteMessage(msg, in)) done = true; break; if (done) in.remove(); else // Did not write all messages. //没发送完,设置标志,交给 select 多路复用器轮询对应的channel重新发送尚未发送完成的半包信息 if ((interestOps & SelectionKey.OP_WRITE) == 0) key.interestOps(interestOps | SelectionKey.OP_WRITE); break;
AbstractNioMessageChannel 和 AbstractNioByteChannel的区别在于
NioServerSocketChannel 和 NioServerChannel 的分析
NioSocketChannel 和 NioServerSocketChannel 的区别到底是什么?后者是服务端当中负责绑定端口,读取数据功能,连接和断开,写消息都不支持,这些功能都在NioSocketChannel中实现
AbstractNioMessageServerChannel 的具体子类是 NioServerSocketChannel(该类是服务器端接受处理客户端的channel),它的doReadMessages方法(被对应的unsafe类read方法,这里可能有点饶,具体看代码实现)分析如下
@Override protected int doReadMessages(List<Object> buf) throws Exception SocketChannel ch = javaChannel().accept(); try if (ch != null) //构建一个NioSocketChannel放进数组中 buf.add(new NioSocketChannel(this, childEventLoopGroup().next(), ch)); return 1; catch (Throwable t) logger.warn("Failed to create a new channel from an accepted socket.", t); try ch.close(); catch (Throwable t2) logger.warn("Failed to close a socket.", t2); return 0;
NioServerChannel的源码分析
public class NioServerSocketChannel extends AbstractNioMessageServerChannel implements io.netty.channel.socket.ServerSocketChannel private static final ChannelMetadata METADATA = new ChannelMetadata(false); private static final InternalLogger logger = InternalLoggerFactory.getInstance(NioServerSocketChannel.class); private static ServerSocketChannel newSocket() try return ServerSocketChannel.open(); catch (IOException e) throw new ChannelException( "Failed to open a server socket.", e); private final ServerSocketChannelConfig config; /** * Create a new instance */ public NioServerSocketChannel(EventLoop eventLoop, EventLoopGroup childGroup) super(null, eventLoop, childGroup, newSocket(), SelectionKey.OP_ACCEPT); config = new DefaultServerSocketChannelConfig(this, javaChannel().socket()); @Override public InetSocketAddress localAddress() return (InetSocketAddress) super.localAddress(); @Override public ChannelMetadata metadata() return METADATA; @Override public ServerSocketChannelConfig config() return config; @Override public boolean isActive() return javaChannel().socket().isBound(); @Override public InetSocketAddress remoteAddress() return null; @Override protected ServerSocketChannel javaChannel() return (ServerSocketChannel) super.javaChannel(); @Override protected SocketAddress localAddress0() return javaChannel().socket().getLocalSocketAddress(); @Override protected void doBind(SocketAddress localAddress) throws Exception javaChannel().socket().bind(localAddress, config.getBacklog()); @Override protected void doClose() throws Exception javaChannel().close(); @Override protected int doReadMessages(List<Object> buf) throws Exception SocketChannel ch = javaChannel().accept(); try if (ch != null) buf.add(new NioSocketChannel(this, childEventLoopGroup().next(), ch)); return 1; catch (Throwable t) logger.warn("Failed to create a new channel from an accepted socket.", t); try ch.close(); catch (Throwable t2) logger.warn("Failed to close a socket.", t2); return 0; // Unnecessary stuff @Override protected boolean doConnect( SocketAddress remoteAddress, SocketAddress localAddress) throws Exception throw new UnsupportedOperationException(); @Override protected void doFinishConnect() throws Exception throw new UnsupportedOperationException(); @Override protected SocketAddress remoteAddress0() return null; @Override protected void doDisconnect() throws Exception throw new UnsupportedOperationException(); @Override protected boolean doWriteMessage(Object msg, ChannelOutboundBuffer in) throws Exception throw new UnsupportedOperationException();
可以看到 NioServerChannel 的主要都是 override 父类的方法,即是说大部分的逻辑都在父类 Abstract中进行了一层层的封装,给我们一个启发,好的类结构在
在一开始就已经设计好,最终的具体实现交由尾端实现。
总结
本文介绍了channel的主要功能作用。
参考资料
- 《Netty权威指南》
以上是关于netty channel的主要内容,如果未能解决你的问题,请参考以下文章
Golang面经ChannelContextGoroutine
Golang面经ChannelContextGoroutine