在阻塞式socket.recv()调用期间优雅关闭UDP套接字

这个问题我之前做UDP服务的时候也碰到过——阻塞式的recv()调用真的很棘手，因为线程卡在系统调用里的时候，根本不会去检查你的done标志位。下面给你几个完全不用处理冗余错误的优雅解决方案，按需选就行：

方案一：给套接字设置超时，定期检查终止标志 #

给UDP套接字设置一个合理的超时时间（比如1秒），这样recv()会每隔一段时间自动从阻塞状态退出（抛出超时异常），这时线程就有机会检查done变量是否为True，从而优雅退出。虽然会触发超时异常，但我们可以干净地捕获它，逻辑非常清晰。

import socket
import threading

done = False

def run_sock():
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.bind(('localhost', 12345))
    sock.settimeout(1)  # 设置1秒超时，让线程定期醒来检查终止标志
    while not done:
        try:
            data = sock.recv(1500)
            print(data)
        except socket.timeout:
            # 超时只是用来唤醒线程，啥也不用做，直接回到循环开头检查done
            continue
    sock.close()

thread = threading.Thread(target=run_sock, daemon=True)
thread.start()

# 主线程业务逻辑
while not done:
    # 模拟你的业务操作
    import time
    time.sleep(3)
    # 满足终止条件时设置done
    done = True

thread.join()

方案二：向自身套接字发送唤醒数据包 #

这是我最常用的方案——当主线程想要终止时，直接创建一个临时UDP套接字，向绑定的地址发送一个空包（或者特定标识的包），这样阻塞的recv()会立即收到这个包并返回，线程自然会检查done标志然后退出。全程不需要处理任何异常，非常干净。

import socket
import threading

done = False

def run_sock():
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.bind(('localhost', 12345))
    while not done:
        data = sock.recv(1500)
        print(data)
    sock.close()

thread = threading.Thread(target=run_sock, daemon=True)
thread.start()

# 主线程业务逻辑
while not done:
    # 模拟你的业务操作
    import time
    time.sleep(3)
    # 满足终止条件时设置done，并发送唤醒包
    done = True
    # 用临时套接字向自身发送空包，唤醒阻塞的recv
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as wake_sock:
        wake_sock.sendto(b'', ('localhost', 12345))

thread.join()

如果你的业务需要过滤掉这个唤醒包，可以发送一个特定的标识（比如b'__TERMINATE__'），然后在recv()后判断：如果是终止包就直接跳出循环，不过单纯为了终止的话，空包完全够用。

方案三：用select实现多路监听（适合复杂场景） #

如果你的线程需要同时处理多个IO事件，用select.select()同时监听UDP套接字和一个终止信号管道是更专业的做法。主线程要终止时，向管道写入数据，select会立即返回，线程就能及时响应终止请求。

import socket
import threading
import select
import os

done = False
# 创建一对匿名管道，用于传递终止信号
read_pipe, write_pipe = os.pipe()

def run_sock():
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.bind(('localhost', 12345))
    while not done:
        # 监听套接字和读管道的可读事件
        readable, _, _ = select.select([sock, read_pipe], [], [], None)
        for fd in readable:
            if fd == sock:
                data = sock.recv(1500)
                print(data)
            elif fd == read_pipe:
                # 收到终止信号，读取管道内容避免阻塞
                os.read(read_pipe, 1)
                done = True
                break
    sock.close()
    os.close(read_pipe)
    os.close(write_pipe)

thread = threading.Thread(target=run_sock, daemon=True)
thread.start()

# 主线程业务逻辑
while not done:
    # 模拟你的业务操作
    import time
    time.sleep(3)
    # 满足终止条件时设置done，并向管道写入信号
    done = True
    os.write(write_pipe, b'1')

thread.join()

总结 #

• 简单场景优先选方案二，代码最简洁，无异常处理，直接唤醒阻塞的recv()；
• 如果不想发送额外数据包，选方案一，超时检查逻辑清晰；
• 多IO事件的复杂场景用方案三，扩展性更好。

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