Python中使用线程的技巧


类型:Python,C++ & Qt4,创建时间:三月 14, 2012, 8:59 p.m.

标题无“转载”即原创文章,版权所有。转载请注明来源:http://hgoldfish.com/blogs/article/80/。

经常用Python写程序的朋友应该都知道怎么用threading模块来启动一个新线程。主要有两种方式:

  1. 直接使用threading.Thread类型。这种方法相对简单。比如下面这两行代码演示了如何启动一个新线程,并且当新线程调用sendData()函数时传入'arg1', arg2'两个参数:

    sendDataThread=threading.Thread(target=sendData, args=('arg1', 'arg2'))
    sendDataThread.start()
    
  2. 继承threading.Thread类,重载它的run()方法。这种方法比较麻烦,它的好处是,很方便把一个长的函数拆分成好几部分,方便多个线程之间的同步。比如下面这个发送数据的线程,它会发送数据的时候做统计,其中makeConnection()使用了连接池:

    class SendDataThread(threading.Thread):
        daemon=True
        def __init__(self, counter, dataSource):
            threading.Thread.__init__(self)
            #多个线程共享一个counter
            self.counter, self.dataSource=counter, dataSource
    
        def run(self):
            try:
                self.sendData()
            finally:
                del self.counter, self.dataSource
    
        def sendData(self):
            #此处省略连接,取数据等操作,分别是makeConnection()和makePacket()两个方法
            connection=self.makeConnection()
            data=self.dataSource.makePacket()
            sentBytes=connection.send(data)
            self.counter.increase(sentBytes)
    
        def makeConnection(self):
            "从连接池里面返回一个空闲的连接。"
    

两种方法实际上差不多。如果线程做的工作比较简单,只有一个函数就使用第一种。如果线程的工作繁复,可以拆成多个方法,就写成一个类。不过需要注意的是,第二种方法最好把run()方法像上面一样写成try...finally... 的形式,主要是为了避免循环引用。Python的GC使用了简单的引用计数,所以,如果Counter引用了SendDataThread,而SendDataThread也引用了Counter就会发生循环引用,两个对象可能不会被释放。这时,最好在SendDataThread执行完毕时解除对Counter的引用。当然,如果确定不会发生循环引用,可以不用这样做。

第二种方法需要继承自Thread,有时候不太方便。我们可以结合第一种方法,稍微变通一样。比如这样:

import threading, struct

class Writer:
    def write(self, fout, string, blockSize):
        for block in self.splitToBlock(string, blockSize):
            header=struct.pack("!i", len(block))
            fout.write(header)
            fout.write(block)

    def splitToBlock(self, string, blockSize):
        data=string.encode("utf-8")
        for i in range(0, len(data), blockSize):
            yield data[i:i+blockSize]

writer=Writer()
threading.Thread(target=writer.write, args=(fout, string, blockSize)).start()

最后一句推广开来可以弄成一个decorater,让同步执行的方法立即变成异步执行。

import threading, functools
def async(wrapped):
    def wrapper(*args, **kwargs):
        t=threading.Thread(target=wrapped, args=args, kwargs=kwargs)
        t.daemon=True
        t.start()
    functools.update_wrapper(wrapper, wrapped)
    return wrapper

于是write()可以改写成:

@async
def write(self, fout, string, blockSize):
    pass #和原来一样

下次直接调用writer.write()就变成异步运行了。

在Python里面使用线程最不爽的恐怕是不能强制结束线程。相比之下 ,Java语言的线程类支持Thread.interrupt(),当线程阻塞在Lock或者Event的时候仍然可以很方便地让线程退出。在Python里面,阻塞的调用多是直接调用C语言的系统函数,而不是像Java那样重写各种阻塞的IO。好处是Python的应用程序可能会拥有更好的IO性能,但是直接的坏处就是必须由用户自行处理阻塞函数。而且像Java那样重写阻塞IO的函数工作量太大,也给扩展Python解释器和移植工作带来很多的麻烦。

那么,既然不能强制性地结束线程,只好用一些迂回地办法,让线程自己退出。伪代码类似于这样:

class DoSomethingThread(threading.Thread):
    daemon=True
    def run(self):
        while True:
            self.doBlockIO()
            if self.exiting:break

    def shutdown(self):
        self.exiting=True
        self.interruptBlockIO()
        #self.join()

有两个关键,一是每次在阻塞函数返回后都判断一下标志位,看是不是应该结束线程了。通常每个循环体内只有一个阻塞函数,所以把判断放在循环语句上面就行了。二是采用某种办法让阻塞函数返回。各种阻塞函数都不尽相同。

如果阻塞函数支持超时,那就方便了,直接在阻塞函数内传入超时时间,比如0.2秒,连self.interruptBlockIO()这一句都可以省略掉。Python默认创建的socket是永远阻塞的,可以使用socket.settimeout()来设置超时时间——要小心捕获socket.timeout异常。更好的办法是使用select模块,它不仅可以设置超时,还能够在一个线程内同时处理socket的读和写。socket.connectsocket.accept()都支持超时。threading模块的各种锁、信号都支持超时。

如果阻塞函数不支持超时,那就只好采用一些山寨办法了。假定socket.accept()不支持超时的话,我们可以在shutdown函数里面创建一个新socket连接自己监听的端口,让accept()函数返回。

如果调用的C函数确实没办法让它从阻塞状态退出,可以考虑使用multiprocessing模块。因为强制结束一个进程不是一个多大不了的事。不过multiprocessingthreading相比,交换数据的效率会低一点。

细心的朋友可能会发现上面的几段代码都设置了daemon属性。将这个属性设置为True可以保证线程在主线程退出后也会立即退出。一般说来,主线程是负责用户UI的,如果用户关闭了程序,线程继续运行有什么意义呢。这个属性其实挺常用的,不知道为什么默认值是False

在使用多线程的时候还要注意import语句不支持多线程,也就是不能有多个线程同时在执行import语句。所以最好不要在模块导入过程中再启动一个新线程。比如模块这样写是不好的:

#encoding:utf-8
import threading, logging
logging.basicConfig()
logger=logging.getLogger(__name__)
def doSomething():
    "做一项麻烦的工作,并写日志。"
    logger.error("doSomething")
t=threading.Thread(target=doSomething)
t.start()
t.join()

这个mymodule写得不好。因为import mymodule将会发生死锁,原因是logger.error()的源代码内有一句import multiprocessing。如果主线程和doSomething()都执行到import那里,Python就会锁在import语句那里不会退出。于是t.join()会被阻塞,永远不会返回。事实上,我曾经发现即使去掉t.join()也有可能会出错,不过不知道怎么重现。一个比较好的写法是把线程的启动代码移进函数,不搞并发执行import语句就没关系了。

import mymodule
mymodule.doSomething()

经常使用PyQt的朋友可能会发现PyQt也有一个QThread,而且与threading.Thread相比,还多出了QThread.terminate(),它看起来能省却很多的麻烦。那么,写PyQt程序的时候到底应该选择哪一个呢?

我的看法,QThread不应被使用。

因为QThread是专为C++环境设计的,不适合Python程序。如果看过Qt文档的话,可以注意到QThread.terminate()本身也不是不推荐被使用的,如果一定要使用,就要调用QThread::setTerminationEnabled()设置可中断标志。如果C++代码仔细设置好可中断标志,QThread::terminate()就没有副作用。但是Python环境没办法像C++那样,在读写共享数据的时候设置可中断标志,除非你修改Python虚拟机,把这个调用整合到Python的源代码里面。

另外,当QThread没有被其它对象引用的时候,根据Python的内存管理模型,这个QThread会被删除。不幸的是,如果QThread的析构函数检测到线程是被强制结束的,它会打印出一行错误信息,然后结束整个进程。于是整个Python程序会意外地退出。显然,这种行为相当的不好。竟然不是抛出一个异常。如果你不想因为一个小小的可挽回的错误导致整个程序失败的话,最好不要使用QThread

另外,尽量不要在非主线程使用QObject的对象。Qt的文档说,QObjectQThread有特别的关联。有以下几个注意事项:

  1. 只有创建QObject的进程才能使用它。不能在一个线程里面创建QTimer,而在另外一个线程里面调用QTimer.start()
  2. 在一个线程里面创建的QObject不能在另外一个线程里面被销毁。
  3. Qt的内存管理模型区别“父QObject”和“子QObject”。Qt要求“子QObject”必须和“父QObject”同一个线程。

又一次很不幸,第2条和Python的GC有冲突。Python的GC不固定地在某个线程里面运行。如果刚好回收了一个不在当前线程里面创建的QObject,程序就有可能会崩溃。注:貌似PyQt的开发者提到会解决这个问题,不知道现在怎么样了。

标题无“转载”即原创文章,版权所有。转载请注明来源:http://hgoldfish.com/blogs/article/80/。


cnDenis(三月 17, 2012, 12:10 p.m.)

老鱼兄,你的Blog有RSS输出吗?我想用GR订着看

老鱼(三月 18, 2012, 3:35 a.m.)

没有哦。。等我有空加上去。

shelper(三月 28, 2012, 11:50 p.m.)

这个rss是必须的:)


何不来发表一下您对本文的看法(使用Markdown语法,分段空两行):