初识Netty：背景、现状与趋势

gzx0

2020-07-05

Netty概述

Netty由Trustin Lee (韩国，Line公司) 2004年开发。

Netty is an asynchronous event-driven network application framework for rapid development of maintainable high performance protocol servers & clients.

本质：网络应用程序框架
实现：异步、事件驱动
特性：高性能、可维护、快速开发
用途：开发服务器和客户端

Netty的结构图：
初识Netty：背景、现状与趋势
其结构主要分三个部分：

最底层的核心层：零拷贝的功能丰富的Byte Buffer、通用的通信层API、可扩展的事件模型
支持的传输层服务：TCP-Socket&UDP-Datagram、HTTP Tunnel、In-VM Pipe
支持的协议：HTTP（应用层协议）、WebSocket、Protobuf（编解码方式）等

Netty之Hello World

Netty的Hello world代码相对来说要复杂一些，但客户端和服务端都包含了NioEventLoopGroup，Pipeline等相同的脚手架代码。

在pom中引入netty的依赖

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.50.Final</version>
</dependency>

创建Netty的Hello World。下面的示例程序使用netty的构建了一个基本的客户端-服务端的模型，由NioEventLoopGroup、ServerBootstrap、ChannelInitializer等组件构成。

服务端代码：

/**
构建netty Server端启动的bootstrap.
*/
public class EchoServer {
    public static void main(String[] args) throws InterruptedException {
        // 接收客户端连接
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        // 处理客户端连接
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
            .handler(new LoggingHandler(LogLevel.INFO))
            .childHandler(new EchoServerInitializer());

            ChannelFuture channelFuture = serverBootstrap.bind(8030).sync();
            channelFuture.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully(); 
            workerGroup.shutdownGracefully();
        }
    }
}

/**
初始化server端的socket channel pipiline。
*/
public class EchoServerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        pipeline.addLast(new LoggingHandler(LogLevel.INFO));
        pipeline.addLast(new EchoServerHandler());
    }
}

/**
Server handler
*/
public class EchoServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 接收到客户端消息后，将消息原样输出给客户端
        ctx.write(msg);
    }
    
    @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

客户端代码

public class EchoClient {
    public static void main(String[] args) throws InterruptedException {
        // configure the client.
        EventLoopGroup group = new NioEventLoopGroup();
        
        try{
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
            .channel(NioSocketChannel.class)
            .option(ChannelOption.TCP_NODELAY,true)
            .handler(new EchoClientInitializer());
            
            // start the client.
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1",8030).sync();
            
            // wait util the connection is closed.
            channelFuture.channel().closeFuture().sync();
        }
        finally {
            // shut down the event loop to terminate all threads.
            group.shutdownGracefully();
        }
    }
}

public class EchoClientInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        pipeline.addLast(new LoggingHandler(LogLevel.INFO));
        pipeline.addLast(new EchoClientHandler());
    }
}

public class EchoClientHandler extends ChannelInboundHandlerAdapter {
    private final ByteBuf firstMessage;
    
    public EchoClientHandler() {
        firstMessage = Unpooled.wrappedBuffer("I am echo message".getBytes());
    }
    
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush(firstMessage);
    }
    
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ctx.write(msg);
    }
    
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        TimeUnit.SECONDS.sleep(3);
        ctx.flush();
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}