Python中Socketselect基本使用介绍

目录

• 一. 前言
• 二. 普通直连和使用select连接
• 1. 直接连接方式：
• 2. select连接方式：
• 二. Select介绍
• 三个函数的简要介绍：
• 在使用 select 模块时，需要注意以下几点：
• 三. 代码示例
• 服务端Socket
• 使用线程启动server
• 客户端Socket
• 四. 多个服务端和多个客户端使用select代码示例
• 服务端
• 客户端
• 五. 总结

在Python中，select 是一个用于异步I/O多路复用的模块。它提供了一种简单的方法，用于监视多个文件描述符（file descriptor），以确定其中哪些文件描述符已经就绪可读、可写或者发生了异常。使用 select 模块可以实现一些高效的网络编程，比如在服务端同时监听多个客户端连接，或者在客户端同时连接多个服务端。

使用select和普通的直接连接方式都可以实现连接，但二者有以下区别：

在发送数据前，需要先建立连接，然后通过send函数发送数据。这种方式只能同时处理一个连接，无法同时处理多个连接。

使用select可以监听多个文件描述符，当其中的任意一个文件描述符发生读写事件时，select函数就会返回并通知对应的程序进行处理。这种方式可以同时处理多个连接，提高了系统的并发性能。

select 模块暴露了3个主要的函数，分别是 select.select()、select.poll() 和 select.epoll()，它们可以实现不同的多路复用机制。

• select.select(rlist, wlist, xlist, timeout)： 用于监视文件描述符的变化，可以监视读、写、异常事件。当这些事件中的任何一个发生时，select函数会返回。rlist、wlist和xlist分别是要监视的可读、可写和异常的文件描述符列表，timeout是超时参数，单位是秒。
• select.poll()：使用场景跟 select.select() 相同，但性能更好，适用于监视较大量的文件描述符。
• select.epoll()： Linux系统下的I/O复用机制，也是一种性能很好的多路复用机制。它相比于 select.poll()
  的优点主要在于它可以支持更多的连接数。

除了以上三种函数，还有 select.kevent() 和 select.kqueue() 函数，它们适用于FreeBSD系统。

• 最好使用非阻塞的 socket，以避免程序在等待 socket 数据时被阻塞，从而可以处理其它 socket 数据。
• select 函数可能会有一些性能问题，当需要同时监听大量的文件描述符时，可能会导致 CPU 占用过高，所以使用时需要注意调优。

下面给出一个简单的示例，演示如何使用 select 模块同时监听多个 socket：

import socket
import select

# 设置需要监听的 socket 地址和端口
ADDRESS = ("localhost", 9000)

# 创建一个服务器 socket，并绑定到指定地址和端口
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(ADDRESS)
server_socket.listen(5)

# 将 server_socket 设置为非阻塞模式
server_socket.setblocking(False)

# 存储已连接的 client_socket
connected_clients = []

# 开始监听
while True:
    # 使用 select 函数监听所有连接的 client_socket
    readable_sockets, _, _ = select.select([server_socket] + connected_clients,
                                           [], [], 1)

    # 处理所有可读的 socket
    for sock in readable_sockets:
        # 如果是 server_socket 表示有新的连接
        if sock is server_socket:
            client_socket, client_address = server_socket.accept()
            connected_clients.append(client_socket)
            print(f"New client connected: {client_address}")
        # 否则是已连接的 client_socket，需要处理收到的数据
        else:
            try:
                data = sock.recv(1024)
                if data:
                    print(f"Received data from {sock.getpeername()}: {data.decode()}")
                else:
                    # 如果收到的数据为空，表示连接已经断开，需要关闭 socket 并从 connected_clients 中移除
                    sock.close()
                    print(f"Client {sock.getpeername()} disconnected")
                    connected_clients.remove(sock)
            except Exception as e:
                # 出现异常，也需要关闭 socket 并从 connected_clients 中移除
                print(f"Error occurred while receiving data from {sock.getpeername()}: {e}")
                sock.close()
                connected_clients.remove(sock)

在这个示例中，我们使用 select 模块同时监听 server_socket 和所有 connected_clients，当有新的 client_socket 连接时，会将其添加到 connected_clients 列表中；当存在可读的 socket 时，会根据是 server_socket 还是 client_socket 处理它们的相关操作。

需要注意的是，在处理已断开连接的 client_socket 时，需要将其关闭并从 connected_clients 中移除，否则会一直存在于 connected_clients 中，导致程序出现意料之外的错误。

import logging
import socket
from threading import Thread

import select


class SimSocketServer(Thread):
    def __init__(self, addr, logger=None):
        Thread.__init__(self, name="SimSocketServer")
        self.connected_clients = []
        self.addr = addr
        self.seqNo = 1
        self.socket_server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.logger = logger
        self.jsonTemplate = {
            "Command": "FORWARD_ELEV_INFO",
            "DeviceId": "C0002T",
        }

    def run(self):
        self.logger.info('-----------------------------------------------------------')
        self.logger.info("HC SimBattery Server addr:{} started listen...".format(self.addr))
        self.logger.info('-----------------------------------------------------------')
        # 创建一个服务器 socket，并绑定到指定地址和端口
        self.socket_server.bind(self.addr)
        self.socket_server.listen(5)
        # 将 socket_server 设置为非阻塞模式
        self.socket_server.setblocking(False)

        # 存储已连接的 client_socket
        self.connected_clients = []

        while True:
            # 使用 select 函数监听所有连接的 client_socket
            readable_sockets, _, _ = select.select([self.socket_server] + self.connected_clients, [], [], 1)

            # 处理所有可读的 socket
            for sock in readable_sockets:
                # 如果是 socket_server 表示有新的连接
                if sock is self.socket_server:
                    client_socket, client_address = self.socket_server.accept()
                    self.connected_clients.append(client_socket)
                    self.logger.info(f"New client connected: {client_address}")
                # 否则是已连接的 client_socket，需要处理收到的数据
                else:
                    try:
                        data = sock.recv(1024)
                        if not data:
                            # 如果收到的数据为空，表示连接已经断开，需要关闭 socket 并从 connected_clients 中移除
                            sock.close()
                            self.logger.info(f"Client {sock.getpeername()} disconnected")
                            self.connected_clients.remove(sock)
                            continue
                        print(f"Received data from {sock.getpeername()}: {data.decode()}")

                        # 发送消息到所有连接的客户端（除了发送者）
                        # for client in self.connected_clients:
                        #     if client != sock:
                        #         client.send(data)
                    except Exception as e:
                        # 出现异常，也需要关闭 socket 并从 connected_clients 中移除
                        self.logger.warn(f"Error occurred while receiving data from {sock.getpeername()}: {e}")
                        sock.close()
                        self.connected_clients.remove(sock)

    def send_msg2client(self, msg):
        try:
            # 发送消息到所有连接的客户端
            for client in self.connected_clients:
                client.send(f'hello client -> {msg}'.encode())
        except Exception as e:
            print('send msg error:'.format(e))


if __name__ == '__main__':
    '''simulation battery socket server thread'''
    simSocketServer = SimSocketServer(("127.0.0.1", 8787), logging)
    simSocketServer.start()

    while True:
        try:
            pc_cmd = input('输入命令：n')
            print("pc_cmd", pc_cmd)
            simSocketServer.send_msg2client(pc_cmd)
        except Exception as e:
            print(f'cmd error {e}')

import socket
import select
import sys

# 创建5个socket对象，用于连接5个不同的服务端
s1 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s2 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s3 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s4 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s5 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 设置服务端地址和端口号
server1_address = ('localhost', 9000)
server2_address = ('localhost', 9001)
server3_address = ('localhost', 9002)
server4_address = ('localhost', 9003)
server5_address = ('localhost', 9004)

# 将socket对象添加到列表中
sockets = [s1, s2, s3, s4, s5]

# 连接服务端
for sock, address in zip(sockets, [server1_address, server2_address, server3_address, server4_address, server5_address]):
    try:
        sock.connect(address)
    except Exception as e:
        print(f"Exception: {e}")
        sys.exit()

while True:
    # 使用select函数监控所有socket对象
    ready_to_read, _, _ = select.select(sockets, [], [])

    # 在任何一个可读socket对象中读取数据
    for sock in ready_to_read:
        try:
            data = sock.recv(1024)
            if data:
                print(f"Received: {data}")
            else:
                socket.close()
                sockets.remove(sock)
        except Exception as e:
            sockets.remove(sock)
            print(f"Exception: {e}")
            sock.close()

我们使用select函数监视所有socket对象。如果有数据可读，我们就会读取数据并打印出来。如果在接收数据时发生异常，我们将从sockets列表中删除该socket对象并关闭它。

使用多线程启动多个服务端
下面是服务端代码的示例，它会同时监听8001, 8002和8003端口。当客户端连接成功后会向客户端发送一条欢迎消息，当客户端发送信息时会原样返回给客户端，当客户端关闭连接时，服务器也会关闭相应的socket。

import socket
import threading

# 设置服务端地址和端口号
SERVER_ADDRESS = "localhost"
SERVER_PORT = 8001

# 监听的队列大小
LISTEN_QUEUE_SIZE = 5

# 消息欢迎消息
WELCOME_MESSAGE = "Welcome to server!"

# 为每个客户端建立相应的socket连接
def handle_client(client_socket, client_address):
    print(f"New connection from {client_address}")
    client_socket.send(WELCOME_MESSAGE.encode())

    while True:
        try:
            data = client_socket.recv(1024)
            if data:
                print(f"Received message from {client_address}: {data.decode()}")
                client_socket.send(data)
            else:
                # 关闭客户端连接
                client_socket.close()
                print(f"Connection closed by {client_address}")
                break
        except Exception as e:
            print(f"Error encountered while receiving data from {client_address}: {e}")
            client_socket.close()
            break

# 监听多个socket
def listen(address, connections):
    # 创建socket连接
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_socket.bind(address)
    server_socket.listen(LISTEN_QUEUE_SIZE)

    print(f"Listening on {address[0]}:{address[1]}")
    while True:
        # 等待客户端连接
        client_socket, client_address = server_socket.accept()
        print(f"Accepted new connection from {client_address}")

        # 为客户端启动线程
        thread = threading.Thread(target=handle_client, args=(client_socket, client_address))
        thread.daemon = True
        thread.start()

        # 将客户端的socket连接保存
        connections.append(client_socket)

# 启动监听
connections = []
threads = []
for port in range(SERVER_PORT, SERVER_PORT + 3):
    address = (SERVER_ADDRESS, port)
    thread = threading.Thread(target=listen, args=(address, connections))
    threads.append(thread)
    thread.start()

# 等待所有线程结束
for thread in threads:
    thread.join()

# 关闭所有连接
for connection in connections:
    connection.close()

在这个示例代码中，我们使用了Python中的多线程来同时监听8001, 8002和8003端口的客户端连接，这个方式更加灵活和高效。handle_client()函数用于处理每个客户端连接，它会向客户端发送一条欢迎消息，并在客户端发送数据时原样返回给客户端。listen()函数用于监听一个端口并为每个客户端连接启动一个新线程处理。最后在主线程中，我们启动了三个线程分别监听不同的端口，等待所有线程结束并关闭所有连接。

下面的客户端程序可以连接多个服务端并同时监听每个服务端返回的消息。代码中处理了连接、发送和接收数据时可能出现的异常情况。

import socket
import time
import traceback

import select

# 设置服务端地址列表
SERVER_ADDRESSES = [("localhost", 8001), ("localhost", 8002), ("localhost", 8003)]


def set_socket(server_address):
    sockets = []
    for server_addr in server_address:
        client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        client_socket.setblocking(True)
        sockets.append(client_socket)
    return sockets


def start_client(server_address):
    sockets = set_socket(server_address)
    # 连接服务端
    for index, client_socket in enumerate(sockets):
        server_address = SERVER_ADDRESSES[index]
        try:
            client_socket.connect(server_address)
        except Exception as e:
            print(f"Connect to {server_address} failed: {e}")
    start_listen(sockets)


def start_listen(sockets):
    # 使用select监听服务端
    while True:
        try:
            # 仅监听已连接的socket，获取到了数据则接收数据
            readable, writeable, errors = select.select(sockets, [], sockets)

            for socket in readable:
                try:
                    data = socket.recv(1024)
                    if data:
                        print(f"Received message from {socket.getpeername()}: {data.decode()}")
                    else:
                        # 当对端关闭连接时，对应的可读socket也会被认为是可写的
                        # 并且其recv方法将返回空字节流
                        sockets.remove(socket)
                except Exception as e:
                    print(f"Error encountered while receiving data: {e}")
                    sockets.remove(socket)

            for socket in errors:
                print(f"Error encountered on {socket.getpeername()}")
                sockets.remove(socket)
        except Exception as e:
            traceback.print_exc()
            time.sleep(1)
            start_client(SERVER_ADDRESSES)
            break


if __name__ == '__main__':
    # 创建socket连接
    start_client(SERVER_ADDRESSES)

在实际应用中，select的作用和好处在于它能够使程序监听多个连接，解决了直接连接方式只能处理单个连接的问题。这种方式可以让程序在处理一个连接时，同时监听其他连接，使得程序的并发性能得到大幅提升，适用于高并发环境下的网络通信场景。

总之，select方式适用于多连接场景，可以提高程序的运行效率。而直接连接方式适用于简单的单连接场景。

到此这篇关于Python中Socket select基本使用介绍的文章就介绍到这了,更多相关Python Socket select内容请搜索风君子博客以前的文章或继续浏览下面的相关文章希望大家以后多多支持风君子博客！

您可能感兴趣的文章:

• Python网络编程使用select实现socket全双工异步通信功能示例
• Python基于select实现的socket服务器