netty支持的协议-白红宇

netty支持的协议

阅读量：6076 次

发布时间：2019-06-20

本文共 11051 字，大约阅读时间需要 36 分钟。

流经网络的数据总是具有相同的类型：字节。这些字节是如何流动的主要取决于我们所说的网络传输--一个帮助我们抽象底层数据传输机制的概念。用户并不关心这些细节；他们只想确保他们的字节被可靠地发送和接收。

jdk提供了从阻塞传输到非阻塞传输及异步AIO等，在使用时会因为网络 API 的截然不同而遇到问题。然而，Netty 为它所有的传输实现提供了一个通用 API，这使得这种转换比你直接使用 JDK 所能够达到的简单得多。所产生的代码不会被实现的细节所污染，而你也不需要在你的整个代码库上进行广泛的重构。简而言之，你可以将时间花在其他更有成效的事情上。

一、Channel

Netty除了 TCP 协议以外, Netty 还支持很多其他的连接协议, 并且每种协议还有 NIO(异步 IO) 和 OIO(Old-IO, 即传统的阻塞 IO) 版本的区别.

1.1、Channel 类的层次结构

传输 API 的核心是 interface Channel，它被用于所有的 I/O 操作。Channel 类的层次结构如图 4-1 所示。

如图所示，

1、每个 Channel 都将会被分配一个 ChannelPipeline 和 ChannelConfig。 ChannelConfig 包含了该 Channel 的所有配置设置，并且支持热更新。由于特定的传输可能具有独特的设置，所以它可能会实现一个 ChannelConfig 的子类型。（请参考 ChannelConfig 实现对应的 Javadoc。）

2、每个Channel是独一无二的，所以为了保证顺序将 Channel 声明为 java.lang. Comparable 的一个子接口。因此，如果两个不同的 Channel 实例都返回了相同的散列码，那么 AbstractChannel 中的 compareTo()方法的实现将会抛出一个 Error。

3、ChannelPipeline 持有所有将应用于入站和出站数据以及事件的 ChannelHandler 实例，这些 ChannelHandler 实现了应用程序用于处理状态变化以及数据处理的逻辑。

ChannelHandler 的典型用途包括：

将数据从一种格式转换为另一种格式；

提供异常的通知；

提供 Channel 变为活动的或者非活动的通知；

提供当 Channel 注册到 EventLoop 或者从 EventLoop 注销时的通知；

提供有关用户自定义事件的通知。

拦截过滤器 ChannelPipeline 实现了一种常见的设计模式—拦截过滤器（Intercepting Filter）。UNIX 管道是另外一个熟悉的例子：多个命令被链接在一起，其中一个命令的输出端将连接到命令行中下一个命令的输入端。

1.2、Channel 的方法

除了访问所分配的 ChannelPipeline 和 ChannelConfig 之外，也可以利用 Channel 的其他方法，如下：

eventLoop();// 返回分配给 Channel 的 EventLooppipeline();// 返回分配给 Channel 的 ChannelPipelineisActive();// 如果 Channel 是活动的，则返回 true。活动的意义可能依赖于底层的传输。例如，一个 Socket 传输一旦连接到了远程节点便是活动的，而一个 atagram 传输一旦被打开便是活动的localAddress();// 返回本地的 SokcetAddressremoteAddress();// 返回远程的 SocketAddresswrite 将数据写到远程节点。这个数据将被传递给 ChannelPipeline，并且排队直到它被冲刷flush();// 将之前已写的数据冲刷到底层传输，如一个 SocketwriteAndFlush();// 一个简便的方法，等同于调用 write()并接着调用 flush()

二、Netty中内置的传输

2.1、Netty中内置处理协议的包路径

Netty 内置了一些可开箱即用的传输。因为并不是它们所有的传输都支持每一种协议，所以你必须选择一个和你的应用程序所使用的协议相容的传输。

名称	包	描述
NIO	io.netty.channel.socket.nio	使用 java.nio.channels 包作为基础——基于选择器的方式
Epoll①	io.netty.channel.epoll	由 JNI 驱动的 epoll()和非阻塞 IO。这个传输支持只有在Linux上可用的多种特性，如SO_REUSEPORT，比 NIO 传输更快，而且是完全非阻塞的
OIO	io.netty.channel.socket.oio	使用 java.net 包作为基础——使用阻塞流
Local	io.netty.channel.local	可以在 VM 内部通过管道进行通信的本地传输
Embedded	io.netty.channel.embedded	Embedded 传输，允许使用 ChannelHandler 而又不需要一个真正的基于网络的传输。这在测试你的ChannelHandler 实现时非常有用

2.2、不同连接下的 Channel 类型

不同协议不同的阻塞类型的连接都有不同的 Channel 类型与之对应下面是一些常用的 Channel 类型:

NioSocketChannel, 代表异步的客户端 TCP Socket 连接.

NioServerSocketChannel, 异步的服务器端 TCP Socket 连接.

NioDatagramChannel, 异步的 UDP 连接

NioSctpChannel, 异步的客户端 Sctp 连接.

NioSctpServerChannel, 异步的 Sctp 服务器端连接.

OioSocketChannel, 同步的客户端 TCP Socket 连接.

OioServerSocketChannel, 同步的服务器端 TCP Socket 连接.

OioDatagramChannel, 同步的 UDP 连接

OioSctpChannel, 同步的 Sctp 服务器端连接.

OioSctpServerChannel, 同步的客户端 TCP Socket 连接.

三、各传输类型介绍

3.1、NIO——非阻塞 I/O

NIO 提供了一个所有 I/O 操作的全异步的实现。它利用了自 NIO 子系统被引入 JDK 1.4 时便可用的基于选择器的 API。

选择器背后的基本概念是充当一个注册表，在那里你将可以请求在 Channel 的状态发生变化时得到通知。

可能的状态变化有：

新的 Channel 已被接受并且就绪；

Channel 连接已经完成；

Channel 有已经就绪的可供读取的数据；

Channel 可用于写数据。

选择器运行在一个检查状态变化并对其做出相应响应的线程上，在应用程序对状态的改变做出响应之后，选择器将会被重置，并将重复这个过程。

3.2、NIO——Epoll—用于 Linux 的本地非阻塞传输

　　Linux作为高性能网络编程的平台，其重要性与日俱增，这催生了大量先进特性的开发，其中包括epoll——一个高度可扩展的I/O事件通知特性。这个API自Linux内核版本 2.5.44（2002）被引入，提供了比旧的POSIX select和poll系统调用更好的性能，同时现在也是Linux上非阻塞网络编程的事实标准。Linux JDK NIO API使用了这些epoll调用。

　　Netty为Linux提供了一组NIO API，其以一种和它本身的设计更加一致的方式使用epoll，并且以一种更加轻量的方式使用中断。① 4.3.3 OIO—旧的阻塞 I/O 如果你的应用程序旨在运行于Linux系统，那么请考虑利用这个版本的传输；你将发现在高负载下它的性能要优于JDK的NIO实现。

3.3、OIO—旧的阻塞 I/O

　　Netty 的 OIO 传输实现代表了一种折中：它可以通过常规的传输 API 使用，但是由于它是建立在 java.net 包的阻塞实现之上的，所以它不是异步的。但是，它仍然非常适合于某些用途。

　　例如，你可能需要移植使用了一些进行阻塞调用的库（如JDBC② 有了这个背景，你可能会想，Netty是如何能够使用和用于异步传输相同的API来支持OIO的呢。答案就是，Netty利用了SO_TIMEOUT这个Socket标志，它指定了等待一个I/O操作完成的最大毫秒数。如果操作在指定的时间间隔内没有完成，则将会抛出一个SocketTimeout Exception。Netty 将捕获这个异常并继续处理循环。在EventLoop下一次运行时，它将再次尝试。这实际上也是类似于Netty这样的异步框架能够支持OIO的唯一方式）的遗留代码，而将逻辑转换为非阻塞的可能也是不切实际的。相反，你可以在短期内使用Netty的OIO传输，然后再将你的代码移植到纯粹的异步传输上。

3.4、用于 JVM 内部通信的 Local 传输

　　Netty 提供了一个 Local 传输，用于在同一个 JVM 中运行的客户端和服务器程序之间的异步通信。同样，这个传输也支持对于所有 Netty 传输实现都共同的 API。

　　在这个传输中，和服务器 Channel 相关联的 SocketAddress 并没有绑定物理网络地址；相反，只要服务器还在运行，它就会被存储在注册表里，并在 Channel 关闭时注销。因为这个传输并不接受真正的网络流量，所以它并不能够和其他传输实现进行互操作。因此，客户端希望连接到（在同一个 JVM 中）使用了这个传输的服务器端时也必须使用它。除了这个限制，它的使用方式和其他的传输一模一样。

3.5、Embedded 传输

Netty 提供了一种额外的传输，使得你可以将一组 ChannelHandler 作为帮助器类嵌入到其他的 ChannelHandler 内部。通过这种方式，你将可以扩展一个 ChannelHandler 的功能，而又不需要修改其内部代码。不足为奇的是，Embedded 传输的关键是一个被称为 EmbeddedChannel 的具体的 Channel 实现。在第 9 章中，我们将详细地讨论如何使用这个类来为 ChannelHandler 的实现创建单元测试用例。

四、传输示例

4.1、BIO示例

package com.dxz.nettydemo.bio;import java.io.IOException;import java.net.ServerSocket;import java.net.Socket;public class TimeServer {    public static void main(String[] args) throws IOException {        int port = 8080;        if (args != null && args.length > 0) {            try {                port = Integer.valueOf(args[0]);            } catch (NumberFormatException e) {                // 采用默认值            }        }        ServerSocket server = null;        try {            server = new ServerSocket(port);            System.out.println("The time server is start in port : " + port);            Socket socket = null;            while (true) {                socket = server.accept();                new Thread(new TimeServerHandler(socket)).start();            }        } finally {            if (server != null) {                System.out.println("The time server close");                server.close();                server = null;            }        }    }}package com.dxz.nettydemo.bio;import java.io.BufferedReader;import java.io.IOException;import java.io.InputStreamReader;import java.io.PrintWriter;import java.net.Socket;public class TimeServerHandler implements Runnable {    private Socket socket;    public TimeServerHandler(Socket socket) {        this.socket = socket;    }    @Override    public void run() {        BufferedReader in = null;        PrintWriter out = null;        try {            in = new BufferedReader(new InputStreamReader(this.socket.getInputStream()));            out = new PrintWriter(this.socket.getOutputStream(), true);            String currentTime = null;            String body = null;            while (true) {                body = in.readLine();                if (body == null)                    break;                System.out.println("The time server receive order : " + body);                currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body)                        ? new java.util.Date(System.currentTimeMillis()).toString() : "BAD ORDER";                out.println(currentTime);            }        } catch (Exception e) {            if (in != null) {                try {                    in.close();                } catch (IOException e1) {                    e1.printStackTrace();                }            }            if (out != null) {                out.close();                out = null;            }            if (this.socket != null) {                try {                    this.socket.close();                } catch (IOException e1) {                    e1.printStackTrace();                }                this.socket = null;            }        }    }}package com.dxz.nettydemo.bio;import java.io.BufferedReader;import java.io.IOException;import java.io.InputStreamReader;import java.io.PrintWriter;import java.net.Socket;public class TimeClient {    /**     * @param args     */    public static void main(String[] args) {        int port = 8080;        if (args != null && args.length > 0) {            try {                port = Integer.valueOf(args[0]);            } catch (NumberFormatException e) {                // 采用默认值            }        }        Socket socket = null;        BufferedReader in = null;        PrintWriter out = null;        try {            socket = new Socket("127.0.0.1", port);            in = new BufferedReader(new InputStreamReader(socket.getInputStream()));            out = new PrintWriter(socket.getOutputStream(), true);            out.println("QUERY TIME ORDER");            System.out.println("Send order 2 server succeed.");            String resp = in.readLine();            System.out.println("Now is : " + resp);        } catch (Exception e) {            e.printStackTrace();        } finally {            if (out != null) {                out.close();                out = null;            }            if (in != null) {                try {                    in.close();                } catch (IOException e) {                    e.printStackTrace();                }                in = null;            }            if (socket != null) {                try {                    socket.close();                } catch (IOException e) {                    e.printStackTrace();                }                socket = null;            }        }    }}

4.2、OIO示例

package com.dxz.nettydemo.chapter04;import java.net.InetSocketAddress;import java.nio.charset.Charset;import io.netty.bootstrap.ServerBootstrap;import io.netty.buffer.ByteBuf;import io.netty.buffer.Unpooled;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelFutureListener;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelInboundHandlerAdapter;import io.netty.channel.ChannelInitializer;import io.netty.channel.EventLoopGroup;import io.netty.channel.oio.OioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.oio.OioServerSocketChannel;public class NettyOioServer {    public static void main(String[] args) throws Exception {        NettyOioServer nos = new NettyOioServer();        nos.server(9999);    }        public void server(int port) throws Exception {        final ByteBuf buf = Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("Hi!\r\n", Charset.forName("UTF-8")));        EventLoopGroup group = new OioEventLoopGroup();        try {            //创建 ServerBootstrap            ServerBootstrap b = new ServerBootstrap();            b.group(group)                .channel(OioServerSocketChannel.class) //使用 OioEventLoopGroup以允许阻塞模式（旧的I/O）                .localAddress(new InetSocketAddress(port))                .childHandler(new ChannelInitializer
     
      () {
   //指定 ChannelInitializer，对于每个已接受的连接都调用它                        @Override                        public void initChannel(SocketChannel ch) throws Exception {                            ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
   //添加一个 ChannelInboundHandlerAdapter    以拦截和    处理事件                                                                @Override                                public void channelActive(ChannelHandlerContext ctx) throws Exception {                                    //将消息写到客户端，并添加 ChannelFutureListener，    以便消息一被写完就关闭连接                                    ctx.writeAndFlush(buf.duplicate()).addListener(ChannelFutureListener.CLOSE);                                }                            });                        }                    });            ChannelFuture f = b.bind().sync();            f.channel().closeFuture().sync();        } finally {            group.shutdownGracefully().sync();        }    }}

用telnet localhost 9999测试如下：