Netty 的源码分析和业务场景

public interface Channel extends AttributeMap {
    /**     * Returns the {@link ChannelId} of this {@link Channel}.     */    ChannelId id();
    /**     * Returns the parent {@link Channel} of this channel. {@code null} if this is the top-level channel.     */    Channel parent();
    /**     * Returns the {@link ChannelConfig} of this channel.     */    ChannelConfig config();
    /**     * Returns the local address of this channel.     */   SocketAddress localAddress();
    /**     * Returns the remote address of this channel. {@code null} if the channel is not connected.     */    SocketAddress remoteAddress();
    /**     * Returns {@code true} if this channel is open and may be used.     */    boolean isOpen();
    /**     * Returns {@code true} if this channel is active and may be used for IO.     */    boolean isActive();
    /**     * Returns the {@link ChannelPipeline}.     */    ChannelPipeline pipeline();
    /**     * Returns the {@link ChannelFuture} which is fired once the channel is registered with its {@link EventLoop}.     */    ChannelFuture whenRegistered();
    /**     * Returns the {@link ChannelFuture} which is fired once the channel is deregistered from its {@link EventLoop}.     */    ChannelFuture whenDeregistered();
    /**     * Returns the {@link ChannelFuture} which is fired once the channel is closed.     */    ChannelFuture whenClosed();
    /**     * Register this channel to the given {@link EventLoop}.     */    ChannelFuture register(EventLoop loop);
    /**     * Bind and listen for incoming connections.     */    ChannelFuture bind(SocketAddress localAddress);
    /**     * Connect to the given remote address.     */    ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress);
    /**     * Disconnect if connected.     */    ChannelFuture disconnect();
    /**     * Close this channel.     */    ChannelFuture close();
    /**     * Deregister this channel from its {@link EventLoop}.     */    ChannelFuture deregister();
    /**     * Write the specified message to this channel.     */    ChannelFuture write(Object msg);
    /**     * Write the specified message to this channel and generate a {@link ChannelFuture} which is done     * when the message is written.     */    ChannelFuture writeAndFlush(Object msg);
    /**     * Flushes all pending messages.     */    ChannelFuture flush();
    // ... 更多方法定义}

public interface EventLoopGroup extends ExecutorService {
    /**     * Returns the next {@link EventLoop} this group will use to handle an event.     * This will either return an existing or a new instance depending on the implementation.     */    EventLoop next();
    /**     * Shuts down all {@link EventLoop}s and releases all resources.     */    ChannelFuture shutdownGracefully();
    /**     * Shuts down all {@link EventLoop}s and releases all resources.     */    ChannelFuture shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit);
    /**     * Returns a copy of the list of all {@link EventLoop}s that are part of this group.     */    List<EventLoop> eventLoops();}

public class NettyServer {
    public static void main(String[] args) {        EventLoopGroup bossGroup = new NioEventLoopGroup(1); // 用于接受连接        EventLoopGroup workerGroup = new NioEventLoopGroup(); // 用于处理连接
        try {            ServerBootstrap b = new ServerBootstrap();            b.group(bossGroup, workerGroup)             .channel(NioServerSocketChannel.class)             .childHandler(new ChannelInitializer<SocketChannel>() {                 @Override                 public void initChannel(SocketChannel ch) throws Exception {                     ChannelPipeline p = ch.pipeline();                     p.addLast(new LoggingHandler());                     p.addLast(new MyServerHandler());                 }             });
            ChannelFuture f = b.bind(8080).sync(); // 绑定端口并启动服务器            System.out.println("Server started on port 8080");            f.channel().closeFuture().sync();        } catch (InterruptedException e) {            e.printStackTrace();        } finally {            bossGroup.shutdownGracefully();            workerGroup.shutdownGracefully();        }    }}

public interface ChannelPipeline extends Iterable<ChannelHandler> {
    /**     * Add the specified handler to the context of the current channel.     */    void addLast(EventExecutorGroup executor, String name, ChannelHandler handler);
    /**     * Add the specified handlers to the context of the current channel.     */    void addLast(EventExecutorGroup executor, ChannelHandler... handlers);
    // ... 省略其他 addFirst, addBefore, addAfter, remove, replace 方法
    /**     * Get the {@link ChannelHandler} by its name.     */    ChannelHandler get(String name);
    /**     * Find the first {@link ChannelHandler} in the {@link ChannelPipeline} that matches the specified class.     */    ChannelHandler first();
    /**     * Find the last {@link ChannelHandler} in the {@link ChannelPipeline} that matches the specified class.     */    ChannelHandler last();
    /**     * Returns the context object of the specified handler.     */    ChannelHandlerContext context(ChannelHandler handler);
    // ... 省略 contextFor, remove, replace, fireChannelRegistered, fireChannelUnregistered 等方法}

public interface ChannelHandlerContext extends AttributeMap, ResourceLeakHint {
    /**     * Return the current channel to which this context is bound.     */    Channel channel();
    /**     * Return the current pipeline to which this context is bound.     */    ChannelPipeline pipeline();
    /**     * Return the name of the {@link ChannelHandler} which is represented by this context.     */    String name();
    /**     * Return the {@link ChannelHandler} which is represented by this context.     */    ChannelHandler handler();
    // ... 省略其他方法}

private final AbstractChannelHandlerContext head;private final AbstractChannelHandlerContext tail;private final List<ChannelHandler> handlers = new ArrayList<ChannelHandler>();

public void addLast(EventExecutorGroup executor, String name, ChannelHandler handler) {    if (handler == null) {        throw new NullPointerException("handler");    }    if (name == null) {        throw new NullPointerException("name");    }    AbstractChannelHandlerContext newCtx = new TailContext(this, executor, name, handler);    synchronized (this) {        if (tail == null) {            head = tail = newCtx;        } else {            tail.next = newCtx;            newCtx.prev = tail;            tail = newCtx;        }        handlers.add(newCtx);    }}

ChannelPipeline pipeline = channel.pipeline();pipeline.addLast("myHandler", new MyChannelHandler());

// EventLoop 的事件循环public void run() {    for (;;) {        // ...        processSelectedKeys();        // ...    }}

// 伪代码，展示了事件如何被传递到 ChannelHandlerif (channelRead) {    pipeline.fireChannelRead(msg);}

// 入站事件处理public void channelRead(ChannelHandlerContext ctx, Object msg) {    // 处理消息或传递给下一个处理器    ctx.fireChannelRead(msg);}
// 出站事件处理public void write(ChannelHandlerContext ctx, Object msg) {    // 写消息或传递给下一个处理器    ctx.write(msg);}

// 伪代码，展示了如何获取下一个处理器并调用它ChannelHandlerContext nextCtx = ctx.next();if (nextCtx != null) {    nextCtx.invokeChannelRead(msg);}

pipeline.addLast("newHandler", new AnotherChannelHandler());pipeline.remove(ctx.handler());

public void channelInactive(ChannelHandlerContext ctx) {    // 清理逻辑}

try {    // 可能抛出异常的操作} catch (Exception e) {    ctx.fireExceptionCaught(e);}