Python：select 库高级用法举例和应用详解

import socket
import select

# 创建一个TCP/IP套接字
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(('localhost', 12345))  # 绑定地址和端口
server_socket.listen(5)  # 开始监听，允许最多5个连接
print("服务器已启动，等待连接...")

# 用列表存储所有连接的套接字
sockets_list = [server_socket]

while True:
    # 使用select监听可读的套接字
    read_sockets, _, _ = select.select(sockets_list, [], [])
    
    for notified_socket in read_sockets:
        # 如果是服务器套接字，表示有新连接
        if notified_socket == server_socket:
            client_socket, client_address = server_socket.accept()  # 接收新连接
            sockets_list.append(client_socket)  # 将新连接添加到监控列表
            print(f"新连接：{client_address} 已连接")
        else:
            message = notified_socket.recv(1024)  # 接收消息
            if not message:
                sockets_list.remove(notified_socket)  # 移除关闭的连接
                print("连接已关闭")
                continue
            print(f"收到消息：{message.decode('utf-8')}")

import select
import sys

# 创建一个文件描述符
file = open('example.txt', 'r')
print("开始读取文件，按Ctrl+C退出")

while True:
    # 使用select监听文件的可读事件
    readable, _, _ = select.select([file, sys.stdin], [], [])
    
    for r in readable:
        if r == file:
            line = file.readline()  # 从文件读取一行
            if not line:
                print("文件读取完毕")
                break
            print(f"读取到：{line.strip()}")
        elif r == sys.stdin:
            input()  # 等待用户输入
            print("用户输入，继续执行")

import socket
import select

# 创建 UDP 套接字
udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
udp_socket.bind(('localhost', 12346))  # 绑定地址和端口
print("UDP服务器已启动，等待请求...")

while True:
    # 使用select监听可读的套接字
    ready_to_read, _, _ = select.select([udp_socket], [], [])
    
    for s in ready_to_read:
        data, address = s.recvfrom(1024)  # 接收数据
        print(f"来自 {address} 的数据：{data.decode('utf-8')}")
        s.sendto(b'已收到消息', address)  # 回复客户端