如何退出多线程程序？

Question

我在玩Python的多线程，写了一个简单的即时通讯程序[代码在下面]

我发现当我用C-c终止程序时，它并没有退出，而是一直卡在那里。

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

#!/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

看看Python库中的SocketServer.py文件，特别是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()

Answer 2

我看到你代码中有几个问题导致它表现不正常。

1. 按下Ctrl+C会在主线程中引发一个“KeyboardInterrupt”异常。所以你需要在主线程里处理这个异常。
2. 你的socket处于阻塞模式。这意味着一些socket函数会让调用它的线程停下来，直到函数执行完毕。在这个状态下，线程无法响应任何终止事件。
3. 正如你所说的：你在线程的run()函数中写的无限循环，确实是无限的。所以线程的执行不会结束（至少不会在没有意外异常的情况下）。你应该使用某种同步对象，比如threading.Event对象，这样可以让线程知道外部希望它结束。
4. 我不建议在主线程之外使用raw_input()。想象一下，如果你有多个Shout线程，会发生什么。
5. 为什么在你的Shout类中，每次传输消息后都要关闭并重新连接socket？网络连接应该只在特殊情况下重新建立，因为这样做的开销很大。
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()