代码收藏家技术教程 2024-06-02

头哥Python魔鬼训练：解密生产者-消费者问题中的线程同步

1 任务描述

生产者-消费者问题是一个经典的多进程同步问题：

有两组进程：一组生产者进程和一组消费者进程共享一个初始为空、固定大小为 n 的缓冲区；

生产者的工作是制造数据，只有缓冲区没满时，生产者才能把数据放入到缓冲区，否则必须等待；

只有缓冲区不空时，消费者才能从中取出数据，一次只取一个数据，否则必须等待；

缓冲区只允许一个生产者放入消息，或者一个消费者从中取出消息。

本关任务：利用 Python 的 Condition 对象，解决多线程的生产者-消费者问题。

编程要求

本关的编程任务是，补全右侧编辑器中 Begin 至 End 区间的代码，具体要求如下：

全局变量 x 模拟缓冲区， x 的上限由测试输入给出；

生产者与消费者类的 self.num 变量即为 x 的上限，测试程序会创建一个生产者和一个消费者，并在创建类的同时会传入参数；

完成生产者与消费者类的 run() 方法，使得生产者每次输出 x 的值后，使得 x 加 1；同时消费者每次输出 x 的值后，使得 x 减 1；

注意，在输出语句之后，加入 time.sleep(0.1) 防止输出顺序混乱。

测试说明

平台将运行用户补全后的代码文件，并生成若干组测试数据，接着根据程序的输出判断程序是否正确。

以下是测试样例：

测试输入： 20 预期输出：

0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20
20 19 18 17 16 15 14 13 12 11 10 9 8 7 6 5 4 3 2 1 0

先上答案，后面讲解

# -*- coding: utf-8 -*-

import threading
import time
con = threading.Condition()
class Producer(threading.Thread):
    def __init__(self, num):
        threading.Thread.__init__(self)
        self.num = num
    def run(self):
        global x
        #*********begin*********#
        with con:
            while x >= self.num:
                con.wait()
            for i in range(x, self.num+1):
                if(i!=self.num):
                    print(i, end=" ")
                else:
                    print(i)
                x += 1
                time.sleep(0.1)
            con.notify()
            
        #********* end*********#

class Consumer(threading.Thread):
    def __init__(self, num):
        threading.Thread.__init__(self)
        self.num = num
    def run(self):
        global x
        #*********begin*********#
        with con:
            while x <= 0:
                con.wait()
            for i in range(self.num,-1, -1):
                if(i!=-1):
                    print(i, end=" ")
                else:
                    print(i)
                x -= 1
                time.sleep(0.1)
            con.notify_all()
        #********* end*********#
    
    
num = input()
x = 0
num = int(num)
p = Producer(num)
c = Consumer(num)
p.start()
c.start()
p.join()
c.join()

con = threading.Condition()

创建一个条件变量对象con,Condition对象用于线程间的同步通信。

class Producer(threading.Thread):
def __init__(self, num):
threading.Thread.__init__(self)
self.num = num

定义生产者线程类Producer,继承自threading.Thread。__init__方法接收num参数,表示缓冲区大小上限,并将其赋值给self.num。

def run(self):
global x
#*********begin*********#
with con:

run方法是线程的执行体。声明使用全局变量x,并获取条件变量con的上下文管理器(相当于调用con.acquire())。

while x >= self.num:
con.wait()

如果全局变量x(模拟缓冲区)已经达到上限self.num,则调用con.wait()释放锁并等待通知。

for i in range(x, self.num+1):
if(i!=self.num):
print(i, end=" ")
else:
print(i)
x += 1
time.sleep(0.1)

如果缓冲区未满,进入for循环,从x开始填充到self.num上限。打印当前x值(最后一个x单独一行),x加1,并短暂休眠0.1秒(防止输出混乱)。
con.notify()
当填充完成后,调用con.notify()发送一个通知给其他等待线程。

class Consumer(threading.Thread):
def __init__(self, num):
threading.Thread.__init__(self)
self.num = num

定义消费者线程类Consumer,继承自threading.Thread,__init__方法同Producer。