OIO->NIO

如果你曾经使用Java提供的网络接口工作过&＃xff0c;你可能已经遇到过想从阻塞传输切换到非阻塞传输的情况&＃xff0c;这种切换是比较困难的&＃xff0c;因为阻塞IO和非阻塞IO使用的API有很大的差异&＃xff1b;Netty提供了上层的传输实现接口使得这种情况变得简单。我们可以让所写的代码尽可能通用&＃xff0c;而不会依赖一些实现相关的APIs。当我们想切换传输方式的时候不需要花很大的精力和时间来重构代码。

使用Java的OIO和NIO

下面代码是使用java实现阻塞OIO的例子

package com.beidao.netty.javaio;import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.charset.Charset;/*** Blocking networking without Netty* &＃64;author 0200759**/
public class PlainOioServer {public void server(int port) throws Exception {final ServerSocket socket &＃61; new ServerSocket(port);try{while(true){//accept connectionfinal Socket clientSocket &＃61; socket.accept();System.out.println("Accepted connection from" &＃43; clientSocket);//create new thread to handle connectionnew Thread(new Runnable() {public void run() {OutputStream out;try {out &＃61; clientSocket.getOutputStream();out.write("Hi\r\n".getBytes(Charset.forName("UTF-8")));out.flush();//close connection once message written and flushedclientSocket.close();} catch (Exception e) {try {clientSocket.close();} catch (Exception e2) {e2.printStackTrace();}}}}).start();}}catch (Exception e) {e.printStackTrace();socket.close();}}
}


下面代码是使用Java NIO实现的例子&＃xff1a;

package com.beidao.netty.javaio;import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;public class PlainNioServer {public void server(int port) throws Exception {System.out.println("Listening for connections on port " &＃43; port);//open Selector that handles channelsSelector selector &＃61; Selector.open();//open ServerSocketChannelServerSocketChannel serverChannel &＃61; ServerSocketChannel.open();//get ServerSocketServerSocket serverSocket &＃61; serverChannel.socket();//bind server to portserverSocket.bind(new InetSocketAddress(port));//set to non-blockingserverChannel.configureBlocking(false);//register ServerSocket to selector and specify that it is interested in new accepted clientsserverChannel.register(selector, SelectionKey.OP_ACCEPT);final ByteBuffer msg &＃61; ByteBuffer.wrap("Hi!\r\n".getBytes());while (true) {//Wait for new events that are ready for process. This will block until something happensint n &＃61; selector.select();if(n > 0){//Obtain all SelectionKey instances that received eventsIterator iter &＃61; selector.selectedKeys().iterator();while (iter.hasNext()) {SelectionKey key &＃61; iter.next();iter.remove();try {//Check if event was because new client ready to get acceptedif (key.isAcceptable()) {ServerSocketChannel server &＃61; (ServerSocketChannel) key.channel();SocketChannel client &＃61; server.accept();System.out.println("Accepted connection from "&＃43; client);client.configureBlocking(false);//Accept client and register it to selectorclient.register(selector, SelectionKey.OP_WRITE,msg.duplicate());}//Check if event was because socket is ready to write dataif (key.isWritable()) {SocketChannel client &＃61; (SocketChannel) key.channel();ByteBuffer buff &＃61; (ByteBuffer) key.attachment();//write data to connected clientwhile (buff.hasRemaining()) {if (client.write(buff) &＃61;&＃61; 0) {break;}}client.close();}} catch (Exception e) {key.cancel();key.channel().close();}}}}}
}


netty实现OIO和NIO

下面代码是使用Netty作为网络框架编写的一个阻塞IO例子&＃xff1a;

package com.beidao.netty.io;import java.net.InetSocketAddress;import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
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.nio.NioEventLoopGroup;
import io.netty.channel.sctp.oio.OioSctpServerChannel;
import io.netty.util.CharsetUtil;public class NettyOioServer {public void server(int port) throws Exception {final ByteBuf buf &＃61; Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("Hi!\r\n",CharsetUtil.UTF_8));//事件循环组EventLoopGroup group &＃61; new NioEventLoopGroup();try {//用来引导服务器配置ServerBootstrap bootstrap &＃61; new ServerBootstrap();//使用OIO阻塞模式bootstrap.group(group).channel(OioSctpServerChannel.class).localAddress(new InetSocketAddress(port))//指定ChannelInitialer初始化handlers.childHandler(new ChannelInitializer() {&＃64;Overrideprotected void initChannel(Channel ch) throws Exception {//添加一个“入站”handle到ChannelPiplinech.pipeline().addLast(new ChannelInboundHandlerAdapter() {public void channelActive(ChannelHandlerContext ctx) throws Exception{//连接后&＃xff0c;写消息到客户端&＃xff0c;写完后关闭连接ctx.writeAndFlush(buf.duplicate()).addListener(ChannelFutureListener.CLOSE);}});}});//绑定服务器接受连接ChannelFuture future &＃61; bootstrap.bind().sync();future.channel().closeFuture().sync();} catch (Exception e) {//释放所有资源group.shutdownGracefully();}}
}

下面代码是使用Netty实现异步&＃xff0c;可以看出使用Netty由OIO切换到NIO是非常的方便。

package com.beidao.netty.io;import java.net.InetSocketAddress;import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
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.nio.NioEventLoopGroup;
import io.netty.channel.sctp.nio.NioSctpServerChannel;
import io.netty.util.CharsetUtil;public class NettyNioServer {public void server(int port) throws Exception {final ByteBuf buf &＃61; Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("Hi!\r\n",CharsetUtil.UTF_8));//事件循环组EventLoopGroup group &＃61; new NioEventLoopGroup();try {//用来引导服务器配置ServerBootstrap bootstrap &＃61; new ServerBootstrap();//使用OIO阻塞模式bootstrap.group(group).channel(NioSctpServerChannel.class).localAddress(new InetSocketAddress(port))//指定ChannelInitialer初始化handlers.childHandler(new ChannelInitializer() {&＃64;Overrideprotected void initChannel(Channel ch) throws Exception {//添加一个“入站”handle到ChannelPiplinech.pipeline().addLast(new ChannelInboundHandlerAdapter() {public void channelActive(ChannelHandlerContext ctx) throws Exception{//连接后&＃xff0c;写消息到客户端&＃xff0c;写完后关闭连接ctx.writeAndFlush(buf.duplicate()).addListener(ChannelFutureListener.CLOSE);}});}});//绑定服务器接受连接ChannelFuture future &＃61; bootstrap.bind().sync();future.channel().closeFuture().sync();} catch (Exception e) {//释放所有资源group.shutdownGracefully();}}
}


Transport API

传输API的核心是Channel接口&＃xff0c;它用于所有出站的操作。Channel接口的类层次结构如下:

如上图所示&＃xff0c;每个Channel都会分配一个ChannelPipeline和ChannelConfig。ChannelConfig负责设置并存储配置&＃xff0c;并允许在运行期间更新它们。传输一般有特定的配置设置&＃xff0c;只作用于传输&＃xff0c;没有其他的实现。ChannelPipeline容纳了使用的ChannelHandler实例&＃xff0c;这些ChannelHandler将处理通道传递的“入站”和“出站”数据。ChannelHandler的实现允许你改变数据状态和传输数据。

现在我们可以使用ChannelHandler做下面一些事情&＃xff1a;
- 传输数据时&＃xff0c;将数据从一种格式转换到另一种格式
- 异常通知
- Channel变为有效或无效时获得通知
- Channel被注册或从EventLoop中注销时获得通知
- 通知用户特定事件

ChannelPipeline实现了拦截过滤器模式&＃xff0c;这意味着我们连接不同的ChannelHandler来拦截并处理经过ChannelPipeline的数据或事件。可以把ChannelPipeline想象成UNIX管道&＃xff0c;它允许不同的命令链(ChannelHandler相当于命令)。你还可以在运行时根据需要添加ChannelHandler实例到ChannelPipeline或从ChannelPipeline中删除&＃xff0c;这能帮助我们构建高度灵活的Netty程序。此外&＃xff0c;访问指定的ChannelPipeline和ChannelConfig&＃xff0c;你能在Channel自身上进行操作。Channel提供了很多方法&＃xff0c;如下列表&＃xff1a;
- eventLoop()&＃xff0c;返回分配给Channel的EventLoop
- pipeline()&＃xff0c;返回分配给Channel的ChannelPipeline
- isActive()&＃xff0c;返回Channel是否激活&＃xff0c;已激活说明与远程连接对等
- localAddress()&＃xff0c;返回已绑定的本地SocketAddress
- remoteAddress()&＃xff0c;返回已绑定的远程SocketAddress
- write()&＃xff0c;写数据到远程客户端&＃xff0c;数据通过ChannelPipeline传输过去

Channel是线程安全(thread-safe)&＃xff0c;
下面的代码是一个简单的多线程例子&＃xff1a;

final Channel channel &＃61; ...
//Create ByteBuf that holds data to write
final ByteBuf buf &＃61; Unpooled.copiedBuffer("your data",CharsetUtil.UTF_8);
//Create Runnable which writes data to channel
Runnable writer &＃61; new Runnable() {&＃64;Overridepublic void run() {channel.write(buf.duplicate());}
};
//Obtain reference to the Executor which uses threads to execute tasks
Executor executor &＃61; Executors.newChachedThreadPool();
// write in one thread
//Hand over write task to executor for execution in thread
executor.execute(writer);
// write in another thread
//Hand over another write task to executor for execution in thread
executor.execute(writer);

Netty包含的传输实现

Netty自带了一些传输协议的实现&＃xff0c;虽然没有支持所有的传输协议&＃xff0c;但是其自带的已足够我们来使用。Netty应用程序的传输协议依赖于底层协议&＃xff0c;本节我们将学习Netty中的传输协议。

Netty中的传输方式有如下几种&＃xff1a;
- NIO&＃xff0c;io.netty.channel.socket.nio&＃xff0c;基于java.nio.channels的工具包&＃xff0c;使用选择器作为基础的方法。
- OIO&＃xff0c;io.netty.channel.socket.oio&＃xff0c;基于java.net的工具包&＃xff0c;使用阻塞流。
- Local&＃xff0c;io.netty.channel.local&＃xff0c;用来在虚拟机之间本地通信。
- Embedded&＃xff0c;io.netty.channel.embedded&＃xff0c;嵌入传输&＃xff0c;它允许在没有真正网络的运输中使用ChannelHandler&＃xff0c;可以非常有用的来测试ChannelHandler的实现。

NIO

NIO传输是目前最常用的方式&＃xff0c;它通过使用选择器提供了完全异步的方式操作所有的I/O&＃xff0c;NIO从Java1.4才被提供。NIO中&＃xff0c;我们可以注册一个通道或获得某个通道的改变的状态&＃xff0c;通道状态有下面几种改变&＃xff1a;
- 一个新的Channel被接受并已准备好
- Channel连接完成
- Channel中有数据并已准备好读取
- Channel发送数据出去

处理完改变的状态后需重新设置他们的状态&＃xff0c;用一个线程来检查是否有已准备好的Channel&＃xff0c;如果有则执行相关事件。在这里可能只同时一个注册的事件而忽略其他的。选择器所支持的操作在SelectionKey中定义&＃xff0c;具体如下&＃xff1a;
- OP_ACCEPT&＃xff0c;有新连接时得到通知
- OP_CONNECT&＃xff0c;连接完成后得到通知
- OP_READ&＃xff0c;准备好读取数据时得到通知
- OP_WRITE&＃xff0c;写入数据到通道时得到通知

Netty中的NIO传输就是基于这样的模型来接收和发送数据&＃xff0c;通过封装将自己的接口提供给用户使用&＃xff0c;这完全隐藏了内部实现。如前面所说&＃xff0c;Netty隐藏内部的实现细节&＃xff0c;将抽象出来的API暴露出来供使用&＃xff0c;下面是处理流程图&＃xff1a;

Netty中的NIO传输是“zero-file-copy”,也就是零文件复制&＃xff0c;这种机制可以让程序速度更快&＃xff0c;更高效的从文件系统中传输内容&＃xff0c;零复制就是我们的应用程序不会将发送的数据先复制到JVM堆栈在进行处理&＃xff0c;而是直接从内核空间操作。

OIO

OIO就是java中提供的Socket接口&＃xff0c;java最开始只提供了阻塞的Socket&＃xff0c;阻塞会导致程序性能低。

Local - In VM transport

Netty包含了本地传输&＃xff0c;这个传输实现使用相同的API用于虚拟机之间的通信&＃xff0c;传输是完全异步的。每个Channel使用唯一的
SocketAddress&＃xff0c;客户端通过使用SocketAddress进行连接&＃xff0c;在服务器会被注册为长期运行&＃xff0c;一旦通道关闭&＃xff0c;它会自动注销&＃xff0c;客户端无法再使用它。

连接到本地传输服务器的行为与其他的传输实现几乎是相同的&＃xff0c;需要注意的一个重点是只能在本地的服务器和客户端上使用它们。Local未绑定任何Socket&＃xff0c;值提供JVM进程之间的通信。

Embedded transport

Embedded transport允许更容易的使用不同的ChannelHandler之间的交互&＃xff0c;这也更容易嵌入到其他的ChannelHandler实例并像一个辅助类一样使用它们。它一般用来测试特定的ChannelHandler实现&＃xff0c;也可以在ChannelHandler中重新使用一些ChannelHandler来进行扩展&＃xff0c;为了实现这样的目的&＃xff0c;它自带了一个具体的Channel实现&＃xff0c;即&＃xff1a;EmbeddedChannel。