Python中IO多路复用模块selector的用法详解

目录

• selector 简介
• 阻塞IO模型下的 socket 网络编程
• IO多路复用模型下的socket 网络编程 selector
• selector 原理分析
• 返回就绪状态文件句柄
• 数据拷贝
• asyncio  和  selector 的关系
• 附录asyncio模块事件循环核心模块

import socket

# 创建TCP套接字
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 绑定IP地址和端口号
server_address = ('0.0.0.0', 12346)
server_socket.bind(server_address)

# 监听连接
server_socket.listen()

# 接受客户端连接请求
print("服务器已启动，等待客户端连接...")
client_socket, client_address = server_socket.accept()
print(f"与客户端 {client_address} 建立连接")

# 向客户端发送消息
message = "欢迎连接到服务器！"
client_socket.sendall(message.encode())

while True:

    # 从客户端接收消息
    data = client_socket.recv(1024).decode()
    print(f"客户端消息：{data}")
    client_socket.sendall(f"服务器收到消息：{data}".encode())
    if data == "close":
        # 关闭客户端套接字
        client_socket.close()

import socket

# 创建TCP套接字
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 服务器地址和端口号
server_address = ('localhost', 12346)

# 连接服务器
client_socket.connect(server_address)

# 接收服务器的消息
data = client_socket.recv(1024).decode()
print(f"已连接到服务器, 服务器消息：{data}")

while True:
    # 向服务器发送消息
    message = input()
    if message == "end":
        break
    client_socket.sendall(message.encode())
    print(client_socket.recv(1024).decode())

# 关闭客户端套接字
client_socket.close()

import selectors
import socket

# 选择一个当前平台最优的IO多路复用模型
sel = selectors.DefaultSelector()


def accept(server_socket):
    conn, addr = server_socket.accept()
    print(f"与客户端 {addr} 建立连接")

    conn.setblocking(False)  # 设定非阻塞
    # 注册conn对象到selector中，当conn可读时，返回conn和回调函数read
    sel.register(conn, selectors.EVENT_READ, read)


def read(conn):
    data = conn.recv(1024).decode('utf-8')
    print(f"客户端消息：{data}")
    if data == "close":
        sel.unregister(conn)
        conn.close()
    else:
        conn.sendall(f"服务器收到消息：{data}".encode())


if __name__ == "__main__":
    sock = socket.socket()
    sock.bind(("0.0.0.0", 9999))
    sock.listen()
    sock.setblocking(False)  # 设置sock非阻塞
    # 将sock注册到selector中，当sock可读时，返回sock和回调函数accept
    sel.register(sock, selectors.EVENT_READ, accept)

    print("创建事件循环")
    while True:
        events = sel.select()  # 阻塞运行，有就绪的事件返回就绪事件列表
        for key, _ in events:
            print(key)
            # key.data: 注册的回调函数  key.fileobj: 注册的文件句柄
            callback = key.data  # 注册的回调函数
            callback(key.fileobj)

register(fileobj, events, data=None)

select(timeout=None)

import socket

client = socket.socket(family=socket. AF_INET, type=socket.SOCK_STREAM)
host = socket.gethostname()
client.connect((host, 9999))

while True:
    data = input("客户端发送数据：").strip()
    client.send(data.encode())
    if data == "end" or data == "":
        client.close()
        break
    print(client.recv(1024).decode("utf-8"))

if __name__ == "__main__":
    sock = socket.socket()
    sock.bind(("0.0.0.0", 9999))
    sock.listen()
    sock.setblocking(False)  # 设置sock非阻塞
    # 将sock注册到selector中，当sock可读时，返回sock和回调函数accept
    sel.register(sock, selectors.EVENT_READ, accept)

    print("创建事件循环")
    while True:
        events = sel.select()  # 阻塞运行，有就绪的事件返回就绪事件列表

def accept(server_socket):
    conn, addr = server_socket.accept()
    print(f"与客户端 {addr} 建立连接")

    conn.setblocking(False)  # 设定非阻塞
    # 注册conn对象到selector中，当conn可读时，返回conn和回调函数read
    sel.register(conn, selectors.EVENT_READ, read)

def read(conn):
    data = conn.recv(1024).decode('utf-8')
    print(f"客户端消息：{data}")
    if data == "close":
        sel.unregister(conn)
        conn.close()
    else:
        conn.sendall(f"服务器收到消息：{data}".encode())

sel = selectors.DefaultSelector()
while True:
    events = sel.select()

def run_forever(self):
    """Run until stop() is called."""
    self._check_closed()
    self._check_running()
    self._set_coroutine_origin_tracking(self._debug)

    old_agen_hooks = sys.get_asyncgen_hooks()
    try:
        self._thread_id = threading.get_ident()
        sys.set_asyncgen_hooks(firstiter=self._asyncgen_firstiter_hook,
                               finalizer=self._asyncgen_finalizer_hook)

        events._set_running_loop(self)
        while True:
            self._run_once()
            if self._stopping:
                break
    finally:
        self._stopping = False
        self._thread_id = None
        events._set_running_loop(None)
        self._set_coroutine_origin_tracking(False)
        sys.set_asyncgen_hooks(*old_agen_hooks)