处理这种异步中断的好方法是旧的 C 管道技巧。您可以创建一个管道并在套接字和管道上使用select/poll：现在，当您需要中断接收器时，您只需向管道发送一个字符。
- 优点：
- 缺点：
- 管道上的选择/轮询在 Windows 上不可用，在这种情况下，您应该将其替换为另一个用作通知的 UDP 套接字管道
起点
interruptable_socket.py
import os
import socket
import select
class InterruptableUdpSocketReceiver(object):
def __init__(self, host, port):
self._host = host
self._port = port
self._socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self._r_pipe, self._w_pipe = os.pipe()
self._interrupted = False
def bind(self):
self._socket.bind((self._host, self._port))
def recv(self, buffersize, flags=0):
if self._interrupted:
raise RuntimeError("Cannot be reused")
read, _w, errors = select.select([self._r_pipe, self._socket], [], [self._socket])
if self._socket in read:
return self._socket.recv(buffersize, flags)
return ""
def interrupt(self):
self._interrupted = True
os.write(self._w_pipe, "I".encode())
一个测试套件：
test_interruptable_socket.py
import socket
from threading import Timer
import time
from interruptable_socket import InterruptableUdpSocketReceiver
import unittest
class Sender(object):
def __init__(self, destination_host, destination_port):
self._socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
self._dest = (destination_host, destination_port)
def send(self, message):
self._socket.sendto(message, self._dest)
class Test(unittest.TestCase):
def create_receiver(self, host="127.0.0.1", port=3010):
receiver = InterruptableUdpSocketReceiver(host, port)
receiver.bind()
return receiver
def create_sender(self, host="127.0.0.1", port=3010):
return Sender(host, port)
def create_sender_receiver(self, host="127.0.0.1", port=3010):
return self.create_sender(host, port), self.create_receiver(host, port)
def test_create(self):
self.create_receiver()
def test_recv_async(self):
sender, receiver = self.create_sender_receiver()
start = time.time()
send_message = "TEST".encode('UTF-8')
Timer(0.1, sender.send, (send_message, )).start()
message = receiver.recv(128)
elapsed = time.time()-start
self.assertGreaterEqual(elapsed, 0.095)
self.assertLess(elapsed, 0.11)
self.assertEqual(message, send_message)
def test_interrupt_async(self):
receiver = self.create_receiver()
start = time.time()
Timer(0.1, receiver.interrupt).start()
message = receiver.recv(128)
elapsed = time.time()-start
self.assertGreaterEqual(elapsed, 0.095)
self.assertLess(elapsed, 0.11)
self.assertEqual(0, len(message))
def test_exception_after_interrupt(self):
sender, receiver = self.create_sender_receiver()
receiver.interrupt()
with self.assertRaises(RuntimeError):
receiver.recv(128)
if __name__ == '__main__':
unittest.main()
进化
现在这段代码只是一个起点。为了使其更通用，我认为我们应该解决以下问题：
-
接口：中断情况下返回空消息不好处理，最好使用异常处理
-
泛化：我们应该在
socket.recv()之前只调用一个函数，将中断扩展到其他recv方法变得非常简单
-
可移植性：为了简单地将其移植到 Windows，我们应该将异步通知隔离在一个对象中，以便为我们的操作系统选择正确的实现
首先我们更改test_interrupt_async() 以检查异常而不是空消息：
from interruptable_socket import InterruptException
def test_interrupt_async(self):
receiver = self.create_receiver()
start = time.time()
with self.assertRaises(InterruptException):
Timer(0.1, receiver.interrupt).start()
receiver.recv(128)
elapsed = time.time()-start
self.assertGreaterEqual(elapsed, 0.095)
self.assertLess(elapsed, 0.11)
在此之后，我们可以将return '' 替换为raise InterruptException 并且测试再次通过。
准备好扩展的版本可以是：
interruptable_socket.py
import os
import socket
import select
class InterruptException(Exception):
pass
class InterruptableUdpSocketReceiver(object):
def __init__(self, host, port):
self._host = host
self._port = port
self._socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self._async_interrupt = AsycInterrupt(self._socket)
def bind(self):
self._socket.bind((self._host, self._port))
def recv(self, buffersize, flags=0):
self._async_interrupt.wait_for_receive()
return self._socket.recv(buffersize, flags)
def interrupt(self):
self._async_interrupt.interrupt()
class AsycInterrupt(object):
def __init__(self, descriptor):
self._read, self._write = os.pipe()
self._interrupted = False
self._descriptor = descriptor
def interrupt(self):
self._interrupted = True
self._notify()
def wait_for_receive(self):
if self._interrupted:
raise RuntimeError("Cannot be reused")
read, _w, errors = select.select([self._read, self._descriptor], [], [self._descriptor])
if self._descriptor not in read:
raise InterruptException
def _notify(self):
os.write(self._write, "I".encode())
现在包装更多recv函数，实现windows版本或处理socket超时变得非常简单。