Python实现超市模拟中多顾客同时处理延时的方案

一、Asyncio 异步实现方案 #

针对你的场景，asyncio 是解决阻塞问题的最优选择——它通过事件循环实现非阻塞等待，在顾客处理延时期间，程序可以继续执行其他逻辑（比如生成新顾客、处理其他结账通道）。

修改后的代码 #

import asyncio

class Cashier:
    # 保留你原有的ExtractCustomerData逻辑，此处为示例返回
    def ExtractCustomerData(self):
        return {
            "c1": {"Process Time": 2, "CustomerID": 1},
            "c2": {"Process Time": 3, "CustomerID": 2}
        }

    # 单个顾客的异步处理逻辑
    async def process_single_customer(self, customer_data):
        delay = customer_data["Process Time"]
        await asyncio.sleep(delay)  # 非阻塞等待，让出事件循环给其他任务
        print(f"Customer {customer_data['CustomerID']} time has been completed.")

    # 批量并发处理顾客
    async def ProcessItems(self):
        customers_in_cashier = self.ExtractCustomerData()
        # 为每个顾客创建异步任务
        tasks = [self.process_single_customer(data) for data in customers_in_cashier.values()]
        # 并发执行所有任务，等待全部完成
        await asyncio.gather(*tasks)

# 模拟程序入口
async def main():
    cashier = Cashier()
    # 可同时并行运行多个任务，比如结账通道处理+顾客生成
    await asyncio.gather(
        cashier.ProcessItems(),
        # generate_new_customers()  # 假设这是你的顾客生成异步函数
    )

if __name__ == "__main__":
    asyncio.run(main())

关键改动说明 #

1. 异步函数标记：给涉及等待的函数加上async关键字，用await asyncio.sleep(delay)替代time.sleep(delay)——前者不会阻塞整个程序，只是暂停当前任务，让事件循环去执行其他任务。
2. 任务拆分：把单个顾客的处理逻辑拆成独立函数，每个顾客对应一个异步任务，避免循环内逐个阻塞。
3. 并发执行：用asyncio.gather()打包所有任务，实现多顾客同时处理，同时不影响其他模拟逻辑运行。

Asyncio 适用场景 #

当你的程序以等待操作为主（比如模拟中的处理延时、IO请求），且不需要大量CPU计算时，asyncio是轻量高效的并发方案，适合超市模拟这种多任务并行等待的场景。

二、多线程备选方案 #

如果对异步逻辑理解有困难，也可以用多线程实现——每个顾客的处理在单独线程中执行，主线程不会被阻塞。

修改后的代码 #

import threading
import time

class Cashier:
    def ExtractCustomerData(self):
        return {
            "c1": {"Process Time": 2, "CustomerID": 1},
            "c2": {"Process Time": 3, "CustomerID": 2}
        }

    def process_single_customer(self, customer_data):
        delay = customer_data["Process Time"]
        time.sleep(delay)
        print(f"Customer {customer_data['CustomerID']} time has been completed.")

    def ProcessItems(self):
        customers_in_cashier = self.ExtractCustomerData()
        threads = []
        # 为每个顾客创建线程并启动
        for data in customers_in_cashier.values():
            thread = threading.Thread(target=self.process_single_customer, args=(data,))
            threads.append(thread)
            thread.start()
        
        # 如需等待所有顾客处理完成再执行后续逻辑，取消下面注释
        # for thread in threads:
        #     thread.join()

# 主程序
if __name__ == "__main__":
    cashier = Cashier()
    cashier.ProcessItems()
    # 主线程可继续执行其他逻辑，比如生成新顾客
    print("Main thread is running other simulation tasks...")

关键说明 #

• 每个线程独立处理一个顾客的延时，time.sleep()只会阻塞当前线程，不会影响主线程和其他线程。
• 线程适合快速改造原有同步代码，不需要修改太多逻辑，但线程切换有少量性能开销，适合小型模拟场景。

内容的提问来源于stack exchange，提问作者Aman