Netty框架学习及第一个Netty应用「建议收藏」

大家好，又见面了，我是你们的朋友全栈君。

1.什么是Netty？

    Netty是一个利用Java的高级网络的能力，隐藏其背后的复杂性而提供一个易于使用的API的客户端/服务器框架。Netty提供高性能和可扩展性，让你可以自由地专注于你真正感兴趣的东西。

2.发展历史：

    网络发展初期，花费很多时间学习socket的复杂、寻址等，在C socket库上进行编码，并需要在不同的操作系统上做不同的处理。

    Java早期版本（1995-2002）介绍了足够的面向对象的糖衣来隐藏一些复杂性，但实现客户端-服务器协议仍需要大量的样板代码和大量的监视才能确保他们是对的。早期的API只能通过原生的socket库来支持所谓的 blocking功能。

      JAVA NIO，非阻塞API，阻塞性I/O一般工作流程如图示：

这种方式在连接数比较少的时候还是可以接受的，当并发连接超过10000时，开销会明显增加。此外，每一个县城都有一个默认的堆栈内存分配了128K和1M之间的空间。考虑到整体的内存和操作系统需要处理更多的并发连接资源，所以这似乎不是一个理想的解决方案。

     SELECTOR，实现Java的无阻塞I/O实现的关键，工作流程如下：

最终由Selector决定哪一组注册的socket准备执行I/O。通过通知，一个县城可以同时处理多个并发连接（一个Selector通常由一个线程处理，但具体实施可以使用多个线程）因此，每次读或写操作执行能立即检查完成。该模型可以用较少的线程处理更多连接，这意味着在内存和上下文切换上话费更少的开销，当没有I/O处理时，线程可以被重定向到其他任务上。

    我们可以世界使用JavaAPI构建的NIO构建应用程序，但这样做正确和安全无法保证，实现可靠和可扩展的event-processing（事件处理器）来处理和调度数据并保证尽可能有效，不过这是一个繁琐和容易出错的任务，而这些，就交给了Netty。
3.Netty特点

        设计：

        针对多种传输类型的同一接口-阻塞和非阻塞；

        简单但更强大的线程模型；

        真正的无连接的数据报套接字支持；

        连接逻辑支持复用。

    易用性：

        大量的javadoc和代码实例；

        出了在JDK1.6+额外的限制。

    性能：

        比核心Java API更好的吞吐量，较低的延时；

        资源消耗更少，这个得益于于共享池和重用；

        减少内存拷贝。

    健壮性：

        消除由于慢、快、活重载连接产生的OutOfMemoryError；

        消除经常发现在NIO在告诉网络中的应用中的不公平的读/写比。

    安全：

        完整的SSL/TLS和StartTLS的支持；

        运行在受限的环境例如Applet活OSGI。

    社区：

        发布的更早和更频繁；

        社区驱动。

4.异步和事件驱动

    非阻塞I/O不会强迫我们等待操作的完成。

5.构成部分

    Channel：NIO基本结构，代表一个用于连接到实体如硬件设备、文件、网络套接字或程序组件，能否执行一个或多个不同的I/O操作的开放连接。

    Callback：回调方法，提供给另一种方法作为引用，时间接口可由ChannelHandler的实现来处理。

    Future：提供了另一种通知应用操作已经完成的方式，这个对象作为一个一步操作结果的占位符，他将在将来的某个时候完成并提交结果。Netty提供自己的实现，ChannelFuture，用于执行异步操作时使用。每个Netty的outbound I/O操作都会返回一个ChannelFuture，这样就不会阻塞，这便是Netty所谓的“自底向上的异步和事件驱动”。相关实现的步骤如下：

        1.异步连接到远程对等节点，调用立即返回并提供ChannelFuture；

        2.操作完成后通知注册一个ChannelFutureListener；

        3.当operationComplete()调用时检查操作的撞他；

        4.如果成功就创建一个ByteBuf来保存数据；

        5.异步发送数据到远程，再次返回ChannelFuture；

        6.如果有一个错误则抛出Throwable，描述错误原因。

    Event和Handler：Netty使用不同的事件来通知我们更改的状态或操作的状态，这使我们能够根据发声的事件触发适当的行为。这些行为可能包括：日志、数据转换、流控制、应用程序逻辑，由于Netty是一个网络框架，事件很清晰的跟入栈或出出站数据流相关，因为一些事件可能触发的传入的数据或状态的变化包括：活动或非活动连接、数据的读取、用户事件、错误，出站事件是由于在未来操作将触发的一个动作，这些包括：打开或关闭一个连接到远程、写或冲刷数据到socket。每个事件都可以分配给用户实现处理程序类的方法，这些范例可直接转换为应用程序构建块，如图：

Netty的ChannelHandler是各种处理程序的基本抽象，每个处理器实例就是一个回调，用于执行各种事件的响应。

6.整合

FUTURE,CALLBACK和HANDLER

    Netty的异步编程模型是建立在future和callback的概念上的，所有这些元素的协同为自己的设计提供了强大的力量。拦截操作和转换入站或出站数据只需要提供回调或者利用future操作返回的，这是用的链操作简单、高校，促进编写可重用的、通用的代码。一个Netty的设计的主要目标是促进“关注点分离”，即我们的业务逻辑从网络基础设施应用程序中分离。

SELECTOR,EVENT和EVENT LOOP

    Netty通过触发事件从应用程序中抽象出Selector，从而避免手写调度代码，EventLoop分配给每个Channel来处理所有的事件，包括：注册感兴趣的事件、调度事件到ChannelHandler、安排进一步行动。该EventLoop本身由只有一个线程驱动，它给一个Channel处理所有的I/O事件，并且在EventLoop的生命周期内不会改变，这个简单而强大的线程模型消除你可能对你的ChannelHandler同步的任何关注，这样你就可以专注提供正确的回调逻辑来执行。 

7.第一个Netty应用

7.1 Netty客户端/服务器总览，Echo client/server

图中显示了连接到服务器的多个并发客户端，理论上，客户端可以支持的连接数只受限于使用的JDK版本中的制约。

echo（回声）客户端和服务器之间的交互是很简单的：客户端启动后，建立一个连接发送一个或多个消息到服务器，其中每相呼应消息返回给客户端。这个应用程序并不是非常有用，但这项工作是为了更好的理解请求–相应交互本身，这是一个基本的模式的客户端/服务器系统。

7.2 写一个echo服务器

Netty实现的echo服务器需要下面内容：

    一个服务器handler：该组件实现了服务器的业务逻辑，决定了连接创建后和接受到信息后如何处理；

    Bootstrapping：这个配置服务器的启动代码，最少需要设置服务器绑定的端口，用来监听连接请求。

通过ChannelHandler来实现服务器的逻辑，使用ChannelHandler的方式体现了“关注点分离”的设计原则，并简化业务逻辑的迭代开发的要求，处理程序很简单，每一个方法都可以覆盖到“hook（钩子）”在活动周期适当的点。牢记两点：

    ChannelHandler是给不同类型的事件调用；

    应用程序实现或扩展ChannelHandler挂接到事件生命周期和提供自定义应用逻辑。

Echo server将接受到的数据拷贝发送给客户端，因此，我们需要实现ChannelInboundHandler接口，用于自定义处理入站事件的方法，当前应用简单，只需继承ChannelInboundHandlerAdapter就行了，该类提供了默认ChannelInboundHandler的实现，所以只需覆盖以下方法：

    channelRead()–每个信息入站都会调用，覆盖该方法是因为我们需要处理所有接收到的数据；

    channelReadComplete()–通知处理器最后的channelRead()是当前处理中的最后一条消息调用；

    exceptionCaught()-读操作时捕获到异常时调用，覆盖该方法使我们能够应对任何Throwable的子类   型，在这种情况下我们记录、并关闭所有可能处在未知状态的连接，它通常是难以从连接错误中恢复，所以干脆关闭远程连接，当然，也有可能的情况是可以从错误中恢复的，所以可以用一个更复杂的措施来尝试识别和处理这样的情况。每个Channel都有一个关联的ChannelPipeline，它代表了ChannelHandler实例的链，适配器处理的实现知识讲一个处理方法调用转发到链中的下一个处理器，因此，如果一个Netty应用程序不覆盖exceptionCaught，那么这些错误最终将到达ChannelPipeline，并且结束警告将被记录。

使用工具IntelliJ IDEA 2017.2.3工具创建Maven项目，命名为echoserver,pom.xml内容如下

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com</groupId>
    <artifactId>echo-server</artifactId>
    <version>1.0-SNAPSHOT</version>

    <name>echo-server</name>

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

    </dependencies>

</project>

创建业务核心处理逻辑EchoServerHandler，内容与注释如下：

package com;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;

@ChannelHandler.Sharable //1.标志这类的实例之间可以再channel里面共享
public class EchoServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf in = (ByteBuf)msg;
        System.out.println("Server received:"+in.toString(CharsetUtil.UTF_8));//2.日志输出到控制台
        ctx.write(in);//3.将所接收的消息返回给发送者，注意，此时还没有冲刷数据
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush(Unpooled.EMPTY_BUFFER)//4.冲刷所有待审消息到远程节点，关闭通道后，操作完成
        .addListener(ChannelFutureListener.CLOSE);

    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.fireExceptionCaught(cause);//5.打印异常堆栈跟踪
        ctx.close();//6.关闭通道
    }
}

创建完业务核心处理逻辑之后，创建引导服务器EchoServer，作用如下：

    监听和接收进来的连接请求；

    配置Channel来通知一个关于入站消息的EchoServerHandler实例。

内容如下：

package com;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
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 java.net.InetSocketAddress;


public class EchoServer {

    private final int port;

    public EchoServer(int port) {
        this.port = port;
    }

    public static void main(String[] args) throws Exception{
        //        if(args.length != 1){
//            System.err.println("Usage: " + EchoServer.class.getSimpleName() + "<port>");
//            return;
//        }
//        int port = Integer.parseInt(args[0]);//1.设置端口值
        int port = 8990;//1.设置端口值
        new EchoServer(port).start();//2.启动服务
    }

    public void start() throws Exception{
        EventLoopGroup group = new NioEventLoopGroup();//3.创建EventLoopGroup，一个线程
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(group)//4.创建ServerBootstrap，此处也可放入多个EventLoopGroup
                .channel(NioServerSocketChannel.class)//5.指定使用NIO的传输Channel，指定信道类型
                .localAddress(new InetSocketAddress(port))//6.设置socket地址使用所选端口
                .childHandler(new ChannelInitializer<SocketChannel>() {//7.当有一个新的连接被接受，一个新的子Channel将被创建，ChannelInitializer添加EchoServerHandler到Channel的ChannelPipeline，
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception{
                        ch.pipeline().addLast(
                                new EchoServerHandler());
                    }
                });
            ChannelFuture f = b.bind().sync();//8.绑定的服务器，sync等待服务器关闭，调用sync()的原因是当前线程阻塞
            System.out.println(EchoServer.class.getName() + " started and listen on " + f.channel().localAddress());
            f.channel().closeFuture().sync();//9.关闭Channel和块
        } finally {
            group.shutdownGracefully().sync();//10.关闭EventLoopGroup，释放所有资源
        }
    }
}

服务器的主代码组件是：EchoServerHandler实现了的业务逻辑、在main()方法中引导了服务器。

执行引导服务器所需的步骤是：

    创建ServerBootstrap实例引导服务器并随后绑定；

       创建并分配一个NioEventLoopGroup实例来市里事件的处理，如接受新的连接和读/写数据；

       指定本地InetSocketAddress给服务器绑定； 

    通过EchoServerHandler实例给每一个新的Channel初始化；

    最后调用ServerBootstrap.bind()绑定服务器。

7.3 写一个echo客户端

客户端要做的就是：

    连接服务器；

    发送信息；

    发送的每个信息，等待和接受从服务器返回的同样的信息；

    关闭连接。

用ChannelHandler实现客户端逻辑

跟写服务器一样，Netty提供了ChannelInboundHandler来处理数据，下面的例子中，我们使用SimpleChannelInboundHandler来处理所有的任务，需要覆盖三个方法：

    channelActive()–服务器的连接被建立后调用，一旦建立了连接，字节序列被发送到服务器；

    channelRead0()–在接收到数据时被调用，由服务器所发送的消息可以以块的形式被接收。即，当服务器发送五个字节是不是保证所有的5个字节会立刻收到，即使只有5个字节，channelRead0()方法可被调用两次，第一次用一个ByteBuf装载3个字节和第二次一个ByteBuf装载2个字节，唯一要保证的是，该字节将按照他们发送的顺序分别被接收；

    exceptionCaught()–捕获一个异常时调用。

创建Maven项目，命名echoclient，pom.xml文件内容如下

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com</groupId>
    <artifactId>echo-client</artifactId>
    <version>1.0-SNAPSHOT</version>

    <name>echo-client</name>

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

    </dependencies>

</project>

创建EchoClientHandler.java,内容如下

package com;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.CharsetUtil;

@ChannelHandler.Sharable//1.标记这个类的实例可以在 channel 里共享
public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {

    @Override
    public void channelActive(ChannelHandlerContext ctx){
        ctx.writeAndFlush(Unpooled.copiedBuffer("Netty rocks!", //2.当被通知该 channel 是活动的时候就发送信息
                CharsetUtil.UTF_8));
    }
    @Override
    public void channelRead0(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        System.out.println("Client received: " + in.toString(CharsetUtil.UTF_8));//3.打印接收到的信息
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();//4.打印异常堆栈跟踪
        ctx.close();//5.关闭通道
    }
}

创建引导客户端EchoClient，需要host、port两个参数连接服务器，内容如下

package com;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
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 java.net.InetSocketAddress;

public class EchoClient {

    private final String host;

    private final int port;

    public EchoClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public void start()throws Exception{
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();//1.创建Bootstrap
            b.group(group)//2.指定EventLoopGroup来处理客户端事件，由于我们NIO传输，所以用到了NioEventLoopGroup的实现
                .channel(NioSocketChannel.class)//3.使用的Channel类型是一个用于NIO传输，也可以使用和服务器不一样的类型
                .remoteAddress(new InetSocketAddress(host, port))//4.设置服务器的InetSocketAddress
                .handler(new ChannelInitializer<SocketChannel>() {//5.当建立一个连接和一个新的通道时，创建添加到 EchoClientHandler 实例 到 channelpipeline
                    @Override
                    public void initChannel(SocketChannel ch)throws Exception{
                        ch.pipeline().addLast(new EchoClientHandler());
                    }
                });
            ChannelFuture f = b.connect().sync();//6.连接到远程，等待连接完成
            f.channel().closeFuture().sync();//7.阻塞直到Channel关闭
        } finally {
            group.shutdownGracefully().sync();//8.调用shutdownGracefully来关闭线程池和释放所有资源
        }
    }

    public static void main(String[] args)throws Exception{
//        if(args.length != 2){
//            System.err.println("Usage: " + EchoClient.class.getSimpleName() + "<host> <port>");
//            return;
//        }
//        final String host = args[0];
//        final int port = Integer.parseInt(args[1]);

        final String host = "localhost";
        final int port = 8990;

        new EchoClient(host,port).start();
    }
}

启动Echo服务端，直接运行main()方法，控制台打印如下：

启动Echo客户端，直接运行main()方法，控制台打印如下：

而在这之前的服务端控制台则会打印如下语句：

至此一个简单的Netty应用搭建完成，后续继续提供深入学习。