Netty 4.1 Getting Start （翻译） + Demo

一、先来官方入门页面的翻译（翻译不好请多包涵）

 入门

本章以简单的例子来介绍Netty的核心概念，以便让您快速入门。当您阅读完本章之后，您就能立即在Netty的基础上写一个客户端和一个服务器。

如果您喜欢自上而下的方法来学习某些东西，那么就可能需要从第二章“架构概述”开始，然后回到这一章。

入门之前

运行本章的例子，最低要求有两个：最新版的Netty和JDK1.6+。其余的balabala...

写一个Discard Server（丢弃服务器）

世界上最简单的协议不是“Hello World！”，而是DISCARD（丢弃）。它是一种 丢弃任何收到的数据并且没有任何响应的协议。（说白了就是客户端来消息，服务器不鸟你）

要实现这种协议，你需要做的只有 忽略掉所有收到的数据 。让我们直接从Handler的实现开始，它处理由Netty生成的I/O事件。

package io.netty.example.discard;

import io.netty.buffer.ByteBuf;

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

/**
 * Handles a server-side channel.
 */
public class DiscardServerHandler extends ChannelInboundHandlerAdapter { // (1)

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) { // (2)
        // Discard the received data silently.
        ((ByteBuf) msg).release(); // (3)
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // (4)
        // Close the connection when an exception is raised.
        cause.printStackTrace();
        ctx.close();
    }
}

1. DiscardServerHandler 继承了 ChannelInboundHandlerAdapter,ChannelInboundHandlerAdapter实现了ChannelInboundHandler接口。

public class ChannelInboundHandlerAdapter extends ChannelHandlerAdapter implements ChannelInboundHandler

ChannelInboundHandler提供了很多你可以覆盖的事件处理方法。现在，你只需要继承ChannelInboundHandlerAdapter而不是实现ChannelInboundHandler接口。

2. 我们在这里覆盖了channelRead方法，每当从客户端传来新消息时，都会触发这个方法。在这里，负责接收消息的数据类型是ByteBuf。

3. 为了实现DISCARD协议，我们需要忽略收到的消息。ByteBuf是一个引用计数对象，需要明确的通过调用release()方法去释放。请记住，用户有责任释放传递给处理程序的引用计数对象。通常，channelRead()处理方法的实现方式如下：

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

4. 当Netty发生IO异常时，会调用exceptionCaught()。大多数情况下，发生的异常应该被日志记录，尽管这里的处理方式可能跟您的实际情况有所不同，但是相应的通道应该在这里关闭。比如：在关闭连接之前，您想响应一个错误代码

到目前为止我们完成了服务器的一半，接下来利用DiscardServerHandler来编写启动服务器的main方法。

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 是一个处理IO操作的多线程时间循环。Netty为了不同的传输协议提供了很多EventLoopGroup接口的实现。我们在这个例子里实现了一个服务端程序，所以我们用了两个NioEventLoopGroup。第一个，经常被称作“boss”，用来接收将要到来的连接。第二个，经常被称作“worker”，一旦Boss接收到连接并把连接注册到Worker，就开始处理接收到的连接的通信。使用多少线程以及如何将它们映射到创建的通道中取决于EventLoopGroup的具体实现，可以通过构造函数来配置。

2. ServerBootstrap作为一个辅助类去设置一个服务器。你可以直接用一个Channel来设置服务器。然而，请注意那是一个繁琐的过程，大多数情况下不需要你那样做。

3. 在这里，我们特别使用NioServerSocketChannel来实例化一个新通道去接受将要到来的连接。

4. 这里的处理程序始终被新接收的Channel评估。（这句话不知到咋解释了）。ChannelInitializer是一个特殊的处理程序，目的是帮助用户配置一个新Channel。您可能想要通过添加一个处理程序比如DiscardServerHandler来配置一个新管道的ChannelPipeline来实现你的网络程序。随着应用程序的复杂化，您可能会在管道中添加更多的处理程序，并将这个匿名类最终提取到顶级类中。

5. 您还可以在Channel的具体实现设置一些特殊的参数。我们现在在写一个TCP/IP的服务器，所以允许我们设置一些套接字的选项，例如：tcpNoDelay，keepAlive。请参考API文档中ChannelOption，特别是ChannelConfig的具体实现类，以便去获得关于ChannelOptions相关支持的概述。

6. 你注意到option()和childOption()了吗？option()是为了NioServerSocketChannel接受新的连接，childOption()是为了那些被父ServerChannel接受的Channels，在这里ServerChannel指的是NioServerSocketChannel。

7. 我们现在已经准备好了。最后剩下绑定端口和启动服务器。在这里我们绑定8080端口。你可以根据需要调用bind()方法多次。

现在你已经完成了第一台Netty服务器。

查看接收到的数据

现在我们已经写完了我们第一台服务器，现在需要验证它是否正常工作。这里有一个简单的方法，通过使用telnet命令。例如，你可以在命令行输入

telnet localhost 8080

但是，我们能看到服务器是否正常工作吗？我们不能知道，因为它是一台丢弃服务器。我们不会得到任何回应。为了证明他已经工作了，我们需要修改服务器，让服务器打印接收到的消息。

我们已经知道当有数据的时候就调用channelRead()方法，所以我们写一些代码在DiscardServerHandler中的channelRead方法里。

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    ByteBuf in = (ByteBuf) msg;
    try {
        while (in.isReadable()) { // (1)
            System.out.print((char) in.readByte());
            System.out.flush();
        }
    } finally {
        ReferenceCountUtil.release(msg); // (2)
    }
}

1. 实际上这个低效循环可以简化为：System.out.println(in.toString(io.netty.util.CharsetUtil.US_ASCII))

2. 或者，您可以在in.release()这里做。

如果你再次执行telnet 命令，你会看见服务器输出了接受到的数据。

写一个响应服务器

目前为止，我们只接受但是没有任何响应。一台服务器，通常应该响应该请求。让我们学习如何通过实现ECHO协议向客户端写入响应消息，其中任何接收到的数据都被发送回来。

与前面部分实现的丢弃服务器的唯一区别在于它将接收到的数据发回，而不是将接收的数据输出到控制台。因此，再次修改channelRead()方法是足够的：

@Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ctx.write(msg); // (1)
        ctx.flush(); // (2)
    }

1. ChannelHandlerConetxt提供了很多方法让你去触发IO时间或操作。这里我们调用write(object)来逐字的写入接受到的消息。注意，我们不像DISCARD例子里的那样，我们没有释放我们收到的消息。这是因为当它被写回到wire时，Netty替我们释放它。

2. ctx.write(Object)不会让消息发送，它存在于内部缓冲区，通过调用ctx.flush()来把消息发送出去，或者，您可以简洁的调用ctx.writeAndFlush(msg)。

如果您再次使用telnet命令，您就会发现你发送什么服务器返回什么。

 写一个时间服务器

 本节要实现的协议是Time协议。与前面的例子不同的是，它发送一个包含32位整数的消息，而不接收任何请求，并在发送消息后关闭连接。在此示例中，您将学习如何构建和发送消息，并在完成时关闭连接。

 因为我们要忽略任何接受到的数据并且当建立好连接就发送一条消息，这次就不能再使用channelRead()了，相反，我们使用channelActive()方法。下面是实现：

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() 方法，让我们在这个方法中写入一个32位整数来代表当前时间 

2. 为了发送一个新消息，我们需要开辟出一块缓冲区来存放我们的消息。我们将要写一个32位的整数，所以我们需要一个容量至少为4字节的ByteBuf。通过ChannelHandlerContext.alloc()获得当前的 ByteBufAllocator，并分配出一块缓冲区。

3. 像往常一样，我们写入包装好的消息。

等等，flip在哪？在NIO里发消息之前我们不调用一下java.nio.ByteBuffer.flip()吗？ByteBuf没有那个方法因为他有两个指针，一个为了读操作，一个为了写操作。当你写点什么的时候，写指针会向前移动而读指针不会移动。读指针和写指针分别代表消息的开始和结束位置。

相比之下，NIO缓冲区没有提供一种清晰地方式指出消息的开始和结束，除非调用flip方法。当你忘记flip的时候你会陷入麻烦，因为将要发送的内容为空或者是错误的数据。这样的错误不会发生在Netty，因为不同的操作类型我们有不同的指针。当你使用它的时候你会发现他让你的生活更容易，因为不用再考虑flip了。

另外一点需要注意的是：ChannelHandlerContext.write()（还有writeAndFlush()）方法返回一个ChannelFuture对象，一个ChannelFuture对象代表还没有发生的IO操作，这意味着任何请求的操作可能尚未执行，因为所有操作在Netty中都是异步的。比如，在消息发送之前，下面的代码也会关闭连接：

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

因此，当 ChannelFuture执行完成后你需要调用 close()方法， ChannelFuture是由write()方法返回的，并且他会告诉所有的监听者当write操作完成时。请注意，close()也可能不会立即关闭连接，并且他返回一个ChannelFuture。

4. 当请求完成的时候我们怎样得到通知呢？就像给返回的ChannelFuture添加ChannelFutureListener一样，在这里我们创建一个匿名内部类ChannelFutureListener，当操作完成时，我们关闭Channel

或者您可以的使用系统预定义的监听器来简化代码

f.addListener(ChannelFutureListener.CLOSE);

为了验证Time服务器是否成功，可以通过Unix命令rdate来检查：

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

port是main方法里面明确的，host一般是localhost。

写一个Time客户端

不像DISCARD和ECHO服务器那样，为了Time协议我们需要一个客户端，因为人类不能讲一个32位整数翻译成日历上表示的日期。在本节中，我们将讨论如何确定服务器正常运行以及如何写一个基于Netty的客户端。

基于Netty的服务器和客户端之间最大且唯一的不同就是：使用的Bootstrap和Channel得实现。请看：

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

 1. Bootstrap和ServerBootstrap非常相似，除了它是一个非服务器通道，比如：客户端或者无连接通道。

 2. 如果你只指定一个EventLoopGroup，他将同时用于boss组和worker组。尽管客户端不使用boos 的 worker

 3.不同于NioServerSocketChannel，NioSocketChannel被用来创建客户端Channel。

 4.请注意在这里我们没有用childOption()，不像我们在ServerBootstrap那样，因为在客户端的SocketChannel没有Parent。

 5.  我们需要调用 connect()方法，而不是bind()方法。

正如你看见的，和服务器的代码没什么区别。怎样实现一个ChannelHandler？他应该接受一个来自服务器的32位的整数，并将它解释为人类可以看得懂的形式，打印出翻译过的时间，并关闭连接。

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

1.在TCP / IP中，Netty将从对等体发送的数据读入ByteBuf。

 它看起来很简单，并且和服务器代码没啥两样。然而，这个处理程序有时会拒绝服务并抛出异常IndexOutOfBoundsException。下一节我们将讨论这个问题。

处理基于流的运输

套接字缓冲区的一个小小注意事项

 在诸如TCP / IP的基于流的传输中，接收的数据被存储到套接字接收缓冲器中。不幸的是，基于流的传输的缓冲区不是数据包的队列，而是字节队列。这意味着，即使您发送两个消息作为两个独立数据包，操作系统也不会将它们视为两个消息，而只是一堆字节。因此，您无法保证您所读取的内容正是您远程对等人写的内容。例如，假设操作系统的TCP / IP堆栈已经收到三个数据包：

由于基于流的协议的这种一般属性，在应用程序中以下列分片形式读取它们的可能性很大：

因此，无论服务器端或客户端如何，接收部分都应将接收的数据进行碎片整理，整理成一个一个有意义的结构以便于很容易的被程序逻辑所理解。在上述示例的情况下，接收到的数据应该如下所示：

第一个解决方案

现在让我们回到之前的Time客户端例子。

我们在这里也有同样的问题。一个32位整数是非常少量的数据，它不太可能经常碎片化。然而，问题是可以分散，碎片化的可能性会随着流量的增加而增加。

简单的解决方案是创建内部累积缓冲区，并等待所有4个字节都被接收到内部缓冲区。以下是TimeClientHandler修复的问题修复实现：

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.readUnsignedInt() - 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. 然后，处理程序会检查buf是否含有足够的数据，在这个例子里是4个字节，然后继续执行现有的业务逻辑。否则，当更多的数据到来时Netty将再次调用channelRead()方法，最终将有4个字节的数据将被积累。

第二个解决方案

尽管第一个解决方案解决了问题，但是修改后的处理程序看起来并不是很干净。想象一下有一个由多个字段组成的更复杂的协议，例如可变长度字段，你的 ChannelInboundHandler 实现马上变的力不从心。


你可能已经注意到，您可以添加多个ChannelHandler到ChannelPipeline上。因此，您可以将一个庞大的ChannelHandler分割成许多小模块，以减少应用程序的复杂性。例如，您可以分割 TimeClientHandler 成两个处理程序：

TimeDecoder 解决碎片问题，和

TimeClientHandler的简单初始程序

幸运的是，Netty提供了一个可扩展的类，可帮助您编写开箱即用的第一个：

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

 1. ByteToMessageDecoder 是 ChannelInboundHandler 接口的实现，让处理碎片问题更容易。

2. 无论什么时候有新数据到来时，ByteToMessageDecoder 利用内部维护的一个缓冲区调用decode()方法。

3. 当累积的缓冲区没有足够的数据时，decode()能够决定不向out添加数据。当有更多的数据到来时，ByteToMessageDecoder 能够再次调用decode()方法。

4. 如果decode()添加了一个对象到out，那就意味着解码器成功的解码了一条消息。ByteToMessageDecoder 将丢弃累积缓冲区的读部分，请记住你不需要解码更多的消息。ByteToMessageDecoder 会保持调用decode()直到不再向out添加数据。

现在我们有另外一个处理程序要添加到 ChannelPipeline，我们需要修改TimeClient中 ChannelInitializer 的实现部分：

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

如果你是一个大胆的人，你可能想要尝试更简化的解码器ReplayingDecoder 。你需要查阅API来获取更多的信息。

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

此外，Netty提供了开箱即用的解码器，使您能够轻松实现大多数协议，并帮助您避免实现一个庞大的不可维护的处理程序。有关更多详细示例，请参阅以下软件包：

io.netty.example.factorial 用于二进制协议
io.netty.example.telnet    用于文本行协议。

 使用POJO而不是ByteBuf

 到目前为止我们见过的所有例子都是使用ByteBuf作为协议消息的数据结构。在本节中，我们将改进TIME协议客户端和服务器示例，使用POJO而不是ByteBuf。

在你的ChannelHandlers中使用POJO的优势是明显的，你的处理程序变得可维护，通过分离出提取ByteBuf中的信息的代码使得处理程序变得可重用。在TIME客户端和服务器示例中，我们只读取一个32位整数，它并不是我们直接使用ByteBuf面临的主要问题。然而，你会发现在真实世界协议中把它分离是很有必要的。

首先我们定义一个新类：UnixTime。

package io.netty.example.time;

import java.util.Date;

public class UnixTime {

    private final long value;
    
    public UnixTime() {
        this(System.currentTimeMillis() / 1000L + 2208988800L);
    }
    
    public UnixTime(long value) {
        this.value = value;
    }
        
    public long value() {
        return value;
    }
        
    @Override
    public String toString() {
        return new Date((value() - 2208988800L) * 1000L).toString();
    }
}

现在我们修改TimeDecoder，生成一个 UnixTime而不是ByteBuf

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

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

根据更新的解码器，TimeClientHandler 不再需要一个ByteBuf

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

更简单优雅，对吧？同样的技术可以在服务器端应用。这次让我们先更新TimeServerHandler 。

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

现在，唯一缺少的部分就是编码器。它是ChannelOutboundHandler接口的实现，他将一个UnitTime转换成ByteBuf，它比编写解码器简单得多，因为在编码消息的时候不需要考虑碎片问题。

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((int)m.value());
        ctx.write(encoded, promise); // (1)
    }
}

1. 在这一行有很多重要的问题

第一，我们传输原始 ChannelPromise 的时候按照原样传输，这样当编码的数据被实际写入电线时，Netty将其标记为成功或失败。

第二，我们没有调用ctx.flush()，这里有一个单独的处理程序方法void flush(ChannelHandlerContext ctx)，目的是覆盖flush()方法。

为了使程序更简化，你可以使用MessageToByteEncoder:

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

剩下最后一件事情，就是在服务器端将写好的编码器 插入到  ChannelPipeline 中的 TimeServerHandler 之前。把它留作一个简单的练习。

关闭您的应用程序

 关闭Netty应用程序就像通过调用shutdownGracefully()关闭你创建的所有 EventLoopGroups一样简单。他返回一个Future当 EventLoopGroup 完全关闭的时候，并且所有属于这个组的管道都将关闭。

终于翻译完啦

二、自己写的一个小例子

 服务器端：

package learn.netty.hello;

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)

            // 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 = 8080;
        new DiscardServer(port).run();
    }
}

处理程序：

package learn.netty.hello;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.ReferenceCountUtil;

public class DiscardServerHandler extends ChannelInboundHandlerAdapter {

    
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf in = (ByteBuf) msg;
        try {
            System.out.print("Client Said:");
            while (in.isReadable()) { // (1)
                System.out.print((char) in.readByte());
                System.out.flush();
            }
            String back = "I received!";
            ByteBuf out = ctx.alloc().buffer(back.length());
            out.writeBytes(back.getBytes());
            ctx.writeAndFlush(out);
            ctx.close();
        } finally {
            ReferenceCountUtil.release(msg); // (2)
        }
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

客户端：

package learn.netty.hello;

import io.netty.bootstrap.Bootstrap;
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.NioSocketChannel;

public class DiscardClient {

    public static void main(String[] args) throws Exception {
        String host = "localhost";
        int port = 8080;
        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 DiscardClientHandler());
                }
            });

            // 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 learn.netty.hello;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.ReferenceCountUtil;

public class DiscardClientHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf in = (ByteBuf) msg;  
        try {
            System.out.print("Server Said:");
            while (in.isReadable()) { // (1)
                System.out.print((char) in.readByte());
                System.out.flush();
            }
            ctx.close();
        } finally {
            ReferenceCountUtil.release(msg); // (2)
        }
    }
    
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        String msg = "Are you received?";  
        ByteBuf encoded = ctx.alloc().buffer(msg.length());  
        encoded.writeBytes(msg.getBytes());  
        ctx.write(encoded);  
        ctx.flush(); 
    }
}

运行即可（详细注释在本文最后给出）

另外一个传递对象并解决随碎片问题例子

服务器：

package learn.netty.pojo;

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;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;

public class Server {

    private int port;

    public Server(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 ObjectEncoder(),
                                    new ObjectDecoder(ClassResolvers.cacheDisabled(this.getClass().getClassLoader())),
                                    new ServerHandler());
                        }
                    }).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 = 8888;
        new Server(port).run();
    }
}

处理程序：

package learn.netty.pojo;

import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import learn.netty.entity.User;

public class ServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        User user = (User) msg;
        System.out.println(user.toString());
        ChannelFuture f = ctx.writeAndFlush(new User("Server", "Thank you for your coming!"));
        f.addListener(ChannelFutureListener.CLOSE);
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

客户端：

package learn.netty.pojo;

import io.netty.bootstrap.Bootstrap;
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.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;

public class Client {

    public static void main(String[] args) throws Exception {
        String host = "localhost";
        int port = 8888;
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            //客户端Bootstrap让Channel使用起来更容易
            Bootstrap b = new Bootstrap();
            //设置事件组
            b.group(workerGroup);
            //设置Channel
            b.channel(NioSocketChannel.class);
            //设置选项
            b.option(ChannelOption.SO_KEEPALIVE, true);
            //设置Channel初始化的处理程序
            b.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(
                            new ObjectEncoder(),
                            new ObjectDecoder(ClassResolvers.cacheDisabled(this.getClass().getClassLoader())),
                            new ClientHandler());
                }
            });

            // 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 learn.netty.pojo;

import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import learn.netty.entity.User;

public class ClientHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        User user = (User) msg;
        System.out.println(user.toString());
    }
    
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ChannelFuture f = ctx.writeAndFlush(new User("Client", "I'm coming!"));
        //f.addListener(ChannelFutureListener.CLOSE);    //调用这个会关闭与服务器之间的通道，导致服务器返回的消息接收不到
//        ChannelFutureListener CLOSE = new ChannelFutureListener() {
//            @Override
//            public void operationComplete(ChannelFuture future) {
//                future.channel().close();
//            }
//        };
        //Channel channel()
        //Returns a channel where the I/O operation associated with this future takes place.
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

运行即可。

下面是详细解释每一句话：

Client：

EventLoopGroup workerGroup = new NioEventLoopGroup();
=====================================================
EventLoopGroup是一个接口，NioEventLoopGroup继承了实现EventLoopGroup接口的抽象类MultithreadEventLoopGroup。
实际上NioEventLoopGroup就是一个线程池，他的角色类似于一个大boss，每当有Channel的任务时，就会派人去处理。

Bootstrap b = new Bootstrap();
=====================================================
Bootstrap 继承 AbstractBootstrap。
它帮助客户端去更方便的使用Channel，针对服务器有ServerBootstrap。在AbstractBootstrap里面我们可以发现它需要用户去管理几个部分：

public B group(EventLoopGroup group)
public B channel(Class<? extends C> channelClass)
public B channelFactory(ChannelFactory<? extends C> channelFactory)
public B localAddress(SocketAddress localAddress)
public <T> B option(ChannelOption<T> option, T value)
public <T> B attr(AttributeKey<T> key, T value)
public B handler(ChannelHandler handler)

一般需要我们来设置的是，group，channel，options和handler。
所以仔细想想，Bootstrap类似于建造者模式，给它需要的东西，不需要知道内部实现细节，他就能自动处理，但是他没有返回一个对象，而是调用connect之后才返回一个对象。

b.group(workerGroup);
=====================================================
这里设置我们定义的事件循环组

b.channel(NioSocketChannel.class);
=====================================================
设置Channel类型。
说到这里就不得不说Channel接口了。它是Netty最核心的接口。NioSocketChannel不过是Channel接口的其中一个子接口的一个实现罢了。类结构：

public class NioSocketChannel extends AbstractNioByteChannel implements io.netty.channel.socket.SocketChannel

public interface SocketChannel extends DuplexChannel

public interface DuplexChannel extends Channel

具体详细信息您就百度吧，我功力太浅，不能误导大家。

b.option(ChannelOption.SO_KEEPALIVE, true);
=====================================================
设置一些套接字的选项。io.netty.channel.ChannelOption<T>类里面有好多定义，太多了你就查看文档吧。

b.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(
                            new ObjectEncoder(),
                            new ObjectDecoder(ClassResolvers.cacheDisabled(this.getClass().getClassLoader())),
                            new ClientHandler());
                }
            });
=====================================================
实际上我们通过ChannelHandler来间接操纵Channel。 handler 方法里的参数 ChannelHandler 是一个接口，在这里的类结构是这样的

public abstract class ChannelInitializer<C extends Channel> extends ChannelInboundHandlerAdapter

public class ChannelInboundHandlerAdapter extends ChannelHandlerAdapter implements ChannelInboundHandler

public interface ChannelInboundHandler extends ChannelHandler

可以看出ChannelInitializer是用来初始化Channel的。再看ch.pipeline().addLast(...)方法，ch.pipeline()返回一个 DefaultChannelPipeline 对象

public class DefaultChannelPipeline implements ChannelPipeline

然后DefaultChannelPipeline 调用addLast方法，把ChannelHandler的实现添加进来。然后按照顺序执行，所以说addLast里面的Handler是有顺序的。
其中的ObjectEncoder和ObjectDecoder是系统为我们写好的编码解码类

ChannelFuture f = b.connect(host, port).sync();
=====================================================
这个ChannelFuture又是个什么东西呢。它是Future的子接口。连接后会返回一个ChannelFuture对象

public interface ChannelFuture extends Future<Void>

Future的概念来自JUC，代表异步的意思。Netty是一个异步IO框架，命名为ChannelFuture代表他与Channel有关。
在Netty中，所有操作的都是异步的，意味着任何IO调用都会立即返回，那么如何获取异步操作的结果？ChannelFuture就是为了结局这个问题而设计的，后面我们还会讨论

f.channel().closeFuture().sync();
=====================================================
一直等待，直到链接关闭

workerGroup.shutdownGracefully();
=====================================================
关闭事件线程组

Handler：

public class ClientHandler extends ChannelInboundHandlerAdapter

=====================================================
处理Channel只需要继承ChannelInboundHandlerAdapter这个类即可。你可以覆盖一些方法。

当有消息的时候会自动调用这个方法。

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        User user = (User) msg;
        System.out.println(user.toString());
    }

当建立好连接自动调用这个。

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ChannelFuture f = ctx.writeAndFlush(new User("Client", "I'm coming!"));
        //f.addListener(ChannelFutureListener.CLOSE);    //调用这个会关闭与服务器之间的通道，导致服务器返回的消息接收不到
//        下面是源码
//        ChannelFutureListener CLOSE = new ChannelFutureListener() {
//            @Override
//            public void operationComplete(ChannelFuture future) {
//                future.channel().close();//从这里可以看出，当操作完成就自动关闭channel了，那服务器返回消息肯定收不到啊
//            }
//        };
        //Channel channel()
        //Returns a channel where the I/O operation associated with this future takes place.
    }

Server：

    public void run() throws Exception {
        //服务器需要两个事件线程组，一个用来接收连接，一个用来处理连接
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap(); // 用来建立服务器的Bootstrap
            b.group(bossGroup, workerGroup)    // 注册线程组
                .channel(NioServerSocketChannel.class) // 设置Channel类型
                    .childHandler(new ChannelInitializer<SocketChannel>() { // 初始化
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(
                                    new ObjectEncoder(),
                                    new ObjectDecoder(ClassResolvers.cacheDisabled(this.getClass().getClassLoader())),
                                    new ServerHandler());
                        }
                    }).option(ChannelOption.SO_BACKLOG, 128) // 选项
                    .childOption(ChannelOption.SO_KEEPALIVE, true); // 子选项

            // 绑定端口，准备接受连接
            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();
        }
    }

handler：

public class ServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        User user = (User) msg;    //因为我添加了解码器，所以在这里就直接强转
        System.out.println(user.toString());
        ChannelFuture f = ctx.writeAndFlush(new User("Server", "Thank you for your coming!"));
        f.addListener(ChannelFutureListener.CLOSE); //发送完回应消息，就关闭Channel
    }
    
    //异常处理
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

 如果有的地方不对，日后我会做修改