原创

Java-NIOChannel详解

全网最全 Java 知识点整合总目录入口猛戳-->www.gameboys.cn

Netty 全网最全示例源码地址猛戳-->https://github.com/Sniper2016/NettyStudy

Channel 在 JDK 中注释为:

A channel represents an open connection to an entity such as a hardware device, a file, a network socket, or a program component that is capable of performing one or more distinct I/O operations, for example reading or writing.

译文:通道表示到实体(如硬件)的开放连接有能力的设备、文件、网络套接字或程序组件执行一个或多个不同的 I/O 操作,例如读取或写。

(笔者的理解)通俗讲,channel就是一种连接多个具备处理I/O操作桥梁,也可比作自来水公司到我家水龙头之间的水管。

1.Java-NIOChannel 继承关系

从上面继承图我们可以看出,socket相关的channel都继承SelectableChannel,其中SelectableChannel我们从名字里面可以指导,他提供NIO操作中selector的功效,简单讲就是可以将channel注册到Selector,selector即为多路复用的select,只有FileChannel比较简单。

2.主要的 channel 类

  • (1)FileChannel FileChannel 用于文件的数据读写。

  • (2)DatagramChannel DatagramChannel 用于 UDP 的数据读写。

  • (3)SocketChannel SocketChannel 用于 TCP 的数据读写。

  • (4)ServerSocketChannel ServerSocketChannel 允许我们监听 TCP 链接请求,每个请求会创建会一个 SocketChannel。

这个四种通道,涵盖了 UDP 和 TCP 网络 IO 以及文件 IO 的操作。

3 各种 channel 举例

FileChannel 举例

FileChannel 是操作文件的 Channel,我们可以通过 FileChannel 从一个文件中读取数据,也可以将数据写入到文件中。

package cn.gameboys.nio;

import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;

public class FileChannelTest {

 public static void main(String[] args) throws IOException {

  RandomAccessFile inFile = new RandomAccessFile("D:\\a.txt""rw");
  RandomAccessFile outFile = new RandomAccessFile("D:\\b.txt""rw");

  FileChannel inChannel = inFile.getChannel();
  FileChannel outChannel = outFile.getChannel();

  //从文件channel读取1024长度的数据
  ByteBuffer inBuf = ByteBuffer.allocate(1024);
  inChannel.read(inBuf);
  System.out.println(new String(inBuf.array()));


  //将一个1024长度的byteBuffer写入文件中
  ByteBuffer buf = ByteBuffer.allocate(1024);
  String newData = "New String to write to file..." + System.currentTimeMillis();
  buf.clear();
  buf.put(newData.getBytes());
  buf.flip();
  while (buf.hasRemaining()) {
   outChannel.write(buf);
  }

  //关掉流
  inFile.close();
  outFile.close();
  inChannel.close();
  outChannel.close();

 }

}



SocketChannel、ServerSocketChannel 举例

NIO 服务端

package cn.gameboys.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
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;

/**
 * NIO服务端
 *
 * @author www.gameboys.cn
 */
public class NIOServer {
 // 通道管理器
 private Selector selector;

 /**
  * 获得一个ServerSocket通道,并对该通道做一些初始化的工作
  *
  * @param port
  *            绑定的端口号
  * @throws IOException
  */
 public void initServer(int port) throws IOException {
  // 获得一个ServerSocket通道
  ServerSocketChannel serverChannel = ServerSocketChannel.open();
  // 设置通道为非阻塞
  serverChannel.configureBlocking(false);
  // 将该通道对应的ServerSocket绑定到port端口
  serverChannel.socket().bind(new InetSocketAddress(port));
  // 获得一个通道管理器
  this.selector = Selector.open();
  // 将通道管理器和该通道绑定,并为该通道注册SelectionKey.OP_ACCEPT事件,注册该事件后,
  // 当该事件到达时,selector.select()会返回,如果该事件没到达selector.select()会一直阻塞。
  serverChannel.register(selector, SelectionKey.OP_ACCEPT);
 }

 /**
  * 采用轮询的方式监听selector上是否有需要处理的事件,如果有,则进行处理
  *
  * @throws IOException
  */
 @SuppressWarnings("unchecked")
 public void listen() throws IOException {
  System.out.println("服务端启动成功!");
  // 轮询访问selector
  while (true) {
   // 当注册的事件到达时,方法返回;否则,该方法会一直阻塞
   selector.select();
   // 获得selector中选中的项的迭代器,选中的项为注册的事件
   Iterator ite = this.selector.selectedKeys().iterator();
   while (ite.hasNext()) {
    SelectionKey key = (SelectionKey) ite.next();
    // 删除已选的key,以防重复处理
    ite.remove();
    // 客户端请求连接事件
    if (key.isAcceptable()) {
     ServerSocketChannel server = (ServerSocketChannel) key.channel();
     // 获得和客户端连接的通道
     SocketChannel channel = server.accept();
     // 设置成非阻塞
     channel.configureBlocking(false);

     // 在这里可以给客户端发送信息哦
     channel.write(ByteBuffer.wrap(new String("向客户端发送了一条信息").getBytes()));
     // 在和客户端连接成功之后,为了可以接收到客户端的信息,需要给通道设置读的权限。
     channel.register(this.selector, SelectionKey.OP_READ);

     // 获得了可读的事件
    } else if (key.isReadable()) {
     read(key);
    }

   }

  }
 }

 /**
  * 处理读取客户端发来的信息 的事件
  *
  * @param key
  * @throws IOException
  */
 public void read(SelectionKey key) throws IOException {
  // 服务器可读取消息:得到事件发生的Socket通道
  SocketChannel channel = (SocketChannel) key.channel();
  // 创建读取的缓冲区
  ByteBuffer buffer = ByteBuffer.allocate(1024);
  channel.read(buffer);
  byte[] data = buffer.array();
  String msg = new String(data).trim();
  System.out.println("服务端收到信息:" + msg);
  ByteBuffer outBuffer = ByteBuffer.wrap(msg.getBytes());
  channel.write(outBuffer);// 将消息回送给客户端
 }

 /**
  * 启动服务端测试
  *
  * @throws IOException
  */
 public static void main(String[] args) throws IOException {
  NIOServer server = new NIOServer();
  server.initServer(8000);
  server.listen();
 }

}



NIO 客户端

package cn.gameboys.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;

/**
 * NIO客户端
 *
 * @author www.gameboys.cn
 */
public class NIOClient {
 // 通道管理器
 private Selector selector;

 /**
  * 获得一个Socket通道,并对该通道做一些初始化的工作
  *
  * @param ip
  *            连接的服务器的ip
  * @param port
  *            连接的服务器的端口号
  * @throws IOException
  */
 public void initClient(String ip, int port) throws IOException {
  // 获得一个Socket通道
  SocketChannel channel = SocketChannel.open();
  // 设置通道为非阻塞
  channel.configureBlocking(false);
  // 获得一个通道管理器
  this.selector = Selector.open();

  // 客户端连接服务器,其实方法执行并没有实现连接,需要在listen()方法中调
  // 用channel.finishConnect();才能完成连接
  channel.connect(new InetSocketAddress(ip, port));
  // 将通道管理器和该通道绑定,并为该通道注册SelectionKey.OP_CONNECT事件。
  channel.register(selector, SelectionKey.OP_CONNECT);
 }

 /**
  * 采用轮询的方式监听selector上是否有需要处理的事件,如果有,则进行处理
  *
  * @throws IOException
  */
 @SuppressWarnings("unchecked")
 public void listen() throws IOException {
  // 轮询访问selector
  while (true) {
   selector.select();
   // 获得selector中选中的项的迭代器
   Iterator ite = this.selector.selectedKeys().iterator();
   while (ite.hasNext()) {
    SelectionKey key = (SelectionKey) ite.next();
    // 删除已选的key,以防重复处理
    ite.remove();
    // 连接事件发生
    if (key.isConnectable()) {
     SocketChannel channel = (SocketChannel) key.channel();
     // 如果正在连接,则完成连接
     if (channel.isConnectionPending()) {
      channel.finishConnect();

     }
     // 设置成非阻塞
     channel.configureBlocking(false);

     System.out.println("连接服务器成功,发送一条下次过去。");
     // 在这里可以给服务端发送信息哦
     channel.write(ByteBuffer.wrap(new String("我是www.gameboys.cn").getBytes()));
     // 在和服务端连接成功之后,为了可以接收到服务端的信息,需要给通道设置读的权限。
     channel.register(this.selector, SelectionKey.OP_READ);

     // 获得了可读的事件
    } else if (key.isReadable()) {
     read(key);
    }

   }

  }
 }

 /**
  * 处理读取服务端发来的信息 的事件
  *
  * @param key
  * @throws IOException
  */
 public void read(SelectionKey key) throws IOException {
  // 和服务端的read方法一样
 }

 /**
  * 启动客户端测试
  *
  * @throws IOException
  */
 public static void main(String[] args) throws IOException {
  NIOClient client = new NIOClient();
  client.initClient("127.0.0.1", 8000);
  client.listen();
 }

}


DatagramSocket 示例

UDP 服务端

package cn.gameboys.nio;

import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;

public class UDPReceiveTest {

 public static void main(String[] args) throws Exception {
  // 创建接收端Socket对象
  DatagramSocket ds = new DatagramSocket(10086);
  // 接收数据
  byte[] bytes = new byte[1024];
  int length = bytes.length;
  DatagramPacket dp = new DatagramPacket(bytes, length);
  ds.receive(dp);
  // 解析数据
  InetAddress address = dp.getAddress();
  byte[] data = dp.getData();
  int len = dp.getLength();
  // 输出数据
  System.out.println(address);
  System.out.println(new String(data, 0, len));

 }
}



UDP 客户端


package cn.gameboys.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;

public class UdpTest {

 public static void main(String[] args) throws IOException {

  //获得一个channel
  DatagramChannel channel = DatagramChannel.open();

  //发送消息
  String newData = "hi gameboys.cn ..." + System.currentTimeMillis();
  ByteBuffer buf = ByteBuffer.allocate(48);
  buf.clear();
  buf.put(newData.getBytes());
  buf.flip();
  channel.connect(new InetSocketAddress("127.0.0.1", 10086));
  channel.write(buf);
 }
}



从上面的代码中,我们可以看出,其实 udp 可以不使用 DatagramChannel 就能实现数据报文的传输,之所以还抽象出一个 DatagramChannel,一方面是为了写法通用,另一方面为了使用多路复用的功能而引入的 channel。

4.总结:

  • 基本上 channel 都有 channel.write(ByteBuffer bf)方法和 channel.read(ByteBuffer bf)方法,其中 tcp 和 udp 的 channel 还有 channel.register(Selector sel, int ops)方法,进而使用多路复用 io 模型

  • channel 用完之后都需要关闭

正文到此结束