condition版生产者与消费者模式

时间：2024-07-31 23:05:38

1.简介

在爬虫中，生产者与消费者模式是经常用到的。我能想到的比较好的办法是使用redis或者mongodb数据库构造生产者消费者模型。如果直接起线程进行构造生产者消费者模型，线程容易假死，也难以构造复杂的生产者消费者模型。这里提供的condition版其实是最基本的生产者消费者模型的改良版，为了保护数据安全依旧是要开锁进行操作，但是不会循环的一直开锁，而是一旦条件不符合，则会阻塞，直到符合运行程序的条件。但是还是太low，不过这种思路值得借鉴。

2.代码

#  -*-coding:utf8 -*-

import threading

# Lock版本的生产者消费者模式可以正常的运行，但是太消耗CPU资源。

# 使用while循环的方式一直开锁解锁，很消耗资源。threading.Condition可以看作Lock的改良版

# 还有一个更好的模式就是用threading.Condition来实现

# threading.Condition可以在没有数据的时候处于阻塞等待状态，一旦有合适的数据了，

# 还可以使用notify相关的函数来通知其他处于等待状态的线程。这样就可以不用做一些无用的上锁

# 和解锁的操作，可以提高程序的性能。

# 1.acquire：获取锁

# 2.release：解锁

# 3.wait：释放内部占用的锁，同时线程被挂起。可以被其他线程用notify和notify_all函数唤醒，

# 被唤醒后会继续等待获取锁，获取锁后继续执行下面的代码

# 4.notify：唤醒一个挂起的线程，默认是第一个等待的线程。注意：notify不会释放所占用的锁

# 5.notify_all：通知所有正在等待的线程。notify和notify_all不会释放锁。并且需要在release之前调用

#  -*-coding:utf8 -*-

import threading

import random

import time

gMoney = 1000

gCondition = threading.Condition()

gTimes = 0

gTotalTimes = 30

class Producer(threading.Thread):

    def run(self):

        global gMoney

        global gTimes

        while True:

            money = random.randint(100, 1000)

            gCondition.acquire()

            if gTimes >= gTotalTimes:

                gCondition.release()

                break

            gMoney += money

            gTimes += 1

            print('%s生产了%s元钱，剩余%s元钱' % (threading.current_thread(), money, gMoney))

            gCondition.notify_all()

            gCondition.release()

            time.sleep(0.5)

class Consumer(threading.Thread):

    def run(self):

        global gMoney

        while True:

            money = random.randint(100, 1000)

            gCondition.acquire()

            while gMoney < money:

                if gTimes >= gTotalTimes:

                    gCondition.release()

                    return

                # 当不满足条件时，直接进入阻塞，不在循环开锁，关锁的消耗资源的操作

                gCondition.wait()

            gMoney -= money

            print('%s消费了%d元钱，剩余%d元钱' % (threading.current_thread(), money, gMoney))

            gCondition.release()

            time.sleep(0.5)

def main():

    for x in range(5):

        t = Consumer(name='消费者线程%s' % x)

        t.start()

    for x in range(2):

        t = Producer(name='生产者线程%s' % x)

        t.start()

if __name__ == '__main__':

    main()

相关文章