SimPy订单履约模拟：单操作员整箱扫描的并行执行问题

• 业务场景：装有订单物品的箱子到货后，需由单名操作员完成箱子扫描+箱内所有物品扫描，扫描后的物品送入传送带，最终分配至对应订单的推车；配置5名扫描操作员，要求支持多箱并行处理。
• 现有问题：
  1. 按box_id分组传入进程后，箱子仍串行处理，无法实现多箱并行。
  2. 时间记录异常：仅box_scan_done和item_scan_done字段记录正确，box_scan_queue、box_scan_start、item_scan_queue、item_scan_start的时间记录存在异常。
• 核心需求：实现单操作员负责整箱的扫描全流程，同时支持多箱并行处理，修复时间记录逻辑。

现有代码片段 #

环境初始化代码 #

env = simpy.Environment()
scan_op = simpy.Resource(env, capacity=5)  # 负责扫描箱子和物品的操作员资源
cart_op = simpy.Resource(env, capacity=3)  # 处理物品和清空推车的操作员资源
belt = simpy.Container(env, init=0)         # 传送带容器

carts = {}  
for order in order_list:  # order_list是包含(order_id, 订单数量)的元组列表
   carts[order[0]]= simpy.Container(env, capacity=order[1], init=0) 

到货处理进程 #

def arrival_activity(df, env, scan_op, cart_op, belt, carts,\
                      min_box, max_box, mode_box, min_item, max_item, mode_item,\
                      min_cart, max_cart, mode_cart, mean_belt, std_belt, emptying_threshold):
    box_df = df.groupby('box_id')
    for _, box in box_df:
        box_process = box_activities(df, env, box, scan_op, cart_op, belt, carts,\
                      min_box, max_box, mode_box, min_item, max_item, mode_item,\
                      min_cart, max_cart, mode_cart, mean_belt, std_belt, emptying_threshold)
        yield(env.process(box_process))

箱子处理进程 #

def box_activities(df, env, box, scan_op, cart_op, belt, carts,\
                      min_box, max_box, mode_box, min_item, max_item, mode_item,\
                      min_cart, max_cart, mode_cart, mean_belt, std_belt, emptying_threshold):
    with scan_op.request() as req:
        for index, _ in box.iterrows():
            if not df.loc[index,'Scanned']:
               box_scan_time = np.random.triangular(min_box, mode_box, max_box)
               yield env.timeout(box_scan_time)
               df.loc[df['box_id'] == df.loc[index,'box_id'], 'Scanned'] = True                
               df.loc[df['box_id'] == df.loc[index,'box_id'], 'box_scan_done'] = env.now
            item_scan_time =np.random.triangular(min_item, mode_item, max_item)
            yield env.timeout(item_scan_time)
            df.loc[index,'item_scan_done']= env.now
            convey = belt_activity(df, env, index, scan_op, cart_op, belt, carts,\
                      min_box, max_box, mode_box, min_item, max_item, mode_item,\
                      min_cart, max_cart, mode_cart, mean_belt, std_belt, emptying_threshold) 
            env.process(convey)
            yield req   

1. 多箱并行失效修复 #

在arrival_activity中，yield(env.process(box_process))会强制等待当前箱子进程完成后才启动下一个，直接阻断了并行逻辑。正确做法是直接启动进程而不等待：

修改后的arrival_activity：

def arrival_activity(df, env, scan_op, cart_op, belt, carts,\
                      min_box, max_box, mode_box, min_item, max_item, mode_item,\
                      min_cart, max_cart, mode_cart, mean_belt, std_belt, emptying_threshold):
    box_df = df.groupby('box_id')
    for _, box in box_df:
        # 直接启动箱子处理进程，不等待完成，实现多箱并行
        env.process(box_activities(df, env, box, scan_op, cart_op, belt, carts,\
                      min_box, max_box, mode_box, min_item, max_item, mode_item,\
                      min_cart, max_cart, mode_cart, mean_belt, std_belt, emptying_threshold))

2. 时间记录与扫描逻辑修复 #

当前box_activities存在两处核心问题：

• 循环遍历物品时重复判断箱子是否扫描，逻辑冗余且导致时间记录时机错误。
• 循环内的yield req会导致每扫描一个物品就释放并重新请求操作员，违背“单操作员负责整箱”的需求。

修改后的box_activities：

def box_activities(df, env, box, scan_op, cart_op, belt, carts,\
                      min_box, max_box, mode_box, min_item, max_item, mode_item,\
                      min_cart, max_cart, mode_cart, mean_belt, std_belt, emptying_threshold):
    box_id = box.iloc[0]['box_id']
    # 提前获取操作员资源，确保单操作员全程负责整箱处理
    with scan_op.request() as req:
        # 记录箱子进入扫描队列的时间
        df.loc[df['box_id'] == box_id, 'box_scan_queue'] = env.now
        # 等待获取操作员，记录扫描开始时间
        yield req
        df.loc[df['box_id'] == box_id, 'box_scan_start'] = env.now
        
        # 执行箱子扫描
        box_scan_time = np.random.triangular(min_box, mode_box, max_box)
        yield env.timeout(box_scan_time)
        df.loc[df['box_id'] == box_id, 'Scanned'] = True                
        df.loc[df['box_id'] == box_id, 'box_scan_done'] = env.now
        
        # 遍历箱内物品逐个扫描
        for index, _ in box.iterrows():
            # 因操作员已占用，物品无需等待，队列时间与开始时间一致
            df.loc[index, 'item_scan_queue'] = env.now
            df.loc[index, 'item_scan_start'] = env.now
            
            item_scan_time = np.random.triangular(min_item, mode_item, max_item)
            yield env.timeout(item_scan_time)
            df.loc[index, 'item_scan_done'] = env.now
            
            # 异步启动传送带处理，不阻塞当前扫描流程
            env.process(belt_activity(df, env, index, scan_op, cart_op, belt, carts,\
                      min_box, max_box, mode_box, min_item, max_item, mode_item,\
                      min_cart, max_cart, mode_cart, mean_belt, std_belt, emptying_threshold))

关键修复点说明： #

• 将操作员请求的上下文管理器放在整箱处理逻辑外层，确保单操作员全程负责，不会中途释放资源。
• 补充所有时间字段的记录，时机符合业务逻辑：
  • box_scan_queue：请求操作员前的时间（进入队列等待）。
  • box_scan_start：获取到操作员后的时间（开始扫描箱子）。
  • item_scan_queue/item_scan_start：因操作员已占用，无需等待，直接记录当前时间。
• 移除循环内的yield req，避免重复请求/释放操作员。

3. 额外注意事项 #

• 提前初始化DataFrame中的所有时间字段（box_scan_queue等），避免出现缺失值。
• 确保belt_activity逻辑不会阻塞当前扫描流程（当前代码已用env.process异步启动，无需担心）。

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