2021-05-05 16:51  阅读(88)
文章分类:Netty 学习之旅 文章标签:NettyNetty 学习
©  原文作者:中间件兴趣圈 原文地址:https://blog.csdn.net/prestigeding/article/details/53977445

1、Channel官方概述

先附上官方关于Channel的介绍。

        /**
         * A nexus to a network socket or a component which is capable of I/O
         * operations such as read, write, connect, and bind.
         * <p>
         * A channel provides a user:
         * <ul>
         * <li>the current state of the channel (e.g. is it open? is it connected?),</li>
         * <li>the {@linkplain ChannelConfig configuration parameters} of the channel (e.g. receive buffer size),</li>
         * <li>the I/O operations that the channel supports (e.g. read, write, connect, and bind), and</li>
         * <li>the {@link ChannelPipeline} which handles all I/O events and requests
         *     associated with the channel.</li>
         * </ul>
         *
         * <h3>All I/O operations are asynchronous.</h3>
         * <p>
         * All I/O operations in Netty are asynchronous.  It means any I/O calls will
         * return immediately with no guarantee that the requested I/O operation has
         * been completed at the end of the call.  Instead, you will be returned with
         * a {@link ChannelFuture} instance which will notify you when the requested I/O
         * operation has succeeded, failed, or canceled.
         *
         * <h3>Channels are hierarchical</h3>
         * <p>
         * A {@link Channel} can have a {@linkplain #parent() parent} depending on
         * how it was created.  For instance, a {@link SocketChannel}, that was accepted
         * by {@link ServerSocketChannel}, will return the {@link ServerSocketChannel}
         * as its parent on {@link #parent()}.
         * <p>
         * The semantics of the hierarchical structure depends on the transport
         * implementation where the {@link Channel} belongs to.  For example, you could
         * write a new {@link Channel} implementation that creates the sub-channels that
         * share one socket connection, as <a href="http://beepcore.org/">BEEP</a> and
         * <a href="http://en.wikipedia.org/wiki/Secure_Shell">SSH</a> do.
         *
         * <h3>Downcast to access transport-specific operations</h3>
         * <p>
         * Some transports exposes additional operations that is specific to the
         * transport.  Down-cast the {@link Channel} to sub-type to invoke such
         * operations.  For example, with the old I/O datagram transport, multicast
         * join / leave operations are provided by {@link DatagramChannel}.
         *
         * <h3>Release resources</h3>
         * <p>
         * It is important to call {@link #close()} or {@link #close(ChannelPromise)} to release all
         * resources once you are done with the {@link Channel}. This ensures all resources are
         * released in a proper way, i.e. filehandles.
         */
  • 通道状态主要包括:打开、关闭、连接。
  • 通道主要的IO操作,读(read)、写(write)、连接(connect)、绑定(bind)。
  • 所有的IO操作都是异步的,调用诸如read,write方法后,并不保证IO操作完成,但会返回一个凭证,在IO操作成功,取消或失败后会记录在该凭证中。
  • channel有父子关系,SocketChannel是通过ServerSocketChannel接受创建的,故SocketChannel的parent()方法返回的就是ServerSocketChannel。
  • 在Channel使用完毕后,请调用close方法,释放通道占用的资源。

2、Netty Channel API

接下来重点介绍一下Channel的API。

        /**
             * Returns the globally unique identifier of this {@link Channel}.
             * 返回全局唯一的channel id
             */
            ChannelId id();
    
            /**
             * Return the {@link EventLoop} this {@link Channel} was registered too.
             * 返回该Channel注册的线程模型,先理解为Ractor模型的Ractor线程。
             */
            EventLoop eventLoop();
    
            /**
             * Returns the parent of this channel.
             *
             * @return the parent channel.
             *         {@code null} if this channel does not have a parent channel.
             *    返回该Channel由谁创建的,ServerSocketChannel返回null,SocketChannel返回创建它的ServerSocketChannel
             */
            Channel parent();
    
            /**
             * Returns the configuration of this channel.
             *  返回通道的配置信息
             */
            ChannelConfig config();
    
            /**
             * Returns {@code true} if the {@link Channel} is open an may get active later
             * 通道是否打开
             */
            boolean isOpen();
    
            /**
             * Returns {@code true} if the {@link Channel} is registered with an {@link EventLoop}.
             * 该通道是否已经注册在事件模型中,此处先参考Nio编程模型,一个通过需要注册在Register上
             */
            boolean isRegistered();
    
            /**
             * Return {@code true} if the {@link Channel} is active and so connected.
             *  通道是否激活
             */
            boolean isActive();
    
            /**
             * Return the {@link ChannelMetadata} of the {@link Channel} which describe the nature of the {@link Channel}.
             *  通道是否支持 调用disconnect方法后,调用connect方法
             */
            ChannelMetadata metadata();
    
            /**
             * Returns the local address where this channel is bound to.  The returned
             * {@link SocketAddress} is supposed to be down-cast into more concrete
             * type such as {@link InetSocketAddress} to retrieve the detailed
             * information.
             *
             * @return the local address of this channel.
             *         {@code null} if this channel is not bound.
              *      返回绑定的地址,服务端的Channel返回监听的地址,而客户端的Channel返回连接到服务端的本地套接字。
             */
            SocketAddress localAddress();
    
            /**
             * Returns the remote address where this channel is connected to.  The
             * returned {@link SocketAddress} is supposed to be down-cast into more
             * concrete type such as {@link InetSocketAddress} to retrieve the detailed
             * information.
             *
             * @return the remote address of this channel.
             *         {@code null} if this channel is not connected.
             *         If this channel is not connected but it can receive messages
             *         from arbitrary remote addresses (e.g. {@link DatagramChannel},
             *         use {@link DatagramPacket#recipient()} to determine
             *         the origination of the received message as this method will
             *         return {@code null}.
             *         返回channel的远程套接字。
             */
            SocketAddress remoteAddress();
    
            /**
             * Returns the {@link ChannelFuture} which will be notified when this
             * channel is closed.  This method always returns the same future instance.
             *  通道的关闭凭证(许可),这里是多线程编程一种典型的设计模式,一个channle返回一个固定的
             */
            ChannelFuture closeFuture();
    
            /**
             * Returns {@code true} if and only if the I/O thread will perform the
             * requested write operation immediately.  Any write requests made when
             * this method returns {@code false} are queued until the I/O thread is
             * ready to process the queued write requests.
             *  是否可写,如果通道的写缓冲区未满,即返回true,表示写操作可以立即     
             *   操作缓冲区,然后返回。
    
             */
            boolean isWritable();
    
            /**
             * Returns an <em>internal-use-only</em> object that provides unsafe operations.
             */
            Unsafe unsafe();
    
            /**
             * Return the assigned {@link ChannelPipeline}
             *  返回管道
             */
            ChannelPipeline pipeline();
    
            /**
             * Return the assigned {@link ByteBufAllocator} which will be used to allocate {@link ByteBuf}s.
             * 返回ByteBuf内存分配器
             */
            ByteBufAllocator alloc();
    
            /**
             * Return a new {@link ChannelPromise}.
             * 诸如newPromise,newSuccessedFuture()方法,就是返回一个凭证,用来保存通知结果的,是多线程编程一         * 中典型的设计模式
             */
            ChannelPromise newPromise();
    
            /**
             * Return an new {@link ChannelProgressivePromise}
             */
            ChannelProgressivePromise newProgressivePromise();
    
            /**
             * Create a new {@link ChannelFuture} which is marked as succeeded already. So {@link ChannelFuture#isSuccess()}
             * will return {@code true}. All {@link FutureListener} added to it will be notified directly. Also
             * every call of blocking methods will just return without blocking.
             */
            ChannelFuture newSucceededFuture();
    
            /**
             * Create a new {@link ChannelFuture} which is marked as failed already. So {@link ChannelFuture#isSuccess()}
             * will return {@code false}. All {@link FutureListener} added to it will be notified directly. Also
             * every call of blocking methods will just return without blocking.
             */
            ChannelFuture newFailedFuture(Throwable cause);
    
            /**
             * Return a special ChannelPromise which can be reused for different operations.
             * <p>
             * It's only supported to use
             * it for {@link Channel#write(Object, ChannelPromise)}.
             * </p>
             * <p>
             * Be aware that the returned {@link ChannelPromise} will not support most operations and should only be used
             * if you want to save an object allocation for every write operation. You will not be able to detect if the
             * operation  was complete, only if it failed as the implementation will call
             * {@link ChannelPipeline#fireExceptionCaught(Throwable)} in this case.
             * </p>
             * <strong>Be aware this is an expert feature and should be used with care!</strong>
             */
            ChannelPromise voidPromise();
    
            /**
             * Request to bind to the given {@link SocketAddress} and notify the {@link ChannelFuture} once the operation
             * completes, either because the operation was successful or because of an error.
             * <p>
             * This will result in having the
             * {@link ChannelHandler#bind(ChannelHandlerContext, SocketAddress, ChannelPromise)} method
             * called of the next {@link ChannelHandler} contained in the  {@link ChannelPipeline} of the
             * {@link Channel}.
             *  绑定
             */
            ChannelFuture bind(SocketAddress localAddress);
    
            /**
             * Request to connect to the given {@link SocketAddress} and notify the {@link ChannelFuture} once the operation
             * completes, either because the operation was successful or because of an error.
             * <p>
             * If the connection fails because of a connection timeout, the {@link ChannelFuture} will get failed with
             * a {@link ConnectTimeoutException}. If it fails because of connection refused a {@link ConnectException}
             * will be used.
             * <p>
             * This will result in having the
             * {@link ChannelHandler#connect(ChannelHandlerContext, SocketAddress, SocketAddress, ChannelPromise)}
             * method called of the next {@link ChannelHandler} contained in the  {@link ChannelPipeline} of the
             * {@link Channel}.
             * 连接
             */
            ChannelFuture connect(SocketAddress remoteAddress);
    
            /**
             * Request to connect to the given {@link SocketAddress} while bind to the localAddress and notify the
             * {@link ChannelFuture} once the operation completes, either because the operation was successful or because of
             * an error.
             * <p>
             * This will result in having the
             * {@link ChannelHandler#connect(ChannelHandlerContext, SocketAddress, SocketAddress, ChannelPromise)}
             * method called of the next {@link ChannelHandler} contained in the  {@link ChannelPipeline} of the
             * {@link Channel}.
             */
            ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress);
    
            /**
             * Request to disconnect from the remote peer and notify the {@link ChannelFuture} once the operation completes,
             * either because the operation was successful or because of an error.
             * <p>
             * This will result in having the
             * {@link ChannelHandler#disconnect(ChannelHandlerContext, ChannelPromise)}
             * method called of the next {@link ChannelHandler} contained in the  {@link ChannelPipeline} of the
             * {@link Channel}.
             * 断开连接
             */
            ChannelFuture disconnect();
    
            /**
             * Request to close this {@link Channel} and notify the {@link ChannelFuture} once the operation completes,
             * either because the operation was successful or because of
             * an error.
             *
             * After it is closed it is not possible to reuse it again.
             * <p>
             * This will result in having the
             * {@link ChannelHandler#close(ChannelHandlerContext, ChannelPromise)}
             * method called of the next {@link ChannelHandler} contained in the  {@link ChannelPipeline} of the
             * {@link Channel}.
             *  关闭,释放通道资源
             */
            ChannelFuture close();
    
            /**
             * Request to deregister this {@link Channel} from its assigned {@link EventLoop} and notify the
             * {@link ChannelFuture} once the operation completes, either because the operation was successful or because of
             * an error.
             * <p>
             * This will result in having the
             * {@link ChannelHandler#deregister(ChannelHandlerContext, ChannelPromise)}
             * method called of the next {@link ChannelHandler} contained in the  {@link ChannelPipeline} of the
             * {@link Channel}.
             * <p>
             * After this method completes (not the {@link ChannelFuture}!) one can not submit new tasks to the
             * {@link Channel}'s {@link EventLoop} until the {@link Channel} is again registered with an {@link EventLoop}.
             * Any attempt to do so will result in a {@link RejectedExecutionException} being thrown.
             * Any tasks that were submitted before the call to {@link #deregister()} will finish before the
             * {@link ChannelFuture} completes. Furthermore, periodic and delayed tasks will not be executed until the
             * {@link Channel} is registered with an {@link EventLoop} again. Theses are tasks submitted
             * to the {@link EventLoop} via one of the methods declared by {@link ScheduledExecutorService}.
             * Please note that all of the above only applies to tasks created from within the deregistered {@link Channel}'s
             * {@link ChannelHandler}s.
             * <p>
             * It's only safe to {@linkplain EventLoop#register(Channel)} the {@link Channel} with another (or the same)
             * {@link EventLoop} after the {@link ChannelFuture} has completed.
             */
            ChannelFuture deregister();
    
            /**
             * Request to bind to the given {@link SocketAddress} and notify the {@link ChannelFuture} once the operation
             * completes, either because the operation was successful or because of an error.
             *
             * The given {@link ChannelPromise} will be notified.
             * <p>
             * This will result in having the
             * {@link ChannelHandler#bind(ChannelHandlerContext, SocketAddress, ChannelPromise)} method
             * called of the next {@link ChannelHandler} contained in the  {@link ChannelPipeline} of the
             * {@link Channel}.
             */
            ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise);
    
            /**
             * Request to connect to the given {@link SocketAddress} and notify the {@link ChannelFuture} once the operation
             * completes, either because the operation was successful or because of an error.
             *
             * The given {@link ChannelFuture} will be notified.
             *
             * <p>
             * If the connection fails because of a connection timeout, the {@link ChannelFuture} will get failed with
             * a {@link ConnectTimeoutException}. If it fails because of connection refused a {@link ConnectException}
             * will be used.
             * <p>
             * This will result in having the
             * {@link ChannelHandler#connect(ChannelHandlerContext, SocketAddress, SocketAddress, ChannelPromise)}
             * method called of the next {@link ChannelHandler} contained in the  {@link ChannelPipeline} of the
             * {@link Channel}.
             */
            ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise);
    
            /**
             * Request to connect to the given {@link SocketAddress} while bind to the localAddress and notify the
             * {@link ChannelFuture} once the operation completes, either because the operation was successful or because of
             * an error.
             *
             * The given {@link ChannelPromise} will be notified and also returned.
             * <p>
             * This will result in having the
             * {@link ChannelHandler#connect(ChannelHandlerContext, SocketAddress, SocketAddress, ChannelPromise)}
             * method called of the next {@link ChannelHandler} contained in the  {@link ChannelPipeline} of the
             * {@link Channel}.
             */
            ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise);
    
            /**
             * Request to disconnect from the remote peer and notify the {@link ChannelFuture} once the operation completes,
             * either because the operation was successful or because of an error.
             *
             * The given {@link ChannelPromise} will be notified.
             * <p>
             * This will result in having the
             * {@link ChannelHandler#disconnect(ChannelHandlerContext, ChannelPromise)}
             * method called of the next {@link ChannelHandler} contained in the  {@link ChannelPipeline} of the
             * {@link Channel}.
             */
            ChannelFuture disconnect(ChannelPromise promise);
    
            /**
             * Request to close this {@link Channel} and notify the {@link ChannelFuture} once the operation completes,
             * either because the operation was successful or because of
             * an error.
             *
             * After it is closed it is not possible to reuse it again.
             * The given {@link ChannelPromise} will be notified.
             * <p>
             * This will result in having the
             * {@link ChannelHandler#close(ChannelHandlerContext, ChannelPromise)}
             * method called of the next {@link ChannelHandler} contained in the  {@link ChannelPipeline} of the
             * {@link Channel}.
             */
            ChannelFuture close(ChannelPromise promise);
    
            /**
             * Request to deregister this {@link Channel} from its assigned {@link EventLoop} and notify the
             * {@link ChannelPromise} once the operation completes, either because the operation was successful or because of
             * an error.
             * <p>
             * This will result in having the
             * {@link ChannelHandler#deregister(ChannelHandlerContext, ChannelPromise)}
             * method called of the next {@link ChannelHandler} contained in the  {@link ChannelPipeline} of the
             * {@link Channel}.
             * <p>
             * After this method completes (not the {@link ChannelPromise}!) one can not submit new tasks to the
             * {@link Channel}'s {@link EventLoop} until the {@link Channel} is again registered with an {@link EventLoop}.
             * Any attempt to do so will result in a {@link RejectedExecutionException} being thrown.
             * Any tasks that were submitted before the call to {@link #deregister()} will finish before the
             * {@link ChannelPromise} completes. Furthermore, periodic and delayed tasks will not be executed until the
             * {@link Channel} is registered with an {@link EventLoop} again. Theses are tasks submitted
             * to the {@link EventLoop} via one of the methods declared by {@link ScheduledExecutorService}.
             * Please note that all of the above only applies to tasks created from within the deregistered {@link Channel}'s
             * {@link ChannelHandler}s.
             * <p>
             * It's only safe to {@linkplain EventLoop#register(Channel)} the {@link Channel} with another (or the same)
             * {@link EventLoop} after the {@link ChannelPromise} has completed.
             */
            ChannelFuture deregister(ChannelPromise promise);
    
            /**
             * Request to Read data from the {@link Channel} into the first inbound buffer, triggers an
             * {@link ChannelHandler#channelRead(ChannelHandlerContext, Object)} event if data was
             * read, and triggers a
             * {@link ChannelHandler#channelReadComplete(ChannelHandlerContext) channelReadComplete} event so the
             * handler can decide to continue reading.  If there's a pending read operation already, this method does nothing.
             * <p>
             * This will result in having the
             * {@link ChannelHandler#read(ChannelHandlerContext)}
             * method called of the next {@link ChannelHandler} contained in the  {@link ChannelPipeline} of the
             * {@link Channel}.
             */
            Channel read();
    
            /**
             * Request to write a message via this {@link Channel} through the {@link ChannelPipeline}.
             * This method will not request to actual flush, so be sure to call {@link #flush()}
             * once you want to request to flush all pending data to the actual transport.
             */
            ChannelFuture write(Object msg);
    
            /**
             * Request to write a message via this {@link Channel} through the {@link ChannelPipeline}.
             * This method will not request to actual flush, so be sure to call {@link #flush()}
             * once you want to request to flush all pending data to the actual transport.
             */
            ChannelFuture write(Object msg, ChannelPromise promise);
    
            /**
             * Request to flush all pending messages.
             */
            Channel flush();
    
            /**
             * Shortcut for call {@link #write(Object, ChannelPromise)} and {@link #flush()}.
             */
            ChannelFuture writeAndFlush(Object msg, ChannelPromise promise);
    
            /**
             * Shortcut for call {@link #write(Object)} and {@link #flush()}.
             */
            ChannelFuture writeAndFlush(Object msg);
    
            /**
             * <em>Unsafe</em> operations that should <em>never</em> be called from user-code. These methods
             * are only provided to implement the actual transport, and must be invoked from an I/O thread except for the
             * following methods:
             * <ul>
             *   <li>{@link #invoker()}</li>
             *   <li>{@link #localAddress()}</li>
             *   <li>{@link #remoteAddress()}</li>
             *   <li>{@link #closeForcibly()}</li>
             *   <li>{@link #register(EventLoop, ChannelPromise)}</li>
             *   <li>{@link #deregister(ChannelPromise)}</li>
             *   <li>{@link #voidPromise()}</li>
             * </ul>
             */
            interface Unsafe {
    
                /**
                 * Return the assigned {@link RecvByteBufAllocator.Handle} which will be used to allocate {@link ByteBuf}'s when
                 * receiving data.
                 */
                RecvByteBufAllocator.Handle recvBufAllocHandle();
    
                /**
                 * Returns the {@link ChannelHandlerInvoker} which is used by default unless specified by a user.
                 */
                ChannelHandlerInvoker invoker();
    
                /**
                 * Return the {@link SocketAddress} to which is bound local or
                 * {@code null} if none.
                 */
                SocketAddress localAddress();
    
                /**
                 * Return the {@link SocketAddress} to which is bound remote or
                 * {@code null} if none is bound yet.
                 */
                SocketAddress remoteAddress();
    
                /**
                 * Register the {@link Channel} of the {@link ChannelPromise} and notify
                 * the {@link ChannelFuture} once the registration was complete.
                 * <p>
                 * It's only safe to submit a new task to the {@link EventLoop} from within a
                 * {@link ChannelHandler} once the {@link ChannelPromise} succeeded. Otherwise
                 * the task may or may not be rejected.
                 * </p>
                 */
                void register(EventLoop eventLoop, ChannelPromise promise);
    
                /**
                 * Bind the {@link SocketAddress} to the {@link Channel} of the {@link ChannelPromise} and notify
                 * it once its done.
                 */
                void bind(SocketAddress localAddress, ChannelPromise promise);
    
                /**
                 * Connect the {@link Channel} of the given {@link ChannelFuture} with the given remote {@link SocketAddress}.
                 * If a specific local {@link SocketAddress} should be used it need to be given as argument. Otherwise just
                 * pass {@code null} to it.
                 *
                 * The {@link ChannelPromise} will get notified once the connect operation was complete.
                 */
                void connect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise);
    
                /**
                 * Disconnect the {@link Channel} of the {@link ChannelFuture} and notify the {@link ChannelPromise} once the
                 * operation was complete.
                 */
                void disconnect(ChannelPromise promise);
    
                /**
                 * Close the {@link Channel} of the {@link ChannelPromise} and notify the {@link ChannelPromise} once the
                 * operation was complete.
                 */
                void close(ChannelPromise promise);
    
                /**
                 * Closes the {@link Channel} immediately without firing any events.  Probably only useful
                 * when registration attempt failed.
                 */
                void closeForcibly();
    
                /**
                 * Deregister the {@link Channel} of the {@link ChannelPromise} from {@link EventLoop} and notify the
                 * {@link ChannelPromise} once the operation was complete.
                 */
                void deregister(ChannelPromise promise);
    
                /**
                 * Schedules a read operation that fills the inbound buffer of the first {@link ChannelHandler} in the
                 * {@link ChannelPipeline}.  If there's already a pending read operation, this method does nothing.
                 */
                void beginRead();
    
                /**
                 * Schedules a write operation.
                 */
                void write(Object msg, ChannelPromise promise);
    
                /**
                 * Flush out all write operations scheduled via {@link #write(Object, ChannelPromise)}.
                 */
                void flush();
    
                /**
                 * Return a special ChannelPromise which can be reused and passed to the operations in {@link Unsafe}.
                 * It will never be notified of a success or error and so is only a placeholder for operations
                 * that take a {@link ChannelPromise} as argument but for which you not want to get notified.
                 */
                ChannelPromise voidPromise();
    
                /**
                 * Returns the {@link ChannelOutboundBuffer} of the {@link Channel} where the pending write requests are stored.
                 * 返回通道的环形缓存区
                 */
                ChannelOutboundBuffer outboundBuffer();
            }

3、Netty Channel主要类图

202105051651296131.png
本文关于Netty Channel接口就描述到这里,主要从Channel概述、Channel接口API,Channel类图三个方面简单介绍一下Channel,为进一步解剖Netty Channel,ChanelPiple,ChannelHandler等打下基础。

4、Netty Channel 研究方向(不成熟想法)

2.1 Channel接口概述

2.2、AbstractChannel源码学习

2.2.1 环形缓存区

2.2.2 Netty线程模型

2.2.3 Chanel read,write方法实现探究

2.3、ServerChannel

2.4、SOcketChannel


来源:https://blog.csdn.net/prestigeding/article/details/53977445

点赞(0)
版权归原创作者所有,任何形式转载请联系作者; Java 技术驿站 >> Netty Channel 概述
上一篇
hreadLocal原理分析与性能优化思考(思考篇)
下一篇
线程模型前置篇Reactor反应堆设计模式实现(基于java.nio)