如果把Netty比作一台工厂车间, 那么IO线程就是车间里面的运作机器, IO线程一直在无限循环地做着三件事 1.轮询IO事件 2.处理IO事件 3.执行task任务
> 无限循环源码位置: io.netty.channel.nio.NioEventLoop#run > select()方法源码位置: io.netty.channel.nio.NioEventLoop#select
在Netty中轮询IO事件是通过调用select()方法, 至于底层基于select,poll,epoll哪一种, 这个和平台有关.总之, 通过select()方法, 监听着ACCEPT,CONNECT,READ,WRITE等事件.一旦有相应的事件发生, Netty就会根据不同的事件调用不同的方法.
> 处理不同事件的源码位置: io.netty.channel.nio.NioEventLoop#processSelectedKey(java.nio.channels.SelectionKey, io.netty.channel.nio.AbstractNioChannel)
![]()
CONNECT事件. 在前面的文章谈及过, Netty客户端在向服务端发起连接的时候, 并不会阻塞, 而是直接返回, 然后会注册一个CONNECT事件. 当三次握手完成之后, Netty客户端监听到CONNECT事件, 于是调用到上面的方法地方, 最终会注册一个READ事件,这才表示客户端可以接收数据了.
WRITE事件. 这个事件一般情况是不会出现, 只有当Netty向网络中写数据的时候, 由于TCP写缓冲区满了, 至于为什么满, 可能是对端处理数据比较慢, 也可能是网络拥塞等原因. 最终导致无法向TCP写缓冲区写数据了, 这个时候Netty就会注册一个写事件, 当TCP写缓冲区有空余空间的时候, 就会触发这个WRITE事件, Netty就会将之前没有写完的数据, 继续向TCP写缓冲区写入.
READ事件. 当网卡接收到数据之后, 经过内核协议栈之后, 最终到达TCP读缓冲区. Netty监测到READ事件之后,就会读取数据, 经过解码等操作, 将数据交给业务逻辑处理.
ACCEPT事件. 当服务端绑定完端口之后, 就会注册一个ACCEPT事件, 表示此时可以接收客户端的连接了. 当一个客户端发起连接的时候, Netty监测到ACCEPT事件, 最终会调用accept()方法, 接收连接, 以及后面的一系列操作.
总结: 在Netty中, 通过一个无限循环(即for( ;; ){...} ), 调用select()方法, 监听着感兴趣的事件. 不同的事件由不同的方法处理.
select + 事件驱动 + 处理逻辑
在Python中, 也是有IO多路复用的实现.
#! /usr/bin/env python
import socket
from selectors import DefaultSelector, EVENT_READ, EVENT_WRITE
# 解码器. 在Netty中也有解码器,它们的作用都是一样的,将接收到的数据流,解码成有意义的数据
class DecoderHandler(object):
def __init__(self):
self.data = b''
def decode(self, data):
self.data = self.data + data
if len(self.data) > 4:
data = self.data[0:5].decode()
print('接收数据:{}'.format(data))
self.data = self.data[5:]
class Server(object):
def __init__(self):
self.address = ('127.0.0.1', 8080)
self.selector = DefaultSelector()
self.decoder = DecoderHandler()
self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
def start(self):
self.server.bind(self.address)
self.server.listen(50)
self.server.setblocking(False)
# 注册ACCEPT事件
self.selector.register(self.server.fileno(), EVENT_READ, (self.connected, self.server))
self.loop()
# 处理ACCEPT事件
def connected(self, key, mask):
client, addr = key.data[1].accept() # client, addr = self.server.accept()
print('接收客户端{}连接...'.format(addr))
client.setblocking(False)
# 注册读事件
self.selector.register(client.fileno(), EVENT_READ, (self.read_write, client))
def read_write(self, key, mask):
if mask & EVENT_READ: # 读事件
self.recv(key)
if mask & EVENT_WRITE: # 写事件
self.send(key, '\r\nHello, Python\r\n')
def recv(self, key):
client = key.data[1]
data = client.recv(1024)
self.decoder.decode(data)
self.send(key, '\r\nHello, World\r\n')
def send(self, key, msg):
client = key.data[1]
# print('发送数据...')
d = client.send(str(msg).encode())
print(d)
def loop(self):
# 一个无限循环, 和Netty中的run()方法功能一样,通过调用select()方法,一直轮询着事件
while True:
data = self.selector.select()
for key, mask in data:
callback = key.data[0]
callback(key, mask)
if __name__ == '__main__':
server = Server()
server.start()
> 代码位置: >https://github.com/infuq/infuq-others/blob/master/Python/my_server_select.py
以上代码, 是自己实现的一个IO多路复用的例子. 它的核心和上面说的Netty一样, 也需要调用select()方法, 轮询着事件, 也有一个无限循环, 当监听到对应的事件之后, 也有对应的方法处理.
其实, 不管是Netty还是Python中的IO多路复用, 它们的思想都是一样的, 处理过程也是一样的,只是实现的语言不同, 写法不同而已.
其中一点要注意的是, 在上面Python代码中, 它只有两个事件, 一个是读事件EVENT_READ, 一个是写事件EVENT_WRITE. 没有接收事件ACCEPT. 如果仔细观察上面Netty处理事件的代码,会发现 READ和ACCEPT是在一起的, 都是调用了read方法. 其实ACCEPT事件和READ事件一样, 都是需要进行读取操作的.
