如何退出多线程程序？

Question

我只是在 python 中搞乱线程，写了这个基本的 IM 东西 [代码在底部]

我注意到当我用抄送终止程序时它不会退出，它只是永远挂起。

我只是猜测它正在等待每个线程完成他们正在做的事情，但因为这是一个永远不会发生的无限循环。
所以我想我需要手动终止每个线程，或者在终止信号到来时结束循环。
我该怎么做呢？

#!/usr/bin/env python
import threading
import socket

class Listen(threading.Thread):

    def run(self):
        conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        conn.bind(('', 2727))
        conn.listen(1)
        while True:
            channel, details = conn.accept()
            print str(details)+": "+channel.recv(250)
            channel.send("got it")
            channel.close()

class Shout(threading.Thread):

    def run(self):
        while True:
            try:    
                address = raw_input("who u talking to? ")
                conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                conn.connect((address, 2727))
                break
            except:
                print "can't connect to "+ str(address)
        while True:
            conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            conn.connect((address, 2727))
            conn.send(raw_input())
            conn.close()

listen = Listen().start()
shout = Shout().start()

Answer 1

我在您的代码中发现了导致不当行为的几个原因。

1. Ctrl+C 在主线程中导致“KeyboardInterrupt”异常。所以你应该在那里处理它。
2. 您的套接字处于阻塞模式。这会导致多个套接字函数阻塞调用线程，直到函数返回。在此状态期间，线程无法对任何终止事件做出反应。
3. 正如您已经说过的：线程 run() 函数中的无限循环......真的是无限的。因此线程执行永远不会结束（至少不会出现意外异常）。您应该使用某种同步对象，例如 threading.Event 对象，以便能够从外部告诉线程它应该自行终止。
4. 我不鼓励在主线程之外使用 raw_input() 。想象一下当您有多个 Shout 线程时会发生什么。
5. 为什么在 Shout 类中传输消息后总是关闭并重新连接套接字？由于设置成本，仅在特殊情况下才应重新建立网络连接。
6. 如果没有用于通信的帧协议，您永远无法指望在 recv() 函数返回时收到其他主机发送的所有数据。
7. 线程对象的start()函数不返回值或对象。所以保存返回值（=None）没有多大意义。
8. 您永远不能期望 send() 函数传输所有传递的数据。因此，您必须检查函数的结果，并适当处理未真正传输所有字节的情况。
9. 要学习线程，肯定有比网络通信更好的问题需要解决，因为该主题本身非常复杂。

除了所有这些事情之外，这是我尝试的解决方案。仍然有很多可以改进的地方。您也应该考虑 Mark Tolonen 的答案，因为肯定提供了 SocketServer 类来简化处理此类内容的一些事情。但你也应该继续学习基础知识。

#!/usr/bin/env python
import threading
import socket
import time
import errno

class StoppableThread(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.stop_event = threading.Event()        

    def stop(self):
        if self.isAlive() == True:
            # set event to signal thread to terminate
            self.stop_event.set()
            # block calling thread until thread really has terminated
            self.join()

class Accept(StoppableThread):
    def __init__(self, port):
        StoppableThread.__init__(self)
        self.port = port
        self.threads = []

    def run(self):     
        # handle connection acception
        conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        conn.bind(('', self.port ))
        conn.listen(5)
        # set socket timeout to ~10ms
        conn.settimeout(0.01)
        while self.stop_event.is_set() == False:
            try:
                csock, caddr = conn.accept()
                # spawn a new thread to handle the client connection
                listen_thread = Listen(csock, caddr)
                self.threads.append(listen_thread)
                listen_thread.start()
            except socket.timeout:
                # socket operation timeout
                # clear all terminated threads from thread list                
                for thread in self.threads:
                    if thread.isAlive() == False:
                        self.threads.remove(thread)

        self.stop_threads()

    def stop_threads(self):
        # stop all running threads
        for listen_thread in self.threads:
            if listen_thread.isAlive() == True:
                listen_thread.stop()
        self.threads = [] 

class Listen(StoppableThread):
    def __init__(self, csock, caddr):
        StoppableThread.__init__(self)
        self.csock = csock
        self.caddr = caddr
        self.csock.setblocking(False)

    def run(self):                
        while self.stop_event.is_set() == False:            
            try:                
                recv_data = self.csock.recv(250)
                if len(recv_data) > 0:       
                    print str(self.caddr)+": " + recv_data
                    self.csock.send("got it")                    
                else:
                    # connection was closed by foreign host
                    self.stop_event.set()
            except socket.error as (sock_errno, sock_errstr):
                if (sock_errno == errno.EWOULDBLOCK):
                    # socket would block - sleep sometime
                    time.sleep(0.1)                    
                else:
                    # unexpected / unhandled error - terminate thread
                    self.stop_event.set()
        channel.close()

class Shout(StoppableThread):
    def __init__(self, sport):
        StoppableThread.__init__(self)
        self.sport = sport

    def run(self):
        while self.stop_event.is_set() == False:
            try:    
                address = raw_input("who u talking to? ")
                conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                conn.connect((address, self.sport))
                break
            except socket.error:
                # handle connection problems
                print "can't connect to "+ str(address)
            except: 
                # exit thread in case of an unexpected error
                self.stop_event.set()

        while self.stop_event.is_set() == False:
            try: 
                # chat loop: send messages to remote host            
                print "what to send? :",
                msg = raw_input()
                # beware: send() function may block indefinitly here and it might not send all bytes as expected !!
                conn.send(msg)
            except:
                # exit thread in case of an unexpected error
                self.stop_event.set()
        # close socket before thread terminates
        conn.close()

def main():
    do_exit = False
    server_port = 2727

    # start server socket thread
    accept = Accept(server_port)
    accept.start()

    # start transmitting client socket thread
    shout = Shout(server_port)
    shout.start()

    while do_exit == False:
        try:
            # sleep some time
            time.sleep(0.1)
        except KeyboardInterrupt:
            # Ctrl+C was hit - exit program
            do_exit = True

    # stop all running threads
    shout.stop()
    accept.stop()

    # exit main program after all threads were terminated gracefully    

if __name__ == "__main__":
    main()

Answer 2

查看 SocketServer.py 的 Python 库源代码，特别是 server_forever() 的实现，以了解服务器如何实现退出。它使用 select() 轮询服务器套接字以获取新连接并测试退出标志。这是对您的源代码使用 SocketServer 的修改，我向 Shout() 添加了一个退出标志。它将运行 Shout 和 Listen 线程 5 秒钟，然后停止它们。

import socket
import SocketServer
import threading
import time

class Handler(SocketServer.StreamRequestHandler):
    def handle(self):
        print str(self.client_address) + ": " + self.request.recv(250)
        self.request.send("got it\n")

class Listen(threading.Thread):
    def run(self):
        self.server = SocketServer.TCPServer(('',2727),Handler)
        self.server.serve_forever()
    def stop(self):
        self.server.shutdown()

class Shout(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.quit = False
    def run(self):
        while not self.quit:
            conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            conn.connect(('localhost', 2727))
            conn.send('sending\n')
            print conn.recv(100)
            conn.close()
    def stop(self):
        self.quit = True

listen = Listen()
listen.start()
shout = Shout()
shout.start()

time.sleep(5)

shout.stop()
listen.stop()