Netty 4.X 用户指南

package io.netty.example.discard;

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;

/**
 * Discards any incoming data.
 */
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)

            // Bind and start to accept incoming connections.
            ChannelFuture f = b.bind(port).sync(); // (7)

            // Wait until the server socket is closed.
            // In this example, this does not happen, but you can do that to gracefully
            // shut down your server.
            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操作的一种多线程event循环。Netty为不同种传输提供了多种EventLoopGroup 的实现。我们在这个例子中实现的是sever端的应用，所以两个NioEventLoopGroup会被用到。第一个，常常称之为‘boss’，接受进来的连接。第二个，常称之为‘woker’，控制着接收到的连接的通行，一旦boss接受了连接并将注册接受到的连接给woker。有多少线程被使用以及他们如何映射到channel 取决于EventLoopGroup的实现并且甚至于通过构造函数实现可配置。
2. ServerBootstrap 是一个建立server的帮助类。你可以用Channel接口直接建立。无论如何，请注意这是一个繁杂的过程，尽可能不要这样做。（作者的意思是：封装成工具类的方式会更好）
3. 这里，我们指定了NioServerSocketChannel 类用于实例化一个新的Channel接收进来的连接。
4. 这里指定的处理器将一直被新接受的Channel所评估。ChannelInitializer 专门用于帮助用户配置一个新的Channel。这就像你想为一个新的Channel配置ChangePipeline通过添加一些像：DiscardServerHandler的处理器去实现你的网络应用一样。随着应用程序变得复杂,很可能你会更多的处理程序添加到管道并提取这最终匿名类到顶级类. 
5. 你也可以设置这些参数到Channel的实现中。我们要协议个TCP/IP server，所以我们允许设置socket选项，例如：tcpNoDelay和keepAlive。请参见ChannelOption的API文档针对于ChannelConfig的ChannelOption配置。
6. 你是否注意到option()和childoption()？option()用于NioServerSocketChannel接受进入的连接。childoption()用于父类ServerChannel接受的，这指的是NioServerSockethannel.
7. 我们可以准备好开始了。剩下的就是绑定端口并启动 server。这里我们绑定的网卡端口是8080。现在你可以根据你需要多次调用bind()方法(但是需要绑定不同的地址)。

package io.netty.example.time;

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. 为发送一条新的消息，我们需要分配一段新的buffer用于容纳这个消息。我们要写一个32为的整数，因此我们需要一个至少容量为4byte的ByteBuf。可以通过ChannelHandlerContext.alloc()获取当前ByteBufAllocator和分配一段新的Buffer。
3. 像以前一样，我们写构造信息。
   但是,等等,flip在哪里?在NIO中发送消息之前，我们不是应该调用java.nio.ByteBuffer.flip()吗?ByteBuf没有这样的方法，因为它有两个指示器；一个是读操作，另一个是写操作。读的索引不改变时，当你想ByteBuf写一些东西时写的索引会不断增长。读的位置和写的位置分别代表消息的开始和结束。
   
   相比之下，在不调用flip方法情况下，NIO buffer 没有提供一个清楚的方法支出消息内容的开始和结束。当你忘记反转缓冲区时你会遇到麻烦，因为没有或者错误的数据被发送。就像一个错误在NETTY中没有发生一样，因为我们对于不同的操作类型有不同的指示器。你将会发现自从你用了它之后你的生活变的越来越简单--人生没有反转出来。
   
   另一个需要指出的是，ChannelHandlerContext.write()（包括writeAndFlush()）方法返回一个ChannelFuture。ChannelFuture代表一个I/O操作尚未发生。它意味着，任何请求操作可能已经执行，因为在NETTY中所有的操作是异步的。例如，接下来的代码可能关闭了连接，尽管是在消息发送之前。
   

   Channel ch = ...;
   ch.writeAndFlush(message);
   ch.close();

   因此，你需要在channelFuture结束时调用close()方法，他通过write()方法返回，并且当写操作已经做时，他会通知它的监听者。请注意这里，close()方法也有可能立刻关闭连接，并将返回channelFuture。
4. 当一个写的请求完成时，接下来我们如何获取通知。这就和向返回的ChannelFuture中添加一个ChannelFutureListener一样简单。这里，当操作执行时，我们创建一个用于关闭Channel的新的匿名的ChannelFutureLiistener 。

f.addListener(ChannelFutureListener.CLOSE);

$ rdate -o <port> -p <host>

写一个计时客户端

package io.netty.example.time;

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());
                }
            });

            // Start the client.
            ChannelFuture f = b.connect(host, port).sync(); // (5)

            // Wait until the connection is closed.
            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
        }
    }
}

package io.netty.example.time;

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();
    }
}

package io.netty.example.time;

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.readInt() - 2208988800L) * 1000L;
            System.out.println(new Date(currentTimeMillis));
            ctx.close();
        }
    }

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

1. ChannelHandler 有两个生存周期监听方法：handlerAdded和handlerRemoved。您可以执行任意初始化任务,只要它不会阻碍很长一段时间。
2. 首先，所有接收到的数据必须累积到buf中。
3. 接着，handler必须检查，如果buf中与足够的数据，就像例子中buf可以容纳4byte数据一样，就要执行实际的业务逻辑。另外，当有更多的数据到达，Netty会再次调用channelREAD()方法，知道积累够4byte的数据。

第二种解决方案

package io.netty.example.time;

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)
    }
}

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

public class TimeDecoder extends ReplayingDecoder<Void> {
    @Override
    protected void decode(
            ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        out.add(in.readBytes(4));
    }
}

package io.netty.example.time;

import java.util.Date;

public class UnixTime {

    private final int value;

    public UnixTime() {
        this((int) (System.currentTimeMillis() / 1000L + 2208988800L));
    }

    public UnixTime(int value) {
        this.value = value;
    }

    public int value() {
        return value;
    }

    @Override
    public String toString() {
        return new Date((value() - 2208988800L) * 1000L).toString();
    }
}

@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
    if (in.readableBytes() < 4) {
        return;
    }

    out.add(new UnixTime(in.readInt()));
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    UnixTime m = (UnixTime) msg;
    System.out.println(m);
    ctx.close();
}

@Override
public void channelActive(ChannelHandlerContext ctx) {
    ChannelFuture f = ctx.writeAndFlush(new UnixTime());
    f.addListener(ChannelFutureListener.CLOSE);
}

package io.netty.example.time;

public class TimeEncoder extends ChannelOutboundHandlerAdapter {
    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
        UnixTime m = (UnixTime) msg;
        ByteBuf encoded = ctx.alloc().buffer(4);
        encoded.writeInt(m.value());
        ctx.write(encoded, promise); // (1)
    }
}

public class TimeEncoder extends MessageToByteEncoder<UnixTime> {
    @Override
    protected void encode(ChannelHandlerContext ctx, UnixTime msg, ByteBuf out) {
        out.writeInt(msg.value());
    }
}