Python Socket服务器：高效处理多客户端连接并等待全部完成信号

import socket
import select
import logging
import sys

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def start_select_server(ip, port, expected_clients_n):
    """
    启动一个基于select的socket服务器，等待指定数量的客户端发送'complete'消息。

    Args:
        ip (str): 服务器绑定的IP地址。
        port (int): 服务器监听的端口。
        expected_clients_n (int): 预期会发送'complete'消息的客户端总数。
    """
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setblocking(False) # 设置为非阻塞模式，select会处理阻塞
    server.bind((ip, port))
    server.listen(5) # 最多允许5个挂起连接

    inputs = [server] # 初始时，只监控服务器自身的socket
    complete_count = 0 # 记录发送'complete'消息的客户端数量

    logging.info(f"服务器启动，监听于 {ip}:{port}，等待 {expected_clients_n} 个客户端完成。")

    try:
        while True:
            # 使用select监控可读、可写和异常事件。timeout=10表示每10秒检查一次
            readable, _, exceptional = select.select(inputs, [], inputs, 10)

            # 如果在timeout时间内没有任何事件发生
            if not (readable or exceptional):
                logging.info("等待客户端连接或数据中 (超时)...")
                continue

            for s in readable:
                if s is server:
                    # 服务器socket可读，表示有新的客户端连接请求
                    conn, addr = s.accept()
                    conn.setblocking(False) # 新连接也设置为非阻塞
                    inputs.append(conn) # 将新连接添加到监控列表
                    logging.info(f"接受新连接：{addr}")
                else:
                    # 客户端socket可读，表示有数据到达
                    try:
                        data = s.recv(1024).decode('utf8').strip()
                        if data:
                            logging.info(f"收到来自 {s.getpeername()} 的数据: {data}")
                            if data == 'complete':
                                complete_count += 1
                                logging.info(f"客户端 {s.getpeername()} 发送'complete'。当前完成数: {complete_count}/{expected_clients_n}")
                                # 客户端完成任务后，从监控列表中移除并关闭其socket
                                inputs.remove(s)
                                s.close()
                        else:
                            # 客户端断开连接（收到空数据）
                            logging.info(f"客户端 {s.getpeername()} 断开连接。")
                            inputs.remove(s)
                            s.close()
                    except ConnectionResetError:
                        # 客户端突然关闭连接，未发送FIN包
                        logging.warning(f"客户端 {s.getpeername()} 强制断开连接。")
                        inputs.remove(s)
                        s.close()
                    except Exception as e:
                        logging.error(f"处理客户端 {s.getpeername()} 数据时发生错误: {e}")
                        inputs.remove(s)
                        s.close()

            for s in exceptional:
                # 处理异常情况，例如客户端socket出现错误
                logging.error(f"处理客户端 {s.getpeername()} 异常。")
                inputs.remove(s)
                s.close()

            # 检查是否所有客户端都已发送'complete'消息
            if complete_count >= expected_clients_n:
                logging.info(f"所有 {expected_clients_n} 个客户端已完成任务。服务器即将关闭。")
                break

    except KeyboardInterrupt:
        logging.info("服务器被用户中断。")
    except Exception as e:
        logging.error(f"服务器运行过程中发生错误: {e}")
    finally:
        # 清理所有打开的socket
        for s in inputs:
            s.close()
        server.close()
        logging.info("服务器已关闭。")

# 客户端模拟代码 (仅供测试，可独立运行)
def simulate_client(ip, port, messages):
    """
    模拟一个客户端连接服务器并发送消息。
    """
    try:
        client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        client_socket.connect((ip, port))
        logging.info(f"客户端连接到 {ip}:{port}")
        for msg in messages:
            client_socket.sendall(msg.encode('utf8'))
            logging.info(f"客户端发送: {msg}")
            # 模拟消息发送间隔
            import time
            time.sleep(0.1)
        client_socket.close()
        logging.info("客户端关闭连接。")
    except Exception as e:
        logging.error(f"客户端发生错误: {e}")

if __name__ == "__main__":
    SERVER_IP = '127.0.0.1'
    SERVER_PORT = 12345
    EXPECTED_CLIENTS = 3 # 假设我们预期有3个客户端会发送'complete'

    # 启动服务器在一个单独的线程中，以便主线程可以启动客户端
    import threading
    server_thread = threading.Thread(target=start_select_server, args=(SERVER_IP, SERVER_PORT, EXPECTED_CLIENTS))
    server_thread.start()

    # 稍等片刻，确保服务器启动
    import time
    time.sleep(1)

    # 模拟不同类型的客户端
    # 客户端1：发送多条消息，最后发送'complete'
    client1_messages = ['Hello from client 1', 'Data piece A', 'Data piece B', 'complete']
    client1_thread = threading.Thread(target=simulate_client, args=(SERVER_IP, SERVER_PORT, client1_messages))
    client1_thread.start()

    # 客户端2：发送一条消息，然后发送'complete'
    client2_messages = ['Just a quick message', 'complete']
    client2_thread = threading.Thread(target=simulate_client, args=(SERVER_IP, SERVER_PORT, client2_messages))
    client2_thread.start()

    # 客户端3：发送一条消息，然后发送'complete'
    client3_messages = ['Another client here', 'complete']
    client3_thread = threading.Thread(target=simulate_client, args=(SERVER_IP, SERVER_PORT, client3_messages))
    client3_thread.start()

    # 等待所有客户端线程完成
    client1_thread.join()
    client2_thread.join()
    client3_thread.join()

    # 等待服务器线程完成
    server_thread.join()
    logging.info("所有客户端和服务器任务已完成。")