Python中如何使用socket模块持续监听传入的套接字？

当然可以用Python的socket模块实现持续监听传入连接，每次有新连接接入时直接触发指定代码逻辑，这是网络编程中非常基础的场景。

核心实现思路 #

• 创建套接字对象并绑定到指定的IP和端口
• 调用listen()方法启动监听模式
• 通过无限循环调用accept()（该方法会阻塞程序，直到有新连接接入）
• 拿到新连接的套接字和客户端地址后，执行你的自定义处理代码
• 如需同时处理多个客户端，可结合多线程/多进程实现并发处理

单线程基础示例（一次处理一个连接） #

适合不需要并发的简单场景：

import socket

def handle_new_connection(conn, client_addr):
    # 这里替换成你要触发的业务代码
    print(f"收到新连接：{client_addr}")
    # 示例：接收客户端消息并回复
    received_data = conn.recv(1024)
    if received_data:
        print(f"来自 {client_addr} 的数据：{received_data.decode('utf-8')}")
        conn.sendall(b"已成功接收你的消息")
    # 处理完成后关闭连接
    conn.close()

def start_listener():
    listen_host = '0.0.0.0'  # 监听所有网络接口（允许外部设备连接）
    listen_port = 65432       # 自定义未被占用的端口

    # 创建TCP套接字
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server_socket:
        server_socket.bind((listen_host, listen_port))
        server_socket.listen()
        print(f"正在监听 {listen_host}:{listen_port}...")
        while True:
            # 阻塞等待新连接
            conn, addr = server_socket.accept()
            # 触发自定义处理逻辑
            handle_new_connection(conn, addr)

if __name__ == "__main__":
    start_listener()

多线程并发示例（同时处理多个连接） #

如果需要同时处理多个客户端连接，用线程来拆分每个连接的处理逻辑：

import socket
import threading

def handle_client(conn, client_addr):
    print(f"新连接接入：{client_addr}")
    try:
        # 自定义处理逻辑
        data = conn.recv(1024)
        if data:
            print(f"{client_addr} 发送：{data.decode('utf-8')}")
            conn.sendall(b"你的请求已处理")
    finally:
        # 确保连接被关闭
        conn.close()
        print(f"{client_addr} 连接已断开")

def start_concurrent_listener():
    listen_host = '0.0.0.0'
    listen_port = 65432

    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server_socket:
        server_socket.bind((listen_host, listen_port))
        server_socket.listen(5)  # 设置挂起连接的最大队列数
        print(f"并发监听启动：{listen_host}:{listen_port}")
        while True:
            conn, addr = server_socket.accept()
            # 为每个新连接创建独立线程处理
            thread = threading.Thread(target=handle_client, args=(conn, addr))
            thread.start()

if __name__ == "__main__":
    start_concurrent_listener()

注意事项 #

• 若仅需本地测试，可将listen_host设为127.0.0.1，仅允许本地程序连接
• 端口号需选择大于1024且未被系统或其他程序占用的数值
• UDP协议的监听逻辑类似，只需将socket.SOCK_STREAM改为socket.SOCK_DGRAM，并使用recvfrom()接收数据（UDP无连接，无需accept()）

内容的提问来源于stack exchange，提问作者guyguy4040