Python 多线程及队列

1. queue 模块介绍

　　queue模块实现了多生产者，多消费者队列。在多线程环境中，该队列能实现多个线程之间安全的信息交换。

2. queue 对象的以下方法

　　Queue.qsize()

　　　　返回队列的近似大小。注意，qsize() > 0 并不能保证接下来的get()方法不被阻塞；同样，qsize() < maxsize 也不能保证 put() 将不被阻塞

　　Queue.empty()

　　　　如果队列是空的，则返回True，否则False. 如果empty()返回True,并不能保证接下来的put()将调用不被阻塞。类似的，empty() 返回False也不能保证接下来的get()调用将不被阻塞。

　　Queue.full()

　　　　如果队列满则返回True，否则返回False。如果full()返回True，并不能保证接下来的get()调用将不被阻塞。类似的，full()返回False也不能保证接下来的put()调用将不被阻塞。

　　Queue.put(item, block=True, timeout=None)

　　　　如果block是True，且timeout是None，该方法将一直等待直到有队列有空余空间。如果timeout是一个正整数，该方法则最多阻塞timeout秒并抛出Full异常。

　　Queue.put_nowait(item)

　　　　等价于put(item, False)。

　　Queue.get(block=True, timeout=None)

　　　　从队列中移除被返回一个条目。如果block是True并且timeout是None，该方法将阻塞直到队列中有条目可用。如果timeout是正整数，该方法将最多阻塞timeout秒并抛出Empty异常。

　　　　如果block是False并且队列为空，则直接抛出Empty异常（这时timeout将被忽略）。

　　Queue.get_nowait()

　　　　等价于get(False)。

　　Queue.task_done()
　　　　表示一个先前的队列中的任务完成了。被队列消费者线程使用。对于每个get()获取到的任务，接下来的task_done()的调用告诉队列该任务的处理已经完成。

　　　　如果join()调用正在阻塞，当队列中所有的条目被处理后它将恢复执行（意味着task_done()调用将被放入队列中的每个条目接收到）。

　　　　如果调用次数超过了队列中放置的条目数目，将抛出ValueError异常。
　　Queue.join()

　　　　阻塞直到队列中所有条目都被获取并处理。

　　　　当一个条目被增加到队列时，未完成任务的计数将增加。当一个消费者线程调用task_done()时，未完成任务的计数将减少。当未完成任务的计数减少到0时，join()解锁。

3. 线程锁

import threading

lock = threading.Lock()

lock.acquire()   #加锁


lock.release()   #释放锁

4. 多线程 + 队列示例

from threading import Thread,current_thread

import time
import random
from queue import Queue


# 创建一个队列，队列长度为5
queue = Queue(5)

# 生产者线程，继承Thread

class ProducerThread(Thread):
    def run(self):
        # 获的线程名称
        name = current_thread().getName()
        # 设置随机数字的范围，在100以内
        nums = range(100)
        # 定义一个队列变量，声明了global 是要在while 循环内部使用变量
        global queue
        while True:
            # 随机选择一个数字
            num = random.choice(nums)
            # 往队列里加入数据
            queue.put(num)
            print("生产者 %s 生产了数据 %s " % (name, num))
            t = random.randint(1, 3)
            time.sleep(t)
            print("生产者 %s 睡眠了 %s 秒" % (name, t) )
            print("队列数量= %s" % queue.qsize() )


# 消费者线程，继承Thread
class ConsumerThread(Thread):
    def run(self):
        name = current_thread().getName()
        global queue
        while True:
            num = queue.get()
            # 线程等待和线程同步
            queue.task_done()
            print("消费者 %s 消耗了数据 %s" % (name, num))
            # 随机等待几秒
            t = random.randint(1,5)
            time.sleep(t)
            print("消费者 %s 睡眠了 %s 秒" % (name, t))

if __name__ == '__main__':
    # 一个生产者，2个消费者
    # p1 = ProducerThread(name="producer1")
    # p1.start()
    # c1 = ConsumerThread(name="consumer1")
    # c1.start()
    # c2 = ConsumerThread(name='consumer2')
    # c2.start()

    # 3个生产者，2个消费者，这种情况生产更多，会导致队列满的时候会停止生产，要等到消费者消费了队列中的数据才会进行生产

    p1 = ProducerThread(name="producer1")
    p1.start()
    p2 = ProducerThread(name="producer2")
    p2.start()
    p3 = ProducerThread(name="producer3")
    p3.start()
    c1 = ConsumerThread(name="consumer1")
    c1.start()
    c2 = ConsumerThread(name='consumer2')
    c2.start()

run 方法

import threading
from threading import current_thread


# 继承threading.Thread ,并覆盖run 方法

class Mythread(threading.Thread):
    def run(self):
        print(current_thread().getName(), 'start')
        print('run')
        print(current_thread().getName(), 'stop')

# 使用类，首先要进行类的实例化
t1 = Mythread()
t1.start()

# 子线程先结束，主线程后结束的方法，用join方法
t1.join()

print(current_thread().getName(),'end')