如何从Python中的另一个线程中止socket.recv()

Question

我有一个等待连接的主线程。它生成客户端线程，该线程将回显来自客户端的响应（在本例中为 telnet）。但是假设我想在一段时间后关闭所有套接字和所有线程，例如在 1 个连接之后。

我该怎么做？如果我从主线程执行

clientSocket.close()

，它不会停止执行

recv

。仅当我首先通过 telnet 发送某些内容时它才会停止，然后它将无法执行进一步的发送和接收。

我的代码如下所示：

# Echo server program
import socket
from threading import Thread
import time

class ClientThread(Thread):
    def __init__(self, clientSocket):
            Thread.__init__(self)
            self.clientSocket = clientSocket

    def run(self):
            while 1:
                    try:
                            # It will hang here, even if I do close on the socket
                            data = self.clientSocket.recv(1024)
                            print "Got data: ", data
                            self.clientSocket.send(data)
                    except:
                            break

            self.clientSocket.close()

HOST = ''
PORT = 6000
serverSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
serverSocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
serverSocket.bind((HOST, PORT))
serverSocket.listen(1)

clientSocket, addr = serverSocket.accept()
print 'Got a new connection from: ', addr
clientThread = ClientThread(clientSocket)
clientThread.start()

time.sleep(1)

# This won't make the recv in the clientThread to stop immediately,
# nor will it generate an exception
clientSocket.close()

Answer 1

我知道这是一个旧线程，塞缪尔可能很久以前就解决了他的问题。然而，我遇到了同样的问题，并在谷歌搜索时发现了这篇文章。找到解决方案并认为值得添加。

您可以在套接字类上使用 shutdown 方法。它可以阻止进一步的发送、接收或两者兼而有之。

socket.shutdown(socket.SHUT_WR)

例如，上述内容会阻止将来的发送。

请参阅 Python 文档以获取更多信息。

Answer 2

我不知道是否可以做你所要求的事情，但应该没有必要。如果没有什么可读的，就不要从套接字读取；使用

select.select

检查套接字的数据。

改变：

data = self.clientSocket.recv(1024)
print "Got data: ", data
self.clientSocket.send(data)

更像这样：

r, _, _ = select.select([self.clientSocket], [], [])
if r:
    data = self.clientSocket.recv(1024)
    print "Got data: ", data
    self.clientSocket.send(data)

编辑：如果你想防止套接字已关闭的可能性，请抓住

socket.error

。

do_read = False
try:
    r, _, _ = select.select([self.clientSocket], [], [])
    do_read = bool(r)
except socket.error:
    pass
if do_read:
    data = self.clientSocket.recv(1024)
    print "Got data: ", data
    self.clientSocket.send(data)

Answer 3

我找到了使用超时的解决方案。这将中断接收（实际上在超时到期之前，这很好）：

# Echo server program
import socket
from threading import Thread
import time


class ClientThread(Thread):
    def __init__(self, clientSocke):
        Thread.__init__(self)
        self.clientSocket = clientSocket

    def run(self):
        while 1:
            try:
                data = self.clientSocket.recv(1024)
                print "Got data: ", data
                self.clientSocket.send(data)
            except socket.timeout: 
                # If it was a timeout, we want to continue with recv
                continue
            except:
                break

        self.clientSocket.close()

HOST = ''
PORT = 6000
serverSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
serverSocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
serverSocket.bind((HOST, PORT))
serverSocket.listen(1)

clientSocket, addr = serverSocket.accept()
clientSocket.settimeout(1)

print 'Got a new connection from: ', addr
clientThread = ClientThread(clientSocket)
clientThread.start()

# Close it down immediatly 
clientSocket.close()

Answer 4

我必须为下面的评论道歉。 @Matt Anderson 的早期评论有效。我在尝试时犯了一个错误，这导致了我在下面的帖子。

使用超时并不是一个很好的解决方案。 瞬间醒来然后再重新进入睡眠状态似乎没什么大不了的，但我发现它极大地影响了应用程序的性能。 您有一个操作，大部分情况下都希望阻塞，直到数据可用为止，从而永远休眠。 但是，如果您由于某种原因想要中止，例如关闭应用程序，那么技巧就是如何退出。 对于套接字，您可以在两个套接字上使用 select 和 Listen。 你的主要的，和一个特殊的关闭的。 不过，创建关闭功能有点痛苦。 你必须创建它。 你必须让监听套接字接受它。 您必须跟踪该管道的两端。 我对同步队列类也有同样的问题。 但是，您至少可以在队列中插入一个虚拟对象来唤醒 get()。 但这要求虚拟对象看起来不像您的正常数据。 有时我希望 Python 有类似 Windows API WaitForMultipleObjects 的东西。

Answer 5

我喜欢下面的选择模式。
它取代了 select/socket 超时模式并允许内核执行线程唤醒。
提高程序退出/清理速度和响应能力。

import socket
import select


class SelectEvent:
    """thread.Event signal equivalent"""

    def __init__(self):
        self.r, self.w = socket.socketpair()
        self.triggered = False

    def set(self):
        if not self.triggered:
            self.triggered = True
            self.w.send(b"1")

    def clear(self):
        if self.triggered:
            self.triggered = False
            self.r.recv(1)

    def wait(self, waitable):
        """return true if signaled to exit"""
        readable, _, _ = select.select([waitable, self], [], [])
        return self in readable

    def close(self):
        self.r.close()
        self.w.close()

    def fileno(self):
        return self.r.fileno()
        
select_event = SelectEvent()

# ===================Thread A======================
while True:
    if self.select_event.wait(self.sock):
        break
    data, server = self.sock.recvfrom(UDP_MAX_LEN)
# =================================================

# ==================Main Thread====================
# Thread responsible for signaling close on reading thread
select_event.set()
threadA.join(5)
select_event.close()
threadA.close()
# =================================================