netty初探(1)

参考目录：

1. user-guide : http://netty.io/wiki/user-guide-for-4.x.html

2. demo:　http://netty.io/wiki/

3. 使用指南： http://udn.yyuap.com/doc/netty-4-user-guide/

下一篇 netty(2)

一、Netty介绍

The Netty project is an effort to provide an asynchronous event-driven network application framework and tooling for the rapid development of maintainable high-performance · high-scalability protocol servers and clients.

In other words, Netty is an NIO client server framework which enables quick and easy development of network applications such as protocol servers and clients. It greatly simplifies and streamlines network programming such as TCP and UDP socket server development.

'Quick and easy' does not mean that a resulting application will suffer from a maintainability or a performance issue. Netty has been designed carefully with the experiences earned from the implementation of a lot of protocols such as FTP, SMTP, HTTP, and various binary and text-based legacy protocols. As a result, Netty has succeeded to find a way to achieve ease of development, performance, stability, and flexibility without a compromise(妥协).

Some users might already have found other network application framework that claims to have the same advantage, and you might want to ask what makes Netty so different from them. The answer is the philosophy(哲学) it is built on. Netty is designed to give you the most comfortable experience both in terms of the API and the implementation from the day one. It is not something tangible but you will realize that this philosophy will make your life much easier as you read this guide and play with Netty.

二、Hello World - Discard Server

netty官网给的例子是一个discard server，用于理解netty的大致流程...

首先，从一个handler开始，handler用来处理接收之后的I/O事件

import io.netty.buffer.ByteBuf;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

/**
 * 处理服务端 channel.
 */
public class DiscardServerHandler extends ChannelInboundHandlerAdapter { // (1)

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) { // (2)
        // 默默地丢弃收到的数据
        ((ByteBuf) msg).release(); // (3)
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // (4)
        // 当出现异常就关闭连接
        cause.printStackTrace();
        ctx.close();
    }
}

说明：

　1.  ChannelInboundHandlerAdapter implements ChannelInboundHandler， 默认提供了一些方法用于事件处理

　2. 这里仅仅覆盖了channelRead() 事件处理方法。每当从客户端接收到新的数据时，这个方法在收到消息的时候被调用

　3. 这里收到的消息是ByteBuf，这是一个引用计数对象，这个对象必须显示的调用release()方法来释放。处理器的职责就是释放所有传递到处理器的引用。

 通常的channelRead()方法是以下模式:

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        try {
            // Do something with msg
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }

   4. exceptionCaught()事件处理方法是当出现Throwable对象才会被调用，例如Netty出现IO错误或者由于处理器处理事件出现异常时。

我们已经完成了DiscardServer的大部分功能，只需要如何处理请求的部分即可

import io.netty.bootstrap.ServerBootstrap;

import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

/**
 * 丢弃任何进入的数据
 */
public class DiscardServer {

    private int port;

    public DiscardServer(int port) {
        this.port = port;
    }

    public void run() throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(); // (1)
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap(); // (2)
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class) // (3)
             .childHandler(new ChannelInitializer<SocketChannel>() { // (4)
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ch.pipeline().addLast(new DiscardServerHandler());
                 }
             })
             .option(ChannelOption.SO_BACKLOG, 128)          // (5)
             .childOption(ChannelOption.SO_KEEPALIVE, true); // (6)

            // 绑定端口，开始接收进来的连接
            ChannelFuture f = b.bind(port).sync(); // (7)

            // 等待服务器  socket 关闭 。
            // 在这个例子中，这不会发生，但你可以优雅地关闭你的服务器。
            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        int port;
        if (args.length > 0) {
            port = Integer.parseInt(args[0]);
        } else {
            port = 8080;
        }
        new DiscardServer(port).run();
    }
}

说明:

(1) NioEventLoopGroup使用来处理I/O事件的多线程事件循环器，Netty提供了许多不同的EventLoopGroup来实现不同的传输。这里使用了2个NioEventLoopGroup。第一个是"Boss"，用来接收进来的连接。第二个"worker"，用来处理被接收的连接，一旦"boss"接收到连接，就会把信息注册到"worker"上。如何知道多少个线程已经被使用，如何映射到已经创建的Channel上都需要依赖于EventLoopGroup的实现，并且可以通过构造函数来配置它们之间的关系。

(2) ServerBootstrap是一个启动NIO服务的辅助启动类。你可以在服务中直接使用Channel，但是这会是一个复杂的处理过程，很多情况不用这么做

(3) NioServerSocketChannel.class 用来指用哪个类处理被接收的SocketChannel

(4) 这里的事件处理经常会被用来处理一个最近的已经接收的Channel。ChannelIntializer是一个特殊的处理类，用来帮助使用者配置一个新的Channel，也许你想通过增加一些处理类比如DiscardServerHandler 来配置一个新的 Channel 或者其对应的ChannelPipeline 来实现你的网络程序。当你的程序变的复杂时，可能你会增加更多的处理类到 pipline 上，然后提取这些匿名类到最顶层的类上。

(5) 你可以设置这里指定的 Channel 实现的配置参数。我们正在写一个TCP/IP 的服务端，因此我们被允许设置 socket 的参数选项比如tcpNoDelay 和 keepAlive。请参考 ChannelOption 和详细的 ChannelConfig 实现的接口文档以此可以对ChannelOption 的有一个大概的认识。

(6) 你关注过 option() 和 childOption() 吗？option() 是提供给NioServerSocketChannel 用来接收进来的连接。childOption() 是提供给由父管道 ServerChannel 接收到的连接，在这个例子中也是 NioServerSocketChannel。

(7) 我们继续，剩下的就是绑定端口然后启动服务。这里我们在机器上绑定了机器所有网卡上的 8080 端口。当然现在你可以多次调用 bind() 方法(基于不同绑定地址)。

三、Time Server

在这个部分被实现的协议是 TIME 协议。和之前的例子不同的是在不接受任何请求时他会发送一个含32位的整数的消息，并且一旦消息发送就会立即关闭连接。在这个例子中，你会学习到如何构建和发送一个消息，然后在完成时关闭连接。

在这里，我们会忽略所有接收的数据，因此，我们不需要read request这个步骤，所以不使用channelRead()方法，代替使用channelActive()方法，在连接被建立并且准备进行通信时调用。

public class TimeServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelActive(final ChannelHandlerContext ctx) { // (1)
        final ByteBuf time = ctx.alloc().buffer(4); // (2)
        time.writeInt((int) (System.currentTimeMillis() / 1000L + 2208988800L));

        final ChannelFuture f = ctx.writeAndFlush(time); // (3)
        f.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) {
                assert f == future;
                ctx.close();
            }
        }); // (4)
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

说明：

　　(1) channelActive()方法会在连接被建立并且进行准备通信的时候调用。

　　(2) 我们需要写入一个32位的证书，因此我们需要至少一个4个字节的ByteBuf。

　　(3) netty中的ByteBuf有2个指针，跟ByteBuffer不同，不需要flip()

　　(4) ChannelFuture代表一个还没有发生的I/O操作。这意味着任何一个请求操作都不会马上执行，因为在Netty所有的代码都是异步的。要在ChannelFuture完成之后添加关闭Channel操作，需要使用Listener关闭，甚至于可以简单的加上

f.addListener(ChannelFutureListener.CLOSE);

3.1 Writing a Time Client

public class TimeClient {

    public static void main(String[] args) throws Exception {

        String host = args[0];
        int port = Integer.parseInt(args[1]);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            Bootstrap b = new Bootstrap(); // (1)
            b.group(workerGroup); // (2)
            b.channel(NioSocketChannel.class); // (3)
            b.option(ChannelOption.SO_KEEPALIVE, true); // (4)
            b.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new TimeClientHandler());
                }
            });

            // 启动客户端
            ChannelFuture f = b.connect(host, port).sync(); // (5)

            // 等待连接关闭
            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
        }
    }
}

(1) BootStrap和ServerBootstrap类似... 客户端或者无连接传输模式用

(2) 如果你指定了一个EventLoopGroup，那他就会作为一个boss group，也会作为一个worker group，尽管客户端不需要使用到boss worker.

(3) NioSocketChannel.class 客户端使用的socket

(4) 没有childoption，因为是客户端，只有option

(5) 用connect()方法连接服务器

实现Client Handler

import java.util.Date;

public class TimeClientHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ByteBuf m = (ByteBuf) msg; // (1)
        try {
            long currentTimeMillis = (m.readUnsignedInt() - 2208988800L) * 1000L;
            System.out.println(new Date(currentTimeMillis));
            ctx.close();
        } finally {
            m.release();
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

(1)  Client端读到的也是ByteBuf

(2) Client处理异常的方式和Server端类似

3.2 Dealing with a Stream-based Transport

In a stream-based transport such as TCP/IP, received data is stored into a socket receive buffer. Unfortunately, the buffer of a stream-based transport is not a queue of packets but a queue of bytes. It means, even if you sent two messages as two independent packets, an operating system will not treat them as two messages but as just a bunch of bytes. Therefore, there is no guarantee that what you read is exactly what your remote peer wrote. For example, let us assume that the TCP/IP stack of an operating system has received three packets:

Because of this general property of a stream-based protocol, there's high chance of reading them in the following fragmented form in your application:

Therefore, a receiving part, regardless it is server-side or client-side, should defrag the received data into one or more meaningful frames that could be easily understood by the application logic. In case of the example above, the received data should be framed like the following:

方案一：

最简单的方案是构造一个内部的可积累的缓冲，直到4个字节全部接收到了内部缓冲。下面的代码修改了 TimeClientHandler 的实现类修复了这个问题

import java.util.Date;

public class TimeClientHandler extends ChannelInboundHandlerAdapter {
    private ByteBuf buf;

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        buf = ctx.alloc().buffer(4); // (1)
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        buf.release(); // (1)
        buf = null;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ByteBuf m = (ByteBuf) msg;
        buf.writeBytes(m); // (2)
        m.release();

        if (buf.readableBytes() >= 4) { // (3)
            long currentTimeMillis = (buf.readUnsignedInt() - 2208988800L) * 1000L;
            System.out.println(new Date(currentTimeMillis));
            ctx.close();
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

方案二：

尽管第一个解决方案已经解决了 TIME 客户端的问题了，但是修改后的处理器看起来不那么的简洁，想象一下如果由多个字段比如可变长度的字段组成的更为复杂的协议时，你的 ChannelInboundHandler 的实现将很快地变得难以维护。

正如你所知的，你可以增加多个 ChannelHandler 到ChannelPipeline ,因此你可以把一整个ChannelHandler 拆分成多个模块以减少应用的复杂程度，比如你可以把TimeClientHandler 拆分成2个处理器：

• TimeDecoder 处理数据拆分的问题
• TimeClientHandler 原始版本的实现

幸运地是，Netty 提供了一个可扩展的类，帮你完成 TimeDecoder 的开发。

public class TimeDecoder extends ByteToMessageDecoder { // (1)
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) { // (2)
        if (in.readableBytes() < 4) {
            return; // (3)
        }

        out.add(in.readBytes(4)); // (4)
    }
}

(1) ByteToMessageDecoder是一个ChannelInboundHandler的实现类，用于处理数据分析问题

(2) 每当有新数据接收，ByteToMessageDecorder会调用decode()方法来处理内部的那个累积缓冲。

(3) Decode()方法可以决定当累积缓冲里没有足够数据时可以往out对象里放任意多数据。当有更多数据被接收了，会再次调用decode()方法

(4) 如果在decode()方法中往out中添加内容，则意味解码器解码消息成功

So，现在只需要往ChannelPipeline中添加解码器即可

b.handler(new ChannelInitializer<SocketChannel>() {
    @Override
    public void initChannel(SocketChannel ch) throws Exception {
        ch.pipeline().addLast(new TimeDecoder(), new TimeClientHandler());
    }
});

四、架构总览

以下内容摘自于 http://udn.yyuap.com/doc/netty-4-user-guide/Architectural%20Overview/Architectural%20Overview.html

 4.1 ByteBuf

Netty 使用自建的 buffer API，而不是使用 NIO 的 ByteBuffer 来表示一个连续的字节序列。与 ByteBuffer 相比这种方式拥有明显的优势。Netty 使用新的 buffer 类型 ByteBuf，被设计为一个可从底层解决 ByteBuffer 问题，并可满足日常网络应用开发需要的缓冲类型。这些很酷的特性包括：

• 如果需要，允许使用自定义的缓冲类型。
• 复合缓冲类型中内置的透明的零拷贝实现。
• 开箱即用的动态缓冲类型，具有像 StringBuffer 一样的动态缓冲能力。
• 不再需要调用的flip()方法。
• 正常情况下具有比 ByteBuffer 更快的响应速度。

Transparent Zero Copy 透明的零拷贝

举一个网络应用到极致的表现，你需要减少内存拷贝操作次数。你可能有一组缓冲区可以被组合以形成一个完整的消息。网络提供了一种复合缓冲，允许你从现有的任意数的缓冲区创建一个新的缓冲区而无需没有内存拷贝。例如，一个信息可以由两部分组成；header 和 body。在一个模块化的应用，当消息发送出去时，这两部分可以由不同的模块生产和装配。

<pre> +--------+----------+
 | header |   body   |
 +--------+----------+
 </pre>

如果你使用的是 ByteBuffer ，你必须要创建一个新的大缓存区用来拷贝这两部分到这个新缓存区中。或者，你可以在 NiO做一个收集写操作，但限制你将复合缓冲类型作为 ByteBuffer 的数组而不是一个单一的缓冲区，打破了抽象，并且引入了复杂的状态管理。此外，如果你不从 NIO channel 读或写，它是没有用的。

    // 复合类型与组件类型不兼容。
    ByteBuffer[] message = new ByteBuffer[] { header, body };

通过对比， ByteBuf 不会有警告，因为它是完全可扩展并有一个内置的复合缓冲区

　　// 复合类型与组件类型是兼容的。
    ByteBuf message = Unpooled.wrappedBuffer(header, body);

    // 因此，你甚至可以通过混合复合类型与普通缓冲区来创建一个复合类型。
    ByteBuf messageWithFooter = Unpooled.wrappedBuffer(message, footer);

    // 由于复合类型仍是 ByteBuf，访问其内容很容易，
    //并且访问方法的行为就像是访问一个单独的缓冲区，
    //即使你想访问的区域是跨多个组件。
    //这里的无符号整数读取位于 body 和 footer
    messageWithFooter.getUnsignedInt(
         messageWithFooter.readableBytes() - footer.readableBytes() - 1);

Automatic Capacity Extension 自动容量扩展

许多协议定义可变长度的消息，这意味着没有办法确定消息的长度，直到你构建的消息。或者，在计算长度的精确值时，带来了困难和不便。这就像当你建立一个字符串。你经常估计得到的字符串的长度，让 StringBuffer 扩大了其本身的需求。

    // 一种新的动态缓冲区被创建。在内部，实际缓冲区是被“懒”创建，从而避免潜在的浪费内存空间。
    ByteBuf b = Unpooled.buffer(4);

    // 当第一个执行写尝试，内部指定初始容量 4 的缓冲区被创建
    b.writeByte('1');

    b.writeByte('2');
    b.writeByte('3');
    b.writeByte('4');

    // 当写入的字节数超过初始容量 4 时，
    //内部缓冲区自动分配具有较大的容量
    b.writeByte('5');

Better Performance 更好的性能

最频繁使用的缓冲区 ByteBuf 的实现是一个非常薄的字节数组包装器（比如，一个字节）。与 ByteBuffer 不同，它没有复杂的边界和索引检查补偿，因此对于 JVM 优化缓冲区的访问更加简单。更多复杂的缓冲区实现是用于拆分或者组合缓存，并且比 ByteBuffer 拥有更好的性能。

4.2 统一的IO API

传统的 Java I/O API 在应对不同的传输协议时需要使用不同的类型和方法。例如：java.net.Socket 和 java.net.DatagramSocket 它们并不具有相同的超类型，因此，这就需要使用不同的调用方式执行 socket 操作。

这种模式上的不匹配使得在更换一个网络应用的传输协议时变得繁杂和困难。由于（Java I/O API）缺乏协议间的移植性，当你试图在不修改网络传输层的前提下增加多种协议的支持，这时便会产生问题。并且理论上讲，多种应用层协议可运行在多种传输层协议之上例如TCP/IP,UDP/IP,SCTP和串口通信。

让这种情况变得更糟的是，Java 新的 I/O（NIO）API与原有的阻塞式的I/O（OIO）API 并不兼容，NIO.2(AIO)也是如此。由于所有的API无论是在其设计上还是性能上的特性都与彼此不同，在进入开发阶段，你常常会被迫的选择一种你需要的API。

例如，在用户数较小的时候你可能会选择使用传统的 OIO(Old I/O) API，毕竟与 NIO 相比使用 OIO 将更加容易一些。然而，当你的业务呈指数增长并且服务器需要同时处理成千上万的客户连接时你便会遇到问题。这种情况下你可能会尝试使用 NIO，但是复杂的 NIO Selector 编程接口又会耗费你大量时间并最终会阻碍你的快速开发。

Netty 有一个叫做 Channel 的统一的异步 I/O 编程接口，这个编程接口抽象了所有点对点的通信操作。也就是说，如果你的应用是基于 Netty 的某一种传输实现，那么同样的，你的应用也可以运行在 Netty 的另一种传输实现上。Netty 提供了几种拥有相同编程接口的基本传输实现：

• 基于 NIO 的 TCP/IP 传输 (见 io.netty.channel.nio),
• 基于 OIO 的 TCP/IP 传输 (见 io.netty.channel.oio),
• 基于 OIO 的 UDP/IP 传输, 和
• 本地传输 (见 io.netty.channel.local).

切换不同的传输实现通常只需对代码进行几行的修改调整，例如选择一个不同的 ChannelFactory 实现。

此外，你甚至可以利用新的传输实现没有写入的优势，只需替换一些构造器的调用方法即可，例如串口通信。而且由于核心 API 具有高度的可扩展性，你还可以完成自己的传输实现。

4.3 基于拦截链模式的事件模型

一个定义良好并具有扩展能力的事件模型是事件驱动开发的必要条件。Netty 具有定义良好的 I/O 事件模型。由于严格的层次结构区分了不同的事件类型，因此 Netty 也允许你在不破坏现有代码的情况下实现自己的事件类型。这是与其他框架相比另一个不同的地方。很多 NIO 框架没有或者仅有有限的事件模型概念；在你试图添加一个新的事件类型的时候常常需要修改已有的代码，或者根本就不允许你进行这种扩展。

在一个 ChannelPipeline 内部一个 [ChannelEvent]() 被一组ChannelHandler 处理。这个管道是 Intercepting Filter (拦截过滤器)模式的一种高级形式的实现，因此对于一个事件如何被处理以及管道内部处理器间的交互过程，你都将拥有绝对的控制力。例如，你可以定义一个从 socket 读取到数据后的操作：

    public class MyReadHandler implements SimpleChannelHandler {
         public void messageReceived(ChannelHandlerContext ctx, MessageEvent evt) {
             Object message = evt.getMessage();
             // Do something with the received message.
                ...

             // And forward the event to the next handler.
             ctx.sendUpstream(evt);
        }
    }

同时你也可以定义一种操作响应其他处理器的写操作请求：

    public class MyWriteHandler implements SimpleChannelHandler {
        public void writeRequested(ChannelHandlerContext ctx, MessageEvent evt) {
            Object message = evt.getMessage();
            // Do something with the message to be written.
                ...

            // And forward the event to the next handler.
            ctx.sendDownstream(evt);
        }
    }