Netty 5用户指南

时间:2022-02-14 16:28:22

Netty是一个提供异步事件驱动的网络应用框架,用以快速开发高性能、高可靠性的网络服务器和客户端程序。换句话说,Netty是一个NIO框架,使用它可以简单快速地开发网络应用程序,比如客户端和服务端的协议。Netty大大简化了网络程序的开发过程比如TCP和UDP的 Socket的开发。

"快速和简单"并不意味着应用程序会有难维护和性能低的问题,Netty是一个精心设计的框架,它从许多协议的实现中吸收了很多的经验比如FTP、SMTP、HTTP、许多二进制和基于文本的传统协议,Netty在不降低开发效率、性能、稳定性、灵活性情况下,成功地找到了解决方案。

Netty的核心组件包括:EventLoopGroup、ServerBootstrap、Handler等,其中像EventLoopGroup、Channel等都是可以灵活调配的。下面我们从代码示例开始我们的讲解。

  • Bootstrap负责启动引导
  • Buffer是Netty自己封装的缓存器
  • Channel负责管理和建立连接
  • Handler是责任链路模式中的处理者
  • Util是Netty提供和使用到的一些工具

一、Netty Server

一个NettyServer程序主要由两部分组成:(1)BootsTrapping:配置服务器端基本信息;(2)ServerHandler:真正的业务逻辑处理。

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操作的多线程事件循环器,Netty提供了许多不同的EventLoopGroup的实现用来处理不同传输协议。在这个例子中我们实现了一个服务端的应用,因此会有2个NioEventLoopGroup会被使用。第一个经常被叫做‘boss’,用来接收进来的连接。第二个经常被叫做‘worker’,用来处理已经被接收的连接,一旦‘boss’接收到连接,就会把连接信息注册到‘worker’上。 如何知道多少个线程已经被使用,如何映射到已经创建的Channels上都需要依赖于EventLoopGroup的实现,并且可以通过构造函数来配置他们的关系;
2、ServerBootstrap 是一个启动NIO服务的辅助启动类。你可以在这个服务中直接使用Channel,但是这会是一个复杂的处理过程,在很多情况下你并不需要这样做;
3、这里我们指定使用NioServerSocketChannel类来举例说明一个新的Channel如何接收进来的连接;
4、这里的事件处理类经常会被用来处理一个最近的已经接收的Channel。ChannelInitializer是一个特殊的处理类,他的目的是帮助使用者配置一个新的Channel。也许你想通过增加一些处理类比如DiscardServerHandle来配置一个新的Channel或者其对应的ChannelPipeline来实现你的网络程序。当你的程序变的复杂时,可能你会增加更多的处理类到pipline上,然后提取这些匿名类到最顶层的类上。

5、可以设置这里指定的通道实现的配置参数。我们正在写一个TCP/IP的服务端,因此我们被允许设置socket的参数选项比如tcpNoDelay和keepAlive;
6、option()是提供给NioServerSocketChannel用来接收进来的连接。childOption()是提供给由父管道ServerChannel接收到的连接,在这个例子中也是NioServerSocketChannel。

ServerHandler配置如下:

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、chanelRead()事件处理方法。每当从客户端收到新的数据时,这个方法会在收到消息时被调用,这个例子中,收到的消息的类型是ByteBuf;其中ByteBuf是一个引用计数对象,这个对象必须显示地调用release()方法来释放;

2、 exceptionCaught()事件处理方法是当出现Throwable对象才会被调用,即当Netty由于IO错误或者处理器在处理事件时抛出的异常时。在大部分情况下,捕获的异常应该被记录下来并且把关联的channel给关闭掉。然而这个方法的处理方式会在遇到不同异常的情况下有不同的实现,比如你可能想在关闭连接之前发送一个错误码的响应消息。

二、Netty Client

   一般一个简单的Client会扮演如下角色:(1)连接到Server;(2)向Server写数据;(3)等待Server返回数据;(4)关闭连接;
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类似,不过他是对非服务端的channel而言,比如客户端或者无连接传输模式的channel;
2、如果你只指定了一个EventLoopGroup,那他就会即作为一个‘boss’线程,也会作为一个‘workder’线程,尽管客户端不需要使用到‘boss’线程;
3、代替NioServerSocketChannel的是NioSocketChannel,这个类在客户端channel被创建时使用;
4、不像在使用ServerBootstrap时需要用childOption()方法,因为客户端的SocketChannel没有父channel的概念;
5、用connect()方法代替了bind()方法;
ClientHandler实现如下:
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、channelHandlerContext对象提供了许多操作,使你能够触发各种各样的I/O事件和操作;
     ctx.write(Object)方法不会使消息写入到通道上,他被缓冲在了内部,你需要调用ctx.flush()方法来把缓冲区中数据强行输出。或者你可以用更简洁的cxt.writeAndFlush(msg)以达到同样的目的;
     另外一个点需要注意的是ChannelHandlerContext.write()(和writeAndFlush())方法会返回一个ChannelFuture对象,一个ChannelFuture代表了一个还没有发生的I/O操作。
这意味着任何一个请求操作都不会马上被执行,因为在Netty里所有的操作都是异步的。
请参考:http://ifeve.com/netty5-user-guide/
             http://netty.io/wiki/user-guide-for-4.x.html
ChannelAction:此方法会在连接到服务器后被调用;
ChannelRead:此方法会在接收到服务器端数据后调用;
另外,在Client端我们的业务Handler继承的是SimpleChannelInboundHandler,而在服务器端继承的是ChannelInboundHandlerAdapter,那么这二个有什么区别呢?最主要的区别在于SimpleChannelInboundHandler
在接收到数据后会自动release掉数据占用的Bytebuffer资源(自动调用Bytebuffer.release()).而为何服务器端不能用呢?因为我们想让服务器把客户端请求的数据发送回去,而服务器端有可能在ChannelRead方法返回前还没有写完数据,因此不能让它自动release.