浅谈Python-IO多路复用(select、poll、epoll模式)

import socket
# import threading
import select
import queue

HOST, PORT = "localhost", 8020
address = (HOST, PORT)

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(address)
server.listen(10)
print("server is listening...")

# server套接字处理连接，其余套接字处理读操作
inputs = [server, ]
outputs = []
exceptions = []
# 消息接收
msg = {}

def handle_read(readable:list):
    """处理socket新建连接及数据读入"""
    for read_socket in readable:
        if read_socket is server:
            # 新建socket连接（有新用户加入）
            sock, addr = read_socket.accept()
            print("({})：connect successfully...".format(addr))
            sock.setblocking(False)
            inputs.append(sock)
            msg[sock] = queue.Queue()
        else:
            # 已建立连接的socket有消息接收
            # 此时该socket实例已被添加，直接收数据
            data = read_socket.recv(1024)
            if data:
                print("({0}) message: {1}".format(read_socket.getpeername(), data.decode("utf8")))
                # 将消息压入消息队列中
                msg[read_socket].put(data)
                if read_socket not in outputs:
                    outputs.append(read_socket)
            else:
                # socket断开连接
                print("({0})：close successfully...".format(read_socket.getpeername()))

                # 清空消息发送队列，以及输入输出队列
                inputs.remove(read_socket)
                if read_socket in outputs:
                    outputs.remove(read_socket)
                read_socket.close()
                del msg[read_socket]


def handle_write(writable: list):
    """处理socket消息发送"""
    for write_socket in writable:
        # get_nowait可能出现queue.Empty异常
        try:
            cur_writable_queue = msg.get(write_socket, None)
            if cur_writable_queue:
                # 有消息则却出消息并转发
                cur_w_data = cur_writable_queue.get_nowait()
                write_socket.send(cur_w_data)
            else:
                # 没有消息，则退出
                outputs.remove(write_socket)
        except queue.Empty:
            pass


def handle_exception(exceptional:list):
    """处理异常"""
    for e in exceptional:
        print("({0}) connect failed...".format(e.getpeername()))
        inputs.remove(e)
        if e in outputs:
            outputs.remove(e)
        if msg.get(e, None):
            del msg[e]
        e.close()


# server存在则循环监听, 事件循环的方式
while inputs:
    # 开启select监听
    readable, writable, exceptional = select.select(inputs, outputs, exceptions)
    handle_read(readable)
    handle_write(writable)
    handle_exception(exceptional)

import socket

client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.setblocking(0)

try:
    client.connect(("localhost", 8020))
except BlockingIOError:
    pass

while True:
    response = input("回复服务器：").encode("utf8")
    client.send(response)
    if response=="exit":
        break

    # 非阻塞I/O轮询方式
    while True:
        try:
            data = client.recv(1024)
        except BlockingIOError as e:
            pass
        else:
            if data:
                data = data.decode("utf8")
                break

    print("收到来自服务器的消息：%s" % data)

client.close()

from selectors import DefaultSelector, EVENT_READ, EVENT_WRITE
from urllib.parse import urlparse
import socket

selector = DefaultSelector()
urls = []
stop = False

class HTTPSelector(object):
    """使用select或epoll完成http请求"""

    def __init__(self, url):
        self.url = url
        self.domain = urlparse(url).netloc
        self.path = urlparse(url).path
        self.data = b""
        urls.append(self.url)
        if self.path == "":
            self.path = "/"

        self.client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

        # 设置为非阻塞
        self.client.setblocking(0)

        try:
            self.client.connect((self.domain, 80))
        except BlockingIOError:
            pass

        # 注册写事件
        selector.register(self.client.fileno(), EVENT_WRITE, self.connect)


    def connect(self, key):
        """连接http服务器"""

        # 解除注册写事件, 如果未解除则出现异常
        selector.unregister(key.fd)
        request_data = """GET {0} HTTP/1.1
Host: {1}
Connection: close

""".format(self.path, self.domain).encode("utf8")
        self.client.send(request_data)
        # 注册读事件
        selector.register(self.client.fileno(), EVENT_READ, self.read)

    def read(self, key):
        """接收http响应"""
        data = b""
        # 这里没有使用循环读取响应数据，原因在于select仅处理socket文件描述符状态发生变化
        # 的socket实例，此外，该程序只有一个client实例，所以其接收到的数据是属于整个HTML数据的一部分，
        # 就需要数据累加
        # while 1:
        #     try:
        #         cur_data = self.client.recv(1024)
        #     except BlockingIOError as e:
        #         pass
        #     else:
        #         if cur_data:
        #             data += cur_data
        #         else:
        #             break
        cur_data = self.client.recv(1024)
        if cur_data:
            self.data += cur_data
        else:
            selector.unregister(key.fd)
            data = data.decode("utf8")
            html_data = data.split("

")[1]
            print(html_data)
            urls.remove(self.url)
            if not urls:
                global stop
                stop = True
            self.client.close()

def loop():
    # 1.selector本身不支持register模式
    # 需要手动开启事件循环，需要由程序员自己来完成
    while not stop:
        ready = selector.select()
        for key, mask in ready:
            call_back = key.data
            call_back(key)

if __name__ == '__main__':
    test = HTTPSelector("https://www.baidu.com")
    loop()