求助：基于Simpy实现带可中断启动时间的批量作业系统模拟

核心逻辑实现 #

以下代码完全匹配需求：服务器需启动时间、凑齐M个客户触发启动、系统满员客户balk、完成批量后根据队列状态决定服务器启停、可中断启动中的服务器。

import simpy
import random

# 配置参数
M = 2  # 触发服务器启动的批量客户数
K = 50  # 系统客户容量上限
T_ARRIVAL = 1  # 客户平均到达间隔
T_STARTUP = 2  # 服务器启动耗时
T_SERVICE = 3  # 每批客户服务耗时

class Server:
    def __init__(self, env, name):
        self.env = env
        self.name = name
        self.is_running = False  # 是否已启动完成并运行
        self.startup_event = None  # 记录启动事件，用于中断
        self.process = env.process(self.run())

    def run(self):
        while True:
            # 等待外部触发信号
            yield self.env.event()
            if self.is_running:
                # 服务器已就绪，直接处理下一批
                self._process_batch()
                continue
            
            # 启动服务器
            print(f"{self.env.now:.2f}: 服务器{self.name}开始启动")
            self.startup_event = self.env.timeout(T_STARTUP)
            try:
                yield self.startup_event
                self.is_running = True
                print(f"{self.env.now:.2f}: 服务器{self.name}启动完成")
                self._process_batch()
            except simpy.Interrupt:
                # 启动被中断，重置状态
                print(f"{self.env.now:.2f}: 服务器{self.name}启动被中断")
                self.startup_event = None

    def _process_batch(self):
        # 从等待队列取出M个客户
        batch = [yield waiting_customers.get() for _ in range(M)]
        print(f"{self.env.now:.2f}: 服务器{self.name}开始处理客户{[c['id'] for c in batch]}")
        yield self.env.timeout(T_SERVICE)
        print(f"{self.env.now:.2f}: 服务器{self.name}完成客户{[c['id'] for c in batch]}的服务")
        
        # 检查等待队列是否仍有足够客户，决定是否继续运行
        if len(waiting_customers.items) >= M:
            # 中断当前等待，直接处理下一批
            self.process.interrupt()
        else:
            self.is_running = False
            print(f"{self.env.now:.2f}: 服务器{self.name}关闭")

def customer_generator(env):
    customer_id = 0
    while True:
        yield env.timeout(random.expovariate(1/T_ARRIVAL))
        customer_id += 1
        # 计算系统内总客户数：等待队列 + 正在服务的客户数
        total_in_system = len(waiting_customers.items) + sum(M for s in servers if s.is_running)
        if total_in_system >= K:
            print(f"{env.now:.2f}: 客户{customer_id}到达，系统已满，balk")
            continue
        print(f"{env.now:.2f}: 客户{customer_id}到达并等待")
        yield waiting_customers.put({'id': customer_id})
        
        # 凑够M个客户时，触发服务器调度
        if len(waiting_customers.items) >= M:
            # 优先使用已运行的服务器
            ready_server = next((s for s in servers if s.is_running), None)
            if ready_server:
                ready_server.process.interrupt()
                continue
            
            # 找未启动的空闲服务器
            idle_server = next((s for s in servers if not s.is_running and s.startup_event is None), None)
            if idle_server:
                idle_server.process.interrupt()

# 初始化模拟环境
env = simpy.Environment()
waiting_customers = simpy.Store(env)
servers = [Server(env, "1"), Server(env, "2")]  # 配置2台服务器

# 启动客户生成流程
env.process(customer_generator(env))

# 运行模拟
env.run(until=50)

关键逻辑说明 #

• 服务器启停控制：Server类通过is_running和startup_event维护状态，启动过程中可被Interrupt中断，适配启动中服务器被关闭的场景。
• 批量触发机制：客户加入队列后，若数量达到M，优先调度已运行的服务器，无可用时启动空闲服务器。
• 容量限制：通过计算等待队列+服务中客户的总数判断是否达到K，超过则触发balk逻辑。
• 服务后状态判断：每批服务完成后，检查队列剩余客户数，≥M则继续处理下一批，否则关闭服务器。

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