面向对象Python实现超市收银队列模拟问题优化求助

需求说明 #

• 高效服务30位顾客，目标是总耗时最短
• 收银员配置：
  • Cashier 1：每件商品扫描耗时20秒
  • Cashier 2：每件商品扫描耗时30秒
  • Cashier 3：每件商品扫描耗时40秒
• 顾客规则：
  • 到达间隔为1-5分钟随机（60-300秒）
  • 购买商品数量为1-10件随机
  • 到达后选择等待时间最短的收银台排队
• 代码限制：禁止使用enumeration、lambda及time模块，需通过类对象实现队列逻辑

用户原代码 #

import random

class Customer:
  def __init__(self, id, arrivalTime):
    self.id = id
    self.arrivalTime = arrivalTime
    self.items = random.randit(1,10)  # 拼写错误：randit → randint

class Cashier:
  def __init__(self, id, scanningTime):
    self.id = id
    self.scanningTime = scanningTime
    self.currentCust = None
    self.finish = 0
    self.queue = []  # 原代码缺失：收银台需要独立队列

  def serveCust(self, customer, currentTime):
    self.currentCust = customer
    self.finish = currentTime + customer.items * self.scanningTime

  def is_available(self, currentTime):
    return self.finish <= currentTime  # 原代码转义错误：&lt;= → <=

  def finish_serv(self):
    customer = self.currentCust
    self.currentCust = None
    self.finish = 0 
    return customer

class BQ:
  def __init__(self, numCashier, cashierScanTime):
    self.customers = []
    self.cashiers = []
    for i in range(numCashier):
      self.cashiers.append(Cashier(i, cashierScanTime[i]))
    self.numCustServed = 0
    self.totalWaitTime = 0

  def addCust(self, currentTime):
    customer = Customer(len(self.customers), currentTime)
    self.customers.append(customer)

  def serveCust(self, currentTime):
    for cashier in self.cashiers:
      if cashier.is_available(currentTime) and self.customers:
        minItemsCust = self.customers[0]
        for customer in self.customers:
          if customer.items < minItemsCust.items:
            minItemsCust = customer
        self.customers.remove(minItemsCust)
        self.totalWaitTime += currentTime - minItemsCust.arrivalTime
        cashier.serveCust(minItemsCust, currentTime)
        self.numCustServed += 1

  def run(self, numCust):
    currentTime = 0
    for i in range(numCust):
      self.addCust(currentTime)
      currentTime += random.randit(60,300)  # 拼写错误：randit → randint
    while self.customers:
      self.serveCust(currentTime)
      currentTime += 1
    averageWaitTime = self.totalWaitTime / self.numCustServed

原代码问题分析 #

1. 拼写错误：random.randit 应为 random.randint，会导致运行报错
2. 逻辑不符合需求：原代码将所有顾客放在一个全局队列，给空闲收银员分配商品最少的顾客，但需求是顾客主动选择等待时间最短的收银台（每个收银台需维护独立队列）
3. 时间推进低效：每次循环+1秒模拟，效率极低，应直接跳转到下一个事件节点（收银员完成服务/新顾客到达）
4. 缺少核心逻辑：未实现“顾客选择等待时间最短收银台”的逻辑
5. 无结果输出：计算了平均等待时间但未返回或打印，无法查看模拟结果

修复优化后的代码 #

import random

class Customer:
    def __init__(self, cust_id, arrival_time):
        self.id = cust_id
        self.arrival_time = arrival_time
        self.items = random.randint(1, 10)
        self.start_service_time = 0  # 开始服务的时间
        self.wait_time = 0  # 等待时间

class Cashier:
    def __init__(self, cashier_id, scan_time_per_item):
        self.id = cashier_id
        self.scan_time = scan_time_per_item
        self.queue = []  # 排队的顾客列表
        self.current_customer = None  # 当前正在服务的顾客
        self.next_available_time = 0  # 下一次可用的时间

    def get_total_wait_time(self, current_time):
        # 计算新顾客加入后需要等待的总时间
        total = 0
        if self.next_available_time > current_time:
            total += self.next_available_time - current_time
        for cust in self.queue:
            total += cust.items * self.scan_time
        return total

    def process_customer(self, current_time):
        # 如果当前空闲且队列有顾客，开始服务下一位
        if self.next_available_time <= current_time and self.queue:
            self.current_customer = self.queue.pop(0)
            self.current_customer.start_service_time = current_time
            self.current_customer.wait_time = current_time - self.current_customer.arrival_time
            self.next_available_time = current_time + self.current_customer.items * self.scan_time
            return self.current_customer
        return None

class CheckoutSystem:
    def __init__(self, cashier_scan_times):
        self.cashiers = []
        # 初始化收银员，用索引循环避免enumerate
        for i in range(len(cashier_scan_times)):
            self.cashiers.append(Cashier(i, cashier_scan_times[i]))
        self.total_wait_time = 0
        self.served_customers = 0

    def assign_customer(self, customer, current_time):
        # 找到等待时间最短的收银台
        min_wait = float('inf')
        selected_cashier = self.cashiers[0]
        for cashier in self.cashiers:
            wait_time = cashier.get_total_wait_time(current_time)
            if wait_time < min_wait:
                min_wait = wait_time
                selected_cashier = cashier
        selected_cashier.queue.append(customer)

    def run(self, num_customers):
        # 生成所有顾客到达事件（时间+顾客ID）
        events = []
        current_time = 0
        for cust_id in range(num_customers):
            events.append( ('arrival', current_time, cust_id) )
            # 生成下一位顾客到达时间
            current_time += random.randint(60, 300)

        # 添加所有收银员的初始可用事件
        for cashier in self.cashiers:
            events.append( ('service_done', cashier.next_available_time, cashier) )

        # 手动排序事件（按时间从小到大），禁止用lambda
        for i in range(len(events)):
            for j in range(i+1, len(events)):
                if events[i][1] > events[j][1]:
                    events[i], events[j] = events[j], events[i]

        while self.served_customers < num_customers:
            # 取出最早的事件
            event_type, event_time, obj = events.pop(0)
            current_time = event_time

            if event_type == 'arrival':
                # 创建顾客并分配到最优收银台
                customer = Customer(obj, current_time)
                self.assign_customer(customer, current_time)
                # 找到该顾客所在的收银台，添加服务完成事件
                for cashier in self.cashiers:
                    if customer in cashier.queue:
                        next_time = cashier.next_available_time
                        if cashier.next_available_time <= current_time and cashier.queue:
                            cashier.process_customer(current_time)
                            next_time = cashier.next_available_time
                        events.append( ('service_done', next_time, cashier) )
                        break
                # 重新排序事件
                for i in range(len(events)):
                    for j in range(i+1, len(events)):
                        if events[i][1] > events[j][1]:
                            events[i], events[j] = events[j], events[i]

            elif event_type == 'service_done':
                cashier = obj
                # 处理完成服务的顾客，记录等待时间
                if cashier.current_customer is not None:
                    self.total_wait_time += cashier.current_customer.wait_time
                    self.served_customers += 1
                    cashier.current_customer = None
                # 尝试处理队列中的下一位顾客
                cashier.process_customer(current_time)
                # 添加新的服务完成事件
                events.append( ('service_done', cashier.next_available_time, cashier) )
                # 重新排序事件
                for i in range(len(events)):
                    for j in range(i+1, len(events)):
                        if events[i][1] > events[j][1]:
                            events[i], events[j] = events[j], events[i]

        # 计算并输出结果
        average_wait = self.total_wait_time / num_customers
        print(f"模拟完成，共服务{num_customers}位顾客")
        print(f"平均等待时间：{average_wait:.2f}秒")
        # 计算总耗时：最后一个顾客完成服务的时间
        max_finish_time = 0
        for cashier in self.cashiers:
            if cashier.next_available_time > max_finish_time:
                max_finish_time = cashier.next_available_time
        print(f"总服务耗时：{max_finish_time/60:.2f}分钟")

# 运行模拟
if __name__ == "__main__":
    # 收银员扫描时间：Cashier1=20s，Cashier2=30s，Cashier3=40s
    system = CheckoutSystem([20, 30, 40])
    system.run(30)

优化说明 #

1. 修正拼写错误：将random.randit改为random.randint
2. 实现需求逻辑：每个收银台维护独立队列，顾客到达时计算所有收银台的等待时间，选择最短的加入
3. 事件驱动模拟：通过事件列表（顾客到达、收银员完成服务）推进时间，直接跳转到下一个事件节点，大幅提升效率
4. 完善数据记录：记录每位顾客的等待时间，最终计算平均等待时间和总服务耗时
5. 符合代码限制：未使用enumeration、lambda及time模块，通过手动排序实现事件排序

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