Python多线程模块

• • 引言
  • thread
  • threading
    • 1 Thread
      • 11 下面使用threading模块实现与上面相同的功能
      • 12 在创建新线程时还可以给Thread传递可调用类的对象这样使用类本身来保存信息 如
      • 13 从Thread派生一个子类然后创建这个子类的实例
      • 14 实例
  • Queue

0. 引言

在Python中可使用的多线程模块主要有两个，thread和threading模块。thread模块提供了基本的线程和锁的支持，建议新手不要使用。threading模块允许创建和管理线程，提供了更多的同步原语。

1. thread

thread模块函数：

start_new_thread(function, args[, kwargs])

启动新的线程以执行function，返回线程标识。

allocate_lock()

返回LockType对象。

exit()

抛出SystemExit异常，如果没有被捕获，线程静默退出。

LockType类型锁对象的方法：

acquire([waitflag])

无参数，无条件获得锁，如果锁已经被其他线程获取，则等待锁被释放。如果使用整型参数，参数为0，如果锁可获取，则获取且返回True，否则返回False；参数为非0，与无参数相同。

locked()

返回锁的状态，如果已经被获取，则返回True，否则返回False。

release()

释放锁。只有已经被获取的锁才能被释放，不限于同一个线程。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import thread
from time import ctime
from time import sleep


loops = [4, 2]

def loop0():
    print 'start loop 0 at:', ctime()
    sleep(4)
    print 'loop 0 done at:', ctime()


def loop1():
    print 'start loop 1 at:', ctime()
    sleep(2)
    print 'loop 1 done at:', ctime()


def loop(nloop, nsec, lock):
    print 'start loop', nloop, 'at:', ctime()
    sleep(nsec)
    print 'loop', nloop, 'done at:', ctime()
    lock.release()


def main():
    print 'starting  at:', ctime()
    loop0()
    loop1()
    print 'all DONE at:', ctime()


def main1():
    print 'starting at:', ctime()
    thread.start_new_thread(loop0, ())
    thread.start_new_thread(loop1, ())
    sleep(6)
    print 'all DONE at:', ctime()


def main2():
    print 'starting at:', ctime()
    locks = []
    nloops = range(len(loops))
    for i in nloops:
        lock = thread.allocate_lock()
        lock.acquire()
        locks.append(lock)

    for i in nloops:
        print thread.start_new_thread(loop, (i, loops[i], locks[i]))

    for i in nloops:
        while locks[i].locked():
            pass

    print 'all DONE at:', ctime()


if __name__ == '__main__':
    #main()
    #main1()
    main2()

2. threading

threading模块提供了更好的线程间的同步机制。threading模块下有如下对象：

• Thread
• Lock
• RLock
• Condition
• Event
• Semaphore
• BoundedSemaphore
• Timer

threading模块内还有如下的函数：

• active_count()
• activeCount()
  
  • 返回当前alive的线程数量
• Condition()
  
  • 返回新的条件变量对象
• current_thread()
• currentThread()
  
  • 返回当前线程对象
• enumerate()
  
  • 返回当前活动的线程，不包括已经结束和未开始的线程，包括主线程及守护线程。
• settrace(func)
  
  • 为所有线程设置一个跟踪函数。
• setprofile(func)
  
  • 为所有纯种设置一个profile函数。

2.1 Thread

类Thread有如下属性和方法：

• Thread(group=None, target=None, name=None, args=(), kwargs={})
• start()
• run()
• join([timeout])
• name
• getName()
• setName(name)
• ident
• is_alive()
• isAlive()
• daemon
• isDaemon()
• setDaemon(daemonic)

创建线程一般有如下三种方法：
1. 传递函数创建Thread实例。
2. 传递可调用类的实例创建Thread实例。
3. 从Thread派生出一个子类，创建一个子类实例。

2.1.1 下面使用threading模块实现与上面相同的功能：

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import threading
from time import ctime
from time import sleep


loops = [4, 2]


def loop(nloop, nsec):
    print 'start loop', nloop, 'at:', ctime()
    sleep(nsec)
    print 'loop', nloop, 'done at:', ctime()


def main():
    print 'starting at:', ctime()
    threads = []
    nloops = range(len(loops))

    for i in nloops:
        t = threading.Thread(target=loop, args=(i, loops[i]))
        threads.append(t)

    for i in nloops:
        threads[i].start()

    for i in nloops:
        threads[i].join()

    print 'all DONE at:', ctime()


if __name__ == '__main__':
    main()

程序输出如下：

starting at: Fri Jul 15 15:56:25 2016
start loop 0 at: Fri Jul 15 15:56:25 2016
start loop 1 at: Fri Jul 15 15:56:25 2016
loop 1 done at: Fri Jul 15 15:56:27 2016
loop 0 done at: Fri Jul 15 15:56:29 2016
all DONE at: Fri Jul 15 15:56:29 2016

2.1.2 在创建新线程时，还可以给Thread传递可调用类的对象，这样使用类本身来保存信息， 如：

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import threading
from time import ctime
from time import sleep


loops = [4, 2]

class ThreadFunc(object):
    def __init__(self, func, args, name=''):
        self.name = name
        self.func = func
        self.args = args

    def __call__(self):
        apply(self.func, self.args)


def loop(nloop, nsec):
    print 'start loop', nloop, 'at:', ctime()
    sleep(nsec)
    print 'loop', nloop, 'done at:', ctime()


def main():
    print 'starting at:', ctime()
    threads = []
    nloops = range(len(loops))

    for i in nloops:
        t = threading.Thread(target=ThreadFunc(loop, (i, loops[i]),
                                               loop.__name__))
        threads.append(t)

    for i in nloops:
        threads[i].start()

    for i in nloops:
        threads[i].join()

    print 'all DONE at:', ctime()


if __name__ == '__main__':
    main()

程序的输出与上面的是一致的。

2.1.3 从Thread派生一个子类，然后创建这个子类的实例

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import threading
from time import ctime
from time import sleep


loops = [4, 2]

class MyThread(threading.Thread):
    def __init__(self, func, args, name=''):
        threading.Thread.__init__(self)
        self.name = name
        self.func = func
        self.args = args

    def run(self):
        apply(self.func, self.args)


def loop(nloop, nsec):
    print 'start loop', nloop, 'at:', ctime()
    sleep(nsec)
    print 'loop', nloop, 'done at:', ctime()


def main():
    print 'starting at:', ctime()
    threads = []
    nloops = range(len(loops))

    for i in nloops:
        t = MyThread(loop, (i, loops[i]), loop.__name__)
        threads.append(t)

    for i in nloops:
        threads[i].start()

    for i in nloops:
        threads[i].join()

    print 'all DONE at:', ctime()


if __name__ == '__main__':
    main()

程序运行结果与上面也是一样的。

2.1.4 实例

现在将MyThread单独放在一个模块内，就叫myThread：

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import threading
from time import ctime

class MyThread(threading.Thread):
    def __init__(self, func, args, name=''):
        threading.Thread.__init__(self)
        self.name = name
        self.func = func
        self.args = args

    def run(self):
        print 'starting', self.name, 'at', ctime()
        self.res = apply(self.func, self.args)
        print self.name, 'finished at:', ctime()

    def getResult(self):
        return self.res


if __name__ == '__main__':
    pass

现在要计算阶乘、求和、fibinacci。由于计算时间不同，添加适当的sleep()进行时间上控制。

#!/usr/bin/env python
# -*- coding: utf-8 -*-


from myThread import MyThread
from time import ctime
from time import sleep


def fib(x):
    sleep(0.005)
    if x < 2:
        return 1
    return fib(x - 1) + fib(x - 2)


def fac(x):
    sleep(0.1)
    if x < 2:
        return 1
    return x * fac(x - 1)


def sum_c(x):
    sleep(0.1)
    if x < 2:
        return 1
    return x + sum_c(x - 1)


def main():
    nfuncs = range(len(funcs))
    print '*** SINGLE THREAD'
    for i in nfuncs:
        print 'starting', funcs[i].__name__, 'at:', ctime()
        print funcs[i](n)
        print funcs[i].__name__, 'finished at:', ctime()

    print '
*** MULTIPLE THREADS'
    threads = []
    for i in nfuncs:
        t = MyThread(funcs[i], (n,), funcs[i].__name__)
        threads.append(t)

    for i in nfuncs:
        threads[i].start()

    for i in nfuncs:
        threads[i].join()
        print threads[i].getResult()

    print 'all DONE'


funcs = [fib, fac, sum_c]
n = 12

if __name__ == '__main__':
    main()

结果如下：

*** SINGLE THREAD
starting fib at: Fri Jul 15 17:48:02 2016
233
fib finished at: Fri Jul 15 17:48:04 2016
starting fac at: Fri Jul 15 17:48:04 2016
479001600
fac finished at: Fri Jul 15 17:48:05 2016
starting sum_c at: Fri Jul 15 17:48:05 2016
78
sum_c finished at: Fri Jul 15 17:48:07 2016

*** MULTIPLE THREADS
starting fib at Fri Jul 15 17:48:07 2016
starting fac at Fri Jul 15 17:48:07 2016
starting sum_c at Fri Jul 15 17:48:07 2016
fac finished at: Fri Jul 15 17:48:08 2016
sum_c finished at: Fri Jul 15 17:48:08 2016
fib finished at: Fri Jul 15 17:48:09 2016
233
479001600
78
all DONE

3. Queue

Queue模块可以用来线程间通讯，让各个线程之间共享数据。通过Queue模块的工厂方法Queue(maxsize=0)创建Queue对象，maxsize指定了队列的大小，默认为无限大小。对象Queue属性如下：

• qsize()
  
  • 返回队列的大小。
• empty()
  
  • 如果队列为空，返回True，否则返回False。
• full()
  
  • 如果队列已满，返回True，否则返回False。
• put(item[, block[, timeout]])
  
  • 把item放入队列
• get([block[, timeout]])
  
  • 从队列头部取出一个对象

#!/usr/bin/env python
# -*- coding: utf-8 -*-


from myThread import MyThread
from Queue import Queue
from random import randint
from time import sleep


def writeQ(queue):
    print 'producing object for Q...',
    queue.put('xxx', 1)
    print 'size now', queue.qsize()


def readQ(queue):
    val = queue.get(1)
    print 'consumed object from Q... size now', queue.qsize()


def writer(queue, loops):
    for i in range(loops):
        writeQ(queue)
        sleep(randint(1, 3))


def reader(queue, loops):
    for i in range(loops):
        readQ(queue)
        sleep(randint(2, 5))


funcs = [writer, reader]
nfuncs = range(len(funcs))


def main():
    nloops = randint(2, 5)
    q = Queue(32)

    threads = []
    for i in nfuncs:
        t = MyThread(funcs[i], (q, nloops), funcs[i].__name__)
        threads.append(t)

    for i in nfuncs:
        threads[i].start()

    for i in nfuncs:
        threads[i].join()

    print 'all DONE'


if __name__ == '__main__':
    main()

可能运行结果为：

starting writer at Mon Jul 18 10:59:13 2016
producing object for Q... size now 1
starting reader at Mon Jul 18 10:59:13 2016
consumed object from Q... size now 0
producing object for Q... size now 1
consumed object from Q... size now 0
producing object for Q... size now 1
writer finished at: Mon Jul 18 10:59:19 2016
consumed object from Q... size now 0
reader finished at: Mon Jul 18 10:59:24 2016
all DONE