Netty 之入门应用

2017-09-02 本文已影响113人被称为L的男人

说明

系列文章：http://www.jianshu.com/p/594441fb9c9e

本文完全参考自《Netty权威指南（第2版）》，李林峰著。

Netty 环境搭建

例程使用Maven构建工程，在pom文件中，加入Netty的依赖。

<!-- https://mvnrepository.com/artifact/io.netty/netty-all -->
<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.14.Final</version>
</dependency>

服务端程序

public class TimeServer {

    public void bind(int port) throws Exception {
        // 配置服务端的NIO线程
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024).childHandler(new ChildChannelHandler());
            // 绑定端口，同步等待成功
            ChannelFuture f = bootstrap.bind(port).sync();
            // 等待服务端监听端口关闭
            f.channel().closeFuture().sync();
        } finally {
            // 优雅退出，释放线程池资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
    
    private class ChildChannelHandler extends ChannelInitializer<SocketChannel> {

        @Override
        protected void initChannel(SocketChannel arg0) throws Exception {
            arg0.pipeline().addLast(new TimeServerHandler());
        }
    }
    
    public static void main(String[] args) throws Exception {
        int port = 8080;
        new TimeServer().bind(port);
    }
}

public class TimeServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        byte[] req = new byte[buf.readableBytes()];
        buf.readBytes(req);
        String body = new String(req, Charsets.UTF_8);
        
        System.out.println("The time server receive order : " + body);
        String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new Date().toString() : "BAD ORDER";
        ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
        ctx.write(resp);
    }
    
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

流程：

NioEventLoopGroup是线程组，它包含了一组NIO线程，专门用于网络事件的处理。创建两个的原因是一个用于服务端接受客户端的连接，另一个进行SocketChannel的网络读写。
ServerBootStrap对象，是Netty用于启动NIO服务端的辅助启动类，目的是降低服务端的开发复杂度。
调用ServerBootStrap对象的group方法，将两个NIO线程组当作参数传递到ServerBootStrap中。
接着设置Channel为NioServerSocketChannel，功能对应于JDK NIO类中的ServerSocketChannel类。
配置NioServerSocketChannel的TCP参数，backlog设置为1024
绑定I/O事件的处理类ChildChannelHandler，用于处理网络I/O事件，例如记录日志、对消息进行编解码等。
服务端启动辅助类配置完成后，调用它的bind方法绑定监听端口，调用它的同步阻塞方法sync等待绑定操作完成。
f.channel().closeFuture().sync()方法进行阻塞，等待服务端链路关闭后，main函数才退出。

客户端代码

public class TimeClient {

    public void connect(int port, String host) throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new TimeClientHandler());
                        }
                    });
            // 发起异步连接操作
            ChannelFuture future = bootstrap.connect(host, port).sync();

            // 等待客户端链路关闭
            future.channel().closeFuture().sync();
        } finally {
            // 优雅退出，释放NIO线程组
            group.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        int port = 8080;
        new TimeClient().connect(port, "127.0.0.1");
    }
}

public class TimeClientHandler extends ChannelInboundHandlerAdapter {

    private final ByteBuf message;

    public TimeClientHandler() {
        byte[] req = "QUERY TIME ORDER".getBytes();
        message = Unpooled.buffer(req.length);
        message.writeBytes(req);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush(message);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        byte[] req = new byte[buf.readableBytes()];
        buf.readBytes(req);
        String body = new String(req, Charsets.UTF_8);
        System.out.println("Now is : " + body);
    }

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

调试和运行

启动后，客户端输出：

Now is : Sat Aug 19 13:41:54 CST 2017

服务端输出：

The time server receive order : QUERY TIME ORDER

问题：TCP粘包/拆包

是什么

TCP是个“流”协议，流就是没有界限的一串数据。TCP底层并不了解业务上层业务数据的具体含义，它会根据TCP缓冲区的实际情况进行包的划分。所以在业务上，一个完整的包可能会被TCP拆成多个包发送，也有可能把多个小的包封装成一个大包发送。

产生原因

应用程序write写入的字节>套接口发送缓冲区大小
进行MSS大小的TCP分段
以太网帧的payload大于MTU进行IP分片

解决策略

因为底层的TCP无法理解上层的业务数据，所以只能通过上层的应用协议栈来解决这个问题，主要的解决方案有：

消息定长，例如每个报文的大小固定是200个字节，不够则空格补齐。
在包尾增加回车换行进行分割，例如FTP协议。
将消息分成：消息头、消息体。消息头中包含要表示的消息总长度（消息体长度）的字段，通常是将消息头的第一个字段使用int32来表示消息的总长度。
更复杂的应用层协议。

代码模拟

服务端：

public class TimeServerHandler extends ChannelInboundHandlerAdapter {

    private int counter;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        byte[] req = new byte[buf.readableBytes()];
        buf.readBytes(req);
        String body = new String(req, Charsets.UTF_8).substring(0,
                req.length - System.getProperty("line.separator").length());

        System.out.println("count = " + ++counter);
        System.out.println("The time server receive order : " + body);
        String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new Date().toString() : "BAD ORDER";
        ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
        ctx.write(resp);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

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

客户端：

public class TimeClientHandler extends ChannelInboundHandlerAdapter {

    private byte[] req;
    private int counter;

    public TimeClientHandler() {
        req = ("QUERY TIME ORDER" + System.getProperty("line.separator")).getBytes();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ByteBuf message = null;
        for (int i = 0; i < 100; i++) {
            message = Unpooled.buffer(req.length);
            message.writeBytes(req);
            ctx.writeAndFlush(message);
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        byte[] req = new byte[buf.readableBytes()];
        buf.readBytes(req);
        String body = new String(req, Charsets.UTF_8);
        System.out.println("Now is : " + body + ", the counter is " + ++counter);
    }

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

利用LineBasedFrameDecoder解决TCP粘包问题

TimeServer和TimeClient中，都增加下面的代码：

arg0.pipeline().addLast(new LineBasedFrameDecoder(1024));
arg0.pipeline().addLast(new StringDecoder());

ServerHandler和ClientServer中，都将msg直接变成String。

LineBasedFrameDecoder：依次遍历ByteBuf中的可读字节，判断是否有换行符，如果有就以此位置结束。
StringDecoder：将接收到的对象转换成字符串，然后继续调用后面的Handler。

完整代码