手写RPC-NIO学习

发表于 | 评论数： | 热度： ℃   |   字数统计: 2.1k字   |   阅读时长: ≈ 8分钟

概述

NIO是Java1.4推出的一种全新的IO模型，全称是java non-blocking IO，提供ByteBuffer等缓存的容器，达到非阻塞式的高伸缩性网络。

IO模型

IO模型是机器进行IO具体操作方法的一种抽象，每种IO模型都有各自的优缺点，需要注意的是要完成各模型的实际开发需要操作系统的支持，在没有poll、epoll出来之前，java进行非阻塞式的读写操作很复杂，而当上述功能出现之后，java才在基于该功能上添加了nio模块，包名是java.nio，现在在类Linux是基于epoll实现的，类Unix（包含Mac）是基于kqueue 实现的。

Buffer

是一种缓冲数据的容器，可以存储各种基本类型的数据。线程不安全，数据结构如下图所示：
img
类似于一个数组，其中capacity为缓冲数组的长度，为固定的值；
postion表示下一个需要操作的位置；
limit为下一个不可操作的位置；
各种数据的大小关系是0<=position<=limit<=capacity

  • put 写入数据，每次写入数据的地方都是postion，就会使得postion的值变大，当直到填充的数据长度超过了数组的长度，会抛出BufferOverflowException异常；
  • get 读取数据 每次也会进行postion+1操作,这里需要注意到每次读取数据之前必须进行clear操作要不然会出现数据错误的问题，如下使用例子：

错误示例
img

正确示例
img

  • clear() 清空缓冲区的数据，实际上是假清除，只是将postion置位0，limit置位capacity；
    源码如下：
1
2
3
4
5
6
public final Buffer clear() {
 position = 0;
 limit = capacity;
 mark = -1;
 return this;
}
  • allocate(int n) 申请缓冲区，大小由参数决定;
  • wrap(byte[] byets) 同样是申请缓冲区，传入的参数却是byte[],相当于设置的缓冲区大小是byte数组的长度，然后初始化设置了该缓冲容器的值;
  • flip() 切换到读模式，修改limit为postion，position为0;
  • hasRemaining() 查看是否还有数据可读 return position < limit;

一般的使用套路都是:

1
2
3
4
5
6
7
8
ByteBuffer byteBuffer = ByteBuffer.allocate(10);
byteBuffer.clear(); // 清空
byteBuffer.put("hello".getBytes()); // 写入数据
byteBuffer.flip(); // 读数据之前的必备操作
while (byteBuffer.hasRemaining()){ // 数据是否读取完毕
 System.out.print(byteBuffer.get() + "\t" ); // 读取数据
}
System.out.println();

Channel

Channel通道，和pipeline一个意思，类似于IO的Stream，只是stream是单向，要么是Input要么是Output，而Channel是双向的，也就意味着可以通过一个channel进行读写操作了。不过需要注意可以读写操作和能不能读写操作这是两回事。

Nio的channel具体实现主要为FileChannel、DatagramChannel、SocketChannel、ServerSocketChannel四种，分别对应的是文件、UDP和TCP的客户端和服务端。重点介绍SocketChannle和ServerSocketChannel。

SocketChannel

Socketchannel 是客户端，连接一个创建好的TCP网络套接字的通道，可有两种创建方式

  1. 新建一个socketchannel，并连接到服务器上;
1
2
3
SocketChannel socketChannel = SocketChannel.open();
// 连接到本地的8081端口的服务器上
socketChannel.connect(new InetSocketAddress(8081));
  1. 服务器接收到来自客户端的请求接收到的
1
2
3
4
5
// 服务端接收到客户端发送的信息，通过accpet即可获取对应的socketChannel
SocketChannel socketChannel = serverSocketChannel.accept();

// 关闭socketChannel，由于其会抛出异常，最好是放在finally里面，并且做好空判断以及异常捕获
socketChannel.close();

img

由上图所示，需要从channel读数据，以及向外发送数据都需要使用buffer作为缓冲容器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
// 读数据
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
// read 方法会返回读取了多少数据到buffer中，返回-1表示数据以及读取完毕，可以关闭通道了
int count = socketChannel.read(byteBuffer);

// 写数据
String message = "Hello World!";
// 直接调用write方法写入相关的bytebuffer数组
socketChannel.write(message.getBytes());

// 写数据方法2
String message = "Hello World!";
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
byteBuffer.clear();
byetBuffer.put(message.getBytes());
byetBuffer.flip();
while(byteBuffer.hasRemaing()){
 socketChannel.write(byetBuffer);
}

由于其为异步模式，在其调用connect()方法的时候是立即返回结果，连接成功返回true，连接不成功返回false，并继续进行连接（服务自主操作），存在还未建立连接就返回了，所以在使用服务端数据的时候再调用finishConnect()确保链接的建立:

1
2
3
4
5
6
7
8
SocketChannel socketChannel = SocketChannel.open();
socketChannel.configureBlocking (false); // 一定要设置该内容
socketChannel.connect(new InetSocketAddress(8081)); 
while(!socketChannel.finishConnect()){
 ..... // 日志打印等操作，直到连接成功
}
// 这是阻塞模式的
// 连接成功，可以进行读写操作了

可以通过方法isConnectionPending()的返回值确认是否处于连接中。

ServerSocketChannel

ServerSocketChannel 是应用在服务端的，和Socketchannel相对应，主要是用来监听新来的TCP请求的一个通道。绑定的本地端口，IP是本机IP。创建方式如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.socket().bind(new InetSocketAddress(8081));
// 通过对该channel持有的socket进行绑定端口操作
// 之前很奇怪一个问题，这个直接调用socket().bind不会有空指针么？肯定不会有这个错误的
// 在调用socket()，如果发现其内置的socket为null，就会生成一个socket的适配参数替换null
class ServerSocketChannelImpl extends ServerSocketChannel implements SelChImpl {
	private static NativeDispatcher nd;
	private final FileDescriptor fd;
	private int fdVal;
	private volatile long thread = 0L;
	private final Object lock = new Object();
	private final Object stateLock = new Object();
	private static final int ST_UNINITIALIZED = -1;
	private static final int ST_INUSE = 0;
	private static final int ST_KILLED = 1;
	private int state = -1;
	private InetSocketAddress localAddress;
	private boolean isReuseAddress;
	ServerSocket socket; // 这个就是调用socket取得的数据

	public ServerSocket socket() {
		Object var1 = this.stateLock;
		synchronized (this.stateLock) {
			if (this.socket == null) {
				this.socket = ServerSocketAdaptor.create(this);
			}
			return this.socket;
		}
	}
}
public class ServerSocketAdaptor extends ServerSocket {
	private final ServerSocketChannelImpl ssc;
	private volatile int timeout = 0;

	public static ServerSocket create(ServerSocketChannelImpl var0) {
		try {
			return new ServerSocketAdaptor(var0);
		} catch (IOException var2) {
			throw new Error(var2);
		}
	}
}

Selector

selector 是nio中能够管理多个channel通道并感知各个通道的读写状态的一个组件，可以使用单线程管理多个channel的从而同时处理多个网络请求。selector和channel是通过selectorkey绑定的

1
2
3
4
5
6
7
8
// 创建一个selector
Selector selector = Selector.open();
// 创建serversocketChannel并绑定端口
ServerSocketChannel serverSocketChannel ..... 
// 设置为非阻塞模式
ServerSocketChannel.configureBlocking(false); 
// register绑定selector和channel到一个对象SelectionKey中
SelectionKey key = channel.register(selector,Selectionkey.OP_READ);

注意register()方法的第二个参数。这是一个“interest集合”，意思是在通过Selector监听Channel时对什么事件感兴趣。可以监听四种不同类型的事件：

  1. Selectionkey.Connect 可以连接 （客户端接收到可连接到请求）;
  2. Selectionkey.Accept 可以接受 （服务端接到客户端连接的请求）;
  3. Selectionkey.Read 可以读取数据;
  4. Selectionkey.Write 可以写入数据;

SelectionKey
绑定channel和selector的对象，还包含有read集合和interest集合(感兴趣，在register设置的值),还可以通过attachment()方法绑定一些其他数据。配套的还有判断其状态的方法

1
2
3
4
5
6
public class SelectionKeyImpl extends AbstractSelectionKey {
 final SelChImpl channel;
 public final SelectorImpl selector;
 private int index;
 private volatile int interestOps;
 private int readyOps;

选择通道
selector从已经注册好的channel中获取已经准备就绪的通道进行操作，以下三种是获取通道的方法:

  1. int select() // 阻塞模式，至少有一个准备就绪的通道才返回
  2. int select(long timeout) // 加入超时设置
  3. int selectNow() // 会立即返回，返回当前就绪的通道个数
  4. selectedKeys()获取当前就绪的通道集合
  5. close() 关闭当前的selector，使得绑定的key全部不可用，但是通道本身还是可以正常使用的
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
int count = selector.select();
Iterator it = selector.selectedKeys().iterator();
while (it.hasNext()){
 SelectionKey selectionKey = (SelectionKey)it.next();
 it.remove(); // 处理完成一个key就移除掉，无需再次处理
 if(selectionKey.isAcceptable()){
 ServerSocketChannel serverSocketChannel = (ServerSocketChannel) selectionKey.channel();
 SocketChannel socketChannel = serverSocketChannel.accept();
 socketChannel.configureBlocking(false);
 socketChannel.register(selector, SelectionKey.OP_READ);
 }
 if(selectionKey.isReadable()){
 SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
 createProcessor(socketChannel);
 socketChannel.register(selector, SelectionKey.OP_WRITE);
 
 selectionKey.cancel(); // 当前key取消掉了，但是通道依旧可用
 }
}

DatagramChannel

收发UTP包的通道，适用于UTP协议，发送和读取的是用户数据报

1
2
3
4
5
6
DatagramChannel channel = DatagramChannel.open();
channel.socket().bind(new InetSocketAddress(10002));
// 建立了一个本地10002端口的UTP服务端
channel.connect(new InetSocketAddress(10002));
// 连接一个IP默认为本机，端口为10002服务
// 读写和传统的read、write类似

本文标题:手写RPC-NIO学习

文章作者:王洪博

发布时间:2019年05月01日 - 09:05

最后更新:2019年09月12日 - 10:09

原始链接:http://whb1990.github.io/posts/e4966bd.html

▄︻┻═┳一如果你喜欢这篇文章，请点击下方"打赏"按钮请我喝杯 ☕
0%