python threading模块使用以及python多线程操作的实践(使用Queue队列模块)

介绍

今天花了近乎一天的时间研究python关于多线程的问题，查看了大量源码自己也实践了一个生产消费者模型，所以把一天的收获总结一下。

由于GIL(Global Interpreter Lock)锁的关系,纯的python代码处理一般逻辑的确无法活动性能上的极大提升，但是在处理需要等待外部资源返回或多用户的应用程序中，多线程仍然可以作为一个比较好的工具来进行使用。

threading and thread

python提供了两个模块thread和threading 来支持python的多线程操作。通俗的讲一般现在我们只使用threading模块来编程了，thread模块定义了很多原始行为，更接近底层，而threading模块抽象了thread模块可用性更好，同时提供更多特性。

现在创建线程的通用方法一般是创建一个类并且继承threading.Thread，然后重写其中的__init__和run()方法。更多详情可以参考threading模块代码内注释以及代码。下面直接看个例子。

import time
import threading


class Test(threading.Thread):
    def __init__(self, name, delay):
        super(Test, self).__init__()
        self.name = name
        self.delay = delay

    def run(self):
        print "%s delay for %s seconds" % (self.name, self.delay)
        time.sleep(self.delay)
        c = 0
        while True:
            print "This is thread %s on line %s" % (self.name, c)
            c += 1
            if c == 3:
                print "End of thread %s" % self.name
                break


t1 = Test('Thread1', 5)
t2 = Test('Thread2', 5)

t1.start()
print 'Wait t1 to end'
t1.join()
t2.start()
t2.join()
print 'End of main'

注意一下这一句：

super(Test, self).__init__()

这是按照模块要求，必须初始化父类的__init__函数所以使用了super()

其他并没有多少值得注意的地方，

创建线程方便的实例化自己写的继承threading.Thread的类然后传入对应的参数。

最后使用xxx.start()来运行线程。使用xxx.join()来阻塞线程。

特别注意的是。继承自Threading类的子类还有一个daemon参数，如果这个参数适用setDaemon()方法置为True之后，主线程将不会等待子线程都结束之后才结束，而是自己运行完之后就结束，这种方式相当粗暴。如果将daemon参数设置为False的话，主线成将会等待所有子线程结束之后再结束。daemon属性可以通过使用isDaemon()方法获取一个boolean值。

互斥与同步

更进一步的，我必须介绍一下线程之间的同步和互斥问题。下面引用《计算机操作系统》中的介绍。

当线程并发执行时，由于资源共享和线程协作，使用线程之间会存在以下两种制约关系。

（1）间接相互制约。一个系统中的多个线程必然要共享某种系统资源，如共享CPU，共享I/O设备，所谓间接相互制约即源于这种资源共享，打印机就是最好的例子，线程A在使用打印机时，其它线程都要等待。

（2）直接相互制约。这种制约主要是因为线程之间的合作，如有线程A将计算结果提供给线程B作进一步处理，那么线程B在线程A将数据送达之前都将处于阻塞状态。

间接相互制约可以称为互斥，直接相互制约可以称为同步，对于互斥可以这样理解，线程A和线程B互斥访问某个资源则它们之间就会产个顺序问题——要么线程A等待线程B操作完毕，要么线程B等待线程操作完毕，这其实就是线程的同步了。因此同步包括互斥，互斥其实是一种特殊的同步。

在一段时间内只允许一个线程访问的资源就称为临界资源或独占资源，计算机中大多数物理设备，进程中的共享变量等待都是临界资源，它们要求被互斥的访问。每个进程中访问临界资源的代码称为临界区。

生产消费者的经典例子

这里为了介绍这种稍微复杂的概念。再列出一个生产消费者的例子使用到了Queue队列。

# coding:utf-8
import Queue
import time
import random
import threading


# write_lock = threading.Lock()                                                 # 创建primitive锁对象用于控制输出


class Producer(threading.Thread):
    # q传递一个队列参数, con传递了一个链接, name传递了一个名字
    def __init__(self, q, name):
        super(Producer, self).__init__()
        self.q = q
        # self.con = con
        self.name = name
        print "Producer " + self.name + "Started"

    def run(self):
        while True:
            # 锁对象常用的acquire获得锁方法和release释放锁方法
            # 这里使用的是Thread的Condition对象
            # self.con.acquire()
            if self.q.full():
                print 'Queue is full, producer wait!'

                # 手动挂起,并且只能在获得Lock的情况下才可以使用 否则会触发RuntimeError
                # 调用wait()会释放Lock 直到该线程被notify(),notifyall()或超时该线程又重新获得Lock
                # self.con.wait()
            else:
                value = random.randint(0, 10)
                print self.name + " put " + str(value) + "into queue"
                self.q.put((self.name+":"+str(value)))                    # 放置到队列中

                # 通知消费者,notify通知其他线程,被挂起的线程接到通知后会开始运行
                # 默认通知一个正在等待该condition的线程,最多唤醒n个线程 必须在获得Lock的情况下使用否则会报错.
                # self.con.notify()
                # self.con.release()                                            # 释放锁对象


class Consumer(threading.Thread):
    def __init__(self, q, name):
        super(Consumer, self).__init__()
        self.q = q
        # self.con = con
        self.name = name
        print "Consumer " + self.name + "started\n"

    def run(self):
        while True:
            # Condition常用的acquire获得锁方法和release释放锁方法
            # self.con.acquire()
            if self.q.empty():
                print 'queue is empty, consumer wait!'
                # self.con.wait()
            else:
                value = self.q.get()                                            # 从队列中取消息
                print self.name + " get " + value + "from queue"

                # 发送消息通知生产者
                # self.con.notify()
                # self.con.release()                                              # 释放锁对象
                print 'queue still have ' + str(q.qsize()) + 'task\n'


if __name__ == "__main__":
    q = Queue.Queue(10)

    # 使用Condition对象可以在某些事件触发或达到特定的条件后才处理数据.
    # con = threading.Condition()

    # 两个生产者
    p1 = Producer(q, "P1")
    p2 = Producer(q, "P2")
    c1 = Consumer(q, "C1")
    p2.start()
    p1.start()
    c1.start()

若有问题欢迎指出

参考链接

python threading模块文档翻译： http://my.oschina.net/lionets/blog/194577?fromerr=pbWOeveo
多线程7经典线程与互斥总结：http://blog.csdn.net/dazhong159/article/details/7927034
《编写高质量代码改善python程序的91个建议》第48和49建议。