如何在Repast Python多进程(rank)中模拟订单处理队列？

Repast4Py本身没有开箱即用的SharedQueue类，但基于其MPI并行架构，你可以通过集中式队列管理+MPI消息传递的方式实现跨Rank访问的共享队列，完全适配你的订单处理流程模拟需求。

核心思路 #

指定一个主Rank（比如Rank 0）作为所有队列的唯一管理者，其他Rank上的工人Agent通过MPI消息与主Rank交互，完成订单的取/放操作。这种模式既能保证队列的一致性，又能利用Repast4Py的多Rank并行能力分散工人的计算负载。

具体实现步骤 #

1. 定义订单数据结构 #

class Order:
    def __init__(self, order_id, timestamp):
        self.id = order_id
        self.timestamp = timestamp
        # 记录各阶段完成时间
        self.pickup_finish = None
        self.pack_finish = None
        self.deliver_finish = None

2. 主Rank初始化并维护队列 #

在模型初始化阶段，由主Rank生成所有订单并维护三个阶段的队列：

from collections import deque
import repast4py as rp
from mpi4py import MPI

comm = MPI.COMM_WORLD
rank = comm.Get_rank()
size = comm.Get_size()

# 主Rank（Rank 0）维护队列
if rank == 0:
    # 按下单时间戳排序的待处理订单队列
    incoming_orders = deque()
    # 各阶段队列
    pickup_queue = deque()
    pack_queue = deque()
    deliver_queue = deque()

    # 生成1000个带随机时间戳的订单并按时间戳排序
    rng = rp.random.default_rng()
    orders = [Order(i, rng.integers(0, 1000)) for i in range(1000)]
    orders.sort(key=lambda x: x.timestamp)
    incoming_orders.extend(orders)

3. 实现工人Agent逻辑 #

以拣货工为例，工人空闲时向主Rank请求订单，完成任务后将订单提交到下一阶段队列：

class PickupWorker(rp.agent.Agent):
    def __init__(self, agent_id, rank):
        super().__init__(agent_id, rank)
        self.busy_until = 0  # 当前任务结束时间

    def process_order(self, current_time):
        # 向主Rank请求拣货订单
        comm.send(("get_pickup", self.id), dest=0)
        order = comm.recv(source=0)
        
        if order is not None:
            # 生成10-15分钟的拣货时长
            duration = rp.random.default_rng().integers(10, 16)
            self.busy_until = current_time + duration
            order.pickup_finish = self.busy_until
            # 完成后将订单提交到打包队列
            comm.send(("put_pack", order), dest=0)

    def step(self, model):
        if self.busy_until <= model.current_time:
            self.process_order(model.current_time)

打包工、配送工的逻辑与拣货工类似，仅需修改请求的队列标识（比如get_pack/put_deliver）和任务时长范围。

4. 主Rank的消息处理循环 #

在模型的step方法中，主Rank实时处理其他Rank的订单请求和提交：

class OrderModel(rp.model.Model):
    def __init__(self, props, comm):
        super().__init__(props, comm)
        self.current_time = 0
        # 初始化工人并分布到各Rank
        self.pickup_workers = []
        self.pack_workers = []
        self.deliver_workers = []

        # 10名拣货工
        for i in range(10):
            worker_rank = i % size
            self.pickup_workers.append(PickupWorker(rp.agent.UniqueId(worker_rank, i), worker_rank))
        # 5名打包工、2名配送工的初始化逻辑类似...

    def step(self):
        if self.rank == 0:
            # 处理所有待接收的MPI消息
            status = MPI.Status()
            while comm.Iprobe(status=status):
                msg = comm.recv(source=status.source)
                # 处理拣货工的订单请求
                if msg[0] == "get_pickup":
                    order = incoming_orders.popleft() if incoming_orders else None
                    comm.send(order, dest=status.source)
                # 处理拣货完成的订单提交
                elif msg[0] == "put_pack":
                    pack_queue.append(msg[1])
                # 处理打包工的请求与提交、配送工的请求
                elif msg[0] == "get_pack":
                    order = pack_queue.popleft() if pack_queue else None
                    comm.send(order, dest=status.source)
                elif msg[0] == "put_deliver":
                    deliver_queue.append(msg[1])
                elif msg[0] == "get_deliver":
                    order = deliver_queue.popleft() if deliver_queue else None
                    comm.send(order, dest=status.source)
        
        # 同步所有Rank的当前时间
        self.current_time = comm.bcast(self.current_time, root=0)
        
        # 执行当前Rank上的工人step
        for worker in self.pickup_workers:
            if worker.rank == self.rank:
                worker.step(self)
        # 执行打包工、配送工的step...
        
        # 时间步推进（按分钟计）
        self.current_time += 1

5. 运行模拟 #

和随机游走示例一样，用MPI启动多Rank运行：

mpirun -n 4 python order_processing.py order_config.yaml

关键优势 #

• 避免了Simpy单线程的性能瓶颈，通过多Rank并行分散工人的计算负载
• 集中式队列管理保证了订单处理的顺序一致性，同时通过MPI消息传递实现跨Rank共享访问
• 可以根据Rank数量灵活分配工人，平衡各节点的计算压力

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