Netty学习(三）-Netty重要接口讲解

上一节我们写了一个HelloWorld，对于Netty的运行有了一定的了解，知道Netty是如何启动客户端和服务器端。这一节我们简要的讲解一下几个重要的接口，初步探讨Netty的运行机制，当然刚学Netty就深入原理肯定是很枯燥的，所以我们就点到为止。

1. ChannelPipeLine和ChannelHandler

在上一篇中我们在ChannelInitializer类的initChannel方法中使用了ChannelPipeline，然后在ChannelPipeline中使用了handler来处理业务逻辑。

ChannelPipeline是ChannelHandler的容器，它负责ChannelHandler的管理和事件拦截与调度。Netty的ChannelPipeline和ChannelHandler机制类似于Servlet 和Filter 过滤器，这类拦截器实际上是职责链模式的一种变形，主要是为了方便事件的拦截和用户业务逻辑的定制。

Netty的channel运用机制和Filter过滤器机制一样，它将Channel 的数据管道抽象为ChannelPipeline. 消息在ChannelPipeline中流动和传递。ChannelPipeline 持有I/O事件拦截器ChannelHandler 的链表，由ChannelHandler 对I/0 事件进行拦截和处理，可以方便地通过新增和删除ChannelHandler 来实现小同的业务逻辑定制，不需要对已有的ChannelHandler进行修改，能够实现对修改封闭和对扩展的支持。

通过一张图我们来看一下他们之间的关系：

一个Channel中包含一个ChannelPipeline，用来处理Channel中的事件，一个ChannelPipeline中可以包含很多个handler，第二节的示例代码中我们也看到了，使用各种handler来处理通信信息。

同时我们也注意到在hadler中继承了ChannelInboundHandlerAdapter类并实现了他的一些方法，比如：channelRead，channelActive，channelInactive等等，我们看到这些方法中都有一个参数：ChannelHandlerContext ctx。这个ChannelHandlerContext就是handler的上下文对象，有了这个ChannelHandlerContext你就获得了一切，你可以获得通道，获得事件的控制权。

事实上，用户不需要自己创建pipeline，因为使用ServerBootstrap 或者Bootstrap 启动

服务端或者客户端时， Netty 会为每个Channel 连接创建一个独立的pipeline。

ChannelPipeline pipeline = socketChannel.pipeline();

pipeline.addLast("framer", new DelimiterBasedFrameDecoder(8192,Delimiters.lineDelimiter()));

pipeline.addLast("decoder", new StringDecoder());

pipeline.addLast("encoder", new StringEncoder());

// 客户端的逻辑

pipeline.addLast("handler", new HelloWorldClientHandler());

ChannelPipeline 是线程安全的， 这意味着N个业务线程可以并发地操作ChannelPipeline

而不存在多线程并发问题。但是，ChannelHandler却不是线程安全的，这意味着尽管

ChannelPipeline 是线程去全的， 但是仍然需要自己保证ChannelHandler的线程安全。

Netty 中的事件分为inbound 事件和outbound 事件。inbound 事件通常由I/O线程触发，例如TCP 链路建立事件、链路关闭事件、读事件、异常通知事件等。Outbound 事件通常是I/O 用户主动发起的网络I/O 操作，例如用户发起的连接操作、绑定操作、消息发送等操作。

我们常用的inbound事件有：

• ChannelHandlerContext fireChannelRegistered() //channel注册事件
• ChannelHandlerContext fireChannelActive() //channel激活事件
• ChannelHandlerContext fireExceptionCaught(Throwable var1) //channel异常处理事件
• ChannelHandlerContext fireUserEventTriggered(Object var1) //用户自定义事件
• ChannelHandlerContext fireChannelRead(Object var1) //读事件

pipeline 中以fireXXX命名的方法都是从I/O 线程流向用户业务Handler的inbound 事件，它们的实现因功能而异，但是处理步骤类似：

1. 调用HeadHandler对应的fireXXX 方法

2. 执行事件相关的逻辑操作

常用的outbound事件有：

• ChannelFuture bind(SocketAddress var1, ChannelPromise var2) //绑定地址
• ChannelFuture connect(SocketAddress var1, ChannelPromise var2) //连接服务器
• ChannelFuture write(Object var1) //发送事件
• ChannelHandlerContext flush() //刷新事件

上面我们说到事件，netty的事件机制是由前至后的，一般来说，都是一个channel的ChannnelActive方法中调用fireChannelActive来触发调用下一个handler中的ChannelActive方法，即你在ChannelPipeline中添加handler的时候，要在第一个handler的channelActive方法中调用fireChannelActive，以此来触发下一个事件。我们再来写一个案例说明一下：

客户端：

public class HWClient {

    private  int port;

    private  String address;

    public HWClient(int port, String address) {

        this.port = port;

        this.address = address;

    }

    public void start(){

        EventLoopGroup group = new NioEventLoopGroup();

        Bootstrap bootstrap = new Bootstrap();

        bootstrap.group(group)

                .channel(NioSocketChannel.class)

                .handler(new ClientChannelInitializer());

        try {

            ChannelFuture future = bootstrap.connect(address,port).sync();

            future.channel().writeAndFlush("Hello Netty Server ,I am a common client");

            future.channel().closeFuture().sync();

        } catch (Exception e) {

            e.printStackTrace();

        }finally {

            group.shutdownGracefully();

        }

    }

    public static void main(String[] args) {

        HWClient client = new HWClient(7788,"127.0.0.1");

        client.start();

    }

}

客户端ClientChannelInitializer：

public class ClientChannelInitializer extends  ChannelInitializer<SocketChannel> {

    protected void initChannel(SocketChannel socketChannel) throws Exception {

        ChannelPipeline pipeline = socketChannel.pipeline();

        pipeline.addLast("decoder", new StringDecoder());

        pipeline.addLast("encoder", new StringEncoder());

        // 客户端的handler

        //先调用handler在ChannnelActive方法中调用fireChannelActive会激活handler1

        pipeline.addLast("handler", new HWClientHandler());

        pipeline.addLast("handler1", new BaseClientHandler());

    }

}

客户端handler：

public class HWClientHandler extends ChannelInboundHandlerAdapter {

    @Override

    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

        System.out.println("server say : "+msg.toString());

    }

    @Override

    public void channelActive(ChannelHandlerContext ctx) throws Exception {

        System.out.println("Handler1");

        ctx.fireChannelActive();

    }

    @Override

    public void channelInactive(ChannelHandlerContext ctx) throws Exception {

        System.out.println("Client is close");

    }

}

客户端的第二个handler：

public class BaseClientHandler extends ChannelInboundHandlerAdapter {

    @Override

    public void channelActive(ChannelHandlerContext ctx) throws Exception {

        System.out.println("Handler2");

    }

    @Override

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {

        ctx.close();

    }

}

服务端：

public class HWServer {

    private int port;

    public HWServer(int port) {

        this.port = port;

    }

    public void start(){

        EventLoopGroup bossGroup = new NioEventLoopGroup();

        EventLoopGroup workGroup = new NioEventLoopGroup();

        ServerBootstrap server = new ServerBootstrap().group(bossGroup,workGroup)

                                    .channel(NioServerSocketChannel.class)

                                    .childHandler(new ServerChannelInitializer());

        try {

            ChannelFuture future = server.bind(port).sync();

            future.channel().closeFuture().sync();

        } catch (InterruptedException e) {

            e.printStackTrace();

        }finally {

            bossGroup.shutdownGracefully();

            workGroup.shutdownGracefully();

        }

    }

    public static void main(String[] args) {

        HWServer server = new HWServer(7788);

        server.start();

    }

}

服务端ServerChannelInitializer:

public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {

    @Override

    protected void initChannel(SocketChannel socketChannel) throws Exception {

        ChannelPipeline pipeline = socketChannel.pipeline();

        // 字符串解码 和 编码

        pipeline.addLast("decoder", new StringDecoder());

        pipeline.addLast("encoder", new StringEncoder());

        // 自己的逻辑Handler

        pipeline.addLast("handler", new HWServerHandler());

    }

}

服务端handler：

public class HWServerHandler extends ChannelInboundHandlerAdapter {

    @Override

    public void channelActive(ChannelHandlerContext ctx) throws Exception {

        System.out.println("channelActive");

    }

    @Override

    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

        System.out.println(ctx.channel().remoteAddress()+"===>server: "+msg.toString());

        ctx.write("received your msg");

        ctx.flush();

    }

    @Override

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {

        super.exceptionCaught(ctx, cause);

        ctx.close();

    }

}

我们启动服务端和客户端，会发现客户端的两个handler都通过了。

先调用HWClientHandler，打印出：HWClientHandler channelActive；继而调用了BaseClientHandler ，打印出：BaseClient1Handler channelActive.