Python 实现 IO 多路复用

原创

花落花相惜

修改于 2021-11-24 14:39:40

1.1K0

文章被收录于专栏：花落的技术专栏花落的技术专栏

一. IO操作

凡是'在内存中存在的数据交换的操作'都可以认为是IO操作，如：

内存和磁盘的交互：read write

内存和终端的交互：print input

内存和网络的交互：recv send

1.1 阻塞IO

默认形态，效率很低的一种IO；常见的阻塞场景：

因为某种条件没有达到造成的阻塞，如：input accept recv

处理IO事件的时间消耗较长带来的阻塞，如：文件的读写过程，网络数据的发送过程

1.2 非阻塞IO

通过修改IO事件的属性，使其变为非阻塞状态，以避免条件阻塞的情况。非阻塞IO往往和循环搭配使用，这样可以不断执行部分需要执行的代码，也不影响对阻塞事件的判断。以下示例通过s.setblocking(False)设置套接字为非阻塞套接字，并处理由此产生的BlockingIOError异常：

import socket

from time import sleep,ctime

s = socket.socket(socket.AF_INET,socket.SOCK_STREAM)

s.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)

ADDR = ("127.0.0.1",8888)

s.bind(ADDR)

s.listen(5)

#设置套接字为非阻塞

s.setblocking(False)

while True:

    print(ctime(), 'waiting for connect...')

    try:

        connect, address = s.accept()

    except BlockingIOError:

        sleep(2)

        continue

    print("Connect to", address)

    while True:

        print(ctime(), 'waiting for receive...')

        try:

            data = connect.recv(1024).decode()

        except BlockingIOError:

            continue

        if not data:

            break

        print(data)

        n = connect.send(b"Receive your message!")

    connect.close()

s.close()

实现非阻塞的另一种方式是将原本阻塞的IO设置一个最长等待时间，在规定的时间达到条件则正常执行；如果过时仍未达到条件则阻塞结束。下面的示例通过s.settimeout(sec)设置套接字超时时间，并处理socket.timeout异常：

import socket

from time import sleep,ctime

s = socket.socket(socket.AF_INET,socket.SOCK_STREAM)

s.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)

ADDR = ("127.0.0.1",8888)

s.bind(ADDR)

s.listen(5)

# 设置超时阻塞时间

s.settimeout(5)

while True:

    try:

        print(ctime(), 'waiting for connect...')

        try:

            connect, address = s.accept()

        except socket.timeout:

            print(ctime(), 'connect timeout...')

            sleep(2)

            continue

        print(ctime(), 'connect to {}'.format(address))

        while True:

            try:

                data = connect.recv(1024).decode()

            except socket.timeout:

                continue

            if not data:

                break

            connect.send(b'receive your message.')

        connect.close()

    except KeyboardInterrupt:

        s.close()

        exit()

二. IO多路复用

多路复用指的是同时交给内核监控多个IO事件，当哪个IO准备就绪，就立去执行哪个IO事件。以此来形成多个IO事件都可以操作的现象，而不必逐个等待执行。因此，当程序中有多个IO事件时，使用IO多路复用可以提高程序的执行效率。python中实现IO多路复用：

select

poll

epoll

2.1 select

r,w,x = select(rlist,wlist,xlist,timeout)：向内核发起IO监控请求,阻塞等待IO事件发生。

参数说明：

rlist：被动等待处理的IO事件列表

wlist：需要主动处理的IO列表

xlist：发生异常时需要处理的IO列表

timeout：可选参数，超时时间

返回值说明：

r ： rlist中准备就绪的IO列表

w： wlist中准备就绪的IO列表

x： xlist中准备就绪的IO列表

注意事项：

IO多路复用不应该有死循环出现，使一个客户端长期占有服务端

IO多路复用是一种并发行为，但是是单进程程序，效率较高

示例：

'''select　IO多路复用

监控服服务端终端输入及socket网络套接字

提示：请在*nux系统下运行

'''

import socket

import select

import sys

SERVER = ("0.0.0.0",8888)

s = socket.socket()

s.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)

s.bind(SERVER)

s.listen(5)

i = sys.stdin

# 三个关注列表

rlist = [s,i]

wlist = []

xlist = [s,i]

while True:

    print("Waiting for Connection...")

    rt,wt,xt = select.select(rlist,wlist,xlist)

    for x in rt:

        if x == s:

            connfd,addr = x.accept()

            print("Connect to",addr)

            rlist.append(connfd)

        elif x == i:

            data = x.readline()

            wlist.append(x)

        else:

            data = x.recv(1024).decode()

            if not data:

                rlist.remove(x)

                x.close()

            else:

                print(data)

                wlist.append(x)

    for x in wt:

        if x == i:

            data_l = ['From terminal:\n',data]

        else:

            x.send(b"Receive your message!")

            data_l = ['From network:\n',data+'\n']

        wlist.remove(x)

        with open("记录.txt",'at') as f:

            f.writelines(data_l)

s.close()

2.2 poll

p = poll() 创建poll对象

常见的poll IO事件分类

POLLIN，被动等待处理的IO

POLLOUT，主动处理的IO

POLLERR，相当于xlist

使用按位或连接注册多种IO事件：p.register(s,POLLIN | POLLERR)

取消对IO的关注：p.unregister(s)

进行监控

events = p.poll()，监控关注的IO，阻塞等待IO发生

返回值：events是一个列表，列表中每个元素为一个元组：

格式：[ (fileno, event), (), ()... ]

fileno 就绪事件的文件描述符，event就绪的事件

因为要获取IO对象以调用函数，需要建立比照字典 {s.fileno(): s}，通过fileno来获取对象；

import socket

import select

ADDR = ("0.0.0.0",8888)

s = socket.socket()

s.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)

s.bind(ADDR)

s.listen(5)

#创建poll对象

p = select.poll()

#建立通过fileno文件描述符查找套接字的字典

fdmap = {s.fileno():s}

#注册关注的IO事件

p.register(s,select.POLLIN | select.POLLERR)

while True:

    print("Waiting for connection...")

    #开始监测

    events = p.poll()

    for fd,event in events:

        if fd == s.fileno():

            connfd,addr = s.accept()

            print("Connect from:",addr)

            p.register(connfd,select.POLLIN | select.POLLERR)

            fdmap[connfd.fileno()] = connfd

        elif event & select.POLLIN:

            data = fdmap[fd].recv(1024)

            if not data:

                p.unregister(fd)

                fdmap[fd].close()

                del fdmap[fd]

                continue

            print(data.decode())

            fdmap[fd].send(b"Receive your message!")

s.close()

2.3 epoll

使用方法与poll基本相同，生成对象使用epoll()而不是poll()，register注册IO事件类型改为EPOLL事件类型：

import socket

import select

ADDR = ("0.0.0.0",8888)

s = socket.socket()

s.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)

s.bind(ADDR)

s.listen(5)

p = select.epoll()

fdmap = {s.fileno():s}

p.register(s, select.EPOLLIN | select.EPOLLERR)

while True:

    print("Waiting for Connection...")

    events = p.poll()

    for fd,event in events:

        if fd == s.fileno():

            #检测到新的客户端即将连入

            c,addr = s.accept()

            print("Connect to",addr)

            p.register(c,select.EPOLLIN | select.EPOLLERR)

            fdmap[c.fileno()] = c

        elif event & select.EPOLLIN:

            #套接字接收准备就绪

            data = fdmap[fd].recv(1024)

            if not data:

                p.unregister(fd)

                fdmap[fd].close()

                del fdmap[fd]

                continue

            print(data.decode())

            fdmap[fd].send(b"Receive your message!")

s.close()

2.4. select poll epoll三者的区别

epoll比select和poll效率高，select和poll差不多。

EPOLL内核每次仅返回给应用层“准备就绪的IO事件”；
select和poll则内核会将所有的IO事件返回，再由应用层去筛选准备就绪的IO事件。

epoll提供了更多的触发方式，IO就绪的类别更多，如EPOLLET边缘触发。
在并发高同时连接活跃度不是很高的请看下，epoll比select好（网站或web系统中，用户请求一个页面后随时可能会关闭）；并发性不高，同时连接很活跃，select比epoll好。（比如说游戏中数据一但连接了就会一直活跃，不会中断）。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

linux

Python 实现 IO 多路复用

Python 实现 IO 多路复用

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐