1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

import heapq

class Event:
    """事件类，封装事件时间戳和处理动作"""
    def __init__(self, time, action, priority=0, eid=0):
        self.time = time        # 事件发生时刻
        self.action = action    # 事件处理函数
        self.priority = priority  # 次优先级（用户定义）
        self.eid = eid          # 事件ID（打破平局）

    def __lt__(self, other):
        """定义事件排序规则：时间戳优先，其次优先级，最后事件ID"""
        if self.time != other.time:
            return self.time < other.time
        elif self.priority != other.priority:
            return self.priority < other.priority
        else:
            return self.eid < other.eid

# 事件列表（最小堆）
event_list = []

def schedule(time, action, priority=0):
    """将新事件调度到事件列表"""
    eid = next_event_id()  # 生成唯一事件ID
    event = Event(time, action, priority, eid)
    heapq.heappush(event_list, event)

def run_simulation(till=float('inf')):
    """运行仿真直到指定时间或事件列表为空"""
    t = 0
    while event_list and t < till:
        # 取出最早事件
        event = heapq.heappop(event_list)
        t = event.time

        # 执行事件
        event.action()  # 可能向event_list插入新事件

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

import heapq

class DESSimulator:
    """极简离散事件仿真框架"""

    def __init__(self):
        self.time = 0           # 当前时钟
        self.event_queue = []   # 事件队列（最小堆）
        self.event_id = 0       # 事件ID生成器

    def schedule(self, delay, action, priority=0):
        """调度事件：delay时间后执行action"""
        event_time = self.time + delay
        self.event_id += 1
        # 按(时间, 优先级, ID)元组排序
        heapq.heappush(self.event_queue,
                       (event_time, priority, self.event_id, action))

    def run(self, until=None):
        """运行仿真直到until时间或队列空"""
        while self.event_queue:
            # 检查是否到达终止时间
            next_time = self.event_queue[0][0]
            if until is not None and next_time >= until:
                break

            # 取出最早事件
            event_time, priority, eid, action = heapq.heappop(self.event_queue)
            self.time = event_time  # 时钟跳跃

            # 执行事件处理函数
            action(self)  # 传入simulator供调度新事件

# 使用示例
sim = DESSimulator()

def customer_arrival(sim):
    print(f"客户在{sim.time}到达")
    # 调度下一个到达（泊松过程）
    sim.schedule(delay=10, action=customer_arrival)
    # 调度服务完成
    sim.schedule(delay=5, action=service_complete)

def service_complete(sim):
    print(f"服务在{sim.time}完成")

sim.schedule(0, customer_arrival)  # 初始事件
sim.run(until=50)                  # 运行50时间单位

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

import random

def poisson_arrival(lambda_rate, num_events):
    """
    生成泊松过程的到达时间序列

    参数:
        lambda_rate: 到达率（事件/单位时间）
        num_events: 要生成的事件数

    返回:
        到达时间列表
    """
    arrival_times = []
    t = 0

    for i in range(num_events):
        # 采样到达间隔时间（指数分布）
        dt = random.expovariate(lambda_rate)
        t += dt
        arrival_times.append(t)

    return arrival_times

# 示例：模拟银行客户到达
# λ=2客户/分钟，即平均每分钟到达2位客户
arrivals = poisson_arrival(lambda_rate=2.0, num_events=10)

print("客户到达时间序列（分钟）:")
for i, t in enumerate(arrivals, 1):
    print(f"客户{i}: {t:.2f}分钟")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

import numpy as np

# 定义转移矩阵
P = np.array([
    [0.95, 0.05],  # 正常状态的转移概率
    [0.30, 0.70]   # 故障状态的转移概率
])

def simulate_markov_chain(P, initial_state, num_steps):
    """
    仿真马尔可夫链的状态演化

    参数:
        P: 转移概率矩阵
        initial_state: 初始状态
        num_steps: 仿真步数

    返回:
        状态历史记录
    """
    state = initial_state
    history = [state]

    for _ in range(num_steps):
        # 根据当前状态的概率分布采样下一状态
        state = np.random.choice([0, 1], p=P[state])
        history.append(state)

    return history

# 运行仿真
history = simulate_markov_chain(P, initial_state=0, num_steps=1000)

# 统计各状态占比
work_ratio = history.count(0) / len(history)
down_ratio = history.count(1) / len(history)

print(f"正常工作时间占比: {work_ratio:.1%}")
print(f"故障维修时间占比: {down_ratio:.1%}")

# 理论验证（稳态分布）
# 稳态满足 πP = π，可解析求解
pi_1 = 0.05 / (0.05 + 0.30)  # 故障状态稳态概率
pi_0 = 1 - pi_1               # 正常状态稳态概率

print(f"\n理论稳态分布:")
print(f"正常工作: {pi_0:.1%}")
print(f"故障维修: {pi_1:.1%}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

import simpy
import random
import statistics

def mm1_simulation(lambda_rate=0.1, mu_rate=0.15, sim_time=1000, seed=42):
    random.seed(seed)
    env = simpy.Environment()
    counter = simpy.Resource(env, capacity=1)

    wait_times = []

    def customer(env):
        arrival = env.now
        with counter.request() as req:
            yield req
            wait = env.now - arrival
            wait_times.append(wait)
            service = random.expovariate(mu_rate)
            yield env.timeout(service)

    def generator(env):
        while True:
            yield env.timeout(random.expovariate(lambda_rate))
            env.process(customer(env))

    env.process(generator(env))
    env.run(until=sim_time)

    # 计算统计量
    avg_wait = statistics.mean(wait_times) if wait_times else 0
    rho = lambda_rate / mu_rate
    theory_wait = rho / (mu_rate - lambda_rate)

    print(f"利用率 ρ = {rho:.3f}")
    print(f"仿真平均等待: {avg_wait:.2f}秒")
    print(f"理论平均等待: {theory_wait:.2f}秒")
    print(f"Little定律验证 Lq = {rho**2/(1-rho):.3f} ≈ "
          f"λWq = {lambda_rate * avg_wait:.3f}")

mm1_simulation()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

def compare_md1_mm1():
    """对比M/D/1和M/M/1的队列长度"""

    # M/M/1仿真（指数服务）
    wait_mm1 = mm1_simulation(mu_rate=0.15)[0]  # 指数服务

    # M/D/1仿真（确定性服务=6.67秒）
    # 将service采样改为固定值
    # ...

    print(f"M/M/1 平均等待: {wait_mm1:.2f}")
    print(f"M/D/1 平均等待: {wait_md1:.2f}")
    print(f"M/M/1 等待 / M/D/1 等待 = {wait_mm1/wait_md1:.2f}")
    # 理论比 = (1+c_S^2)/2 = (1+1)/2 = 1，但队列长度比=2

1
2
3
4
5
6
7

import simpy

# 创建默认环境，初始时间为0
env = simpy.Environment()

# 创建带初始时间的环境
env_with_offset = simpy.Environment(initial_time=100)

1
2
3

env = simpy.Environment()
print(f"当前仿真时间: {env.now}")           # 输出: 0
print(f"当前活跃进程: {env.active_process}")  # 输出: None（仿真尚未启动）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

def clock(env, name, tick):
    """一个简单的时钟进程，定期打印当前时间"""
    while True:
        print(f"{name} 在时间 {env.now:.1f} 报时")
        yield env.timeout(tick)  # 让出控制权tick时间单位

env = simpy.Environment()
env.process(clock(env, '快钟', 0.5))  # 每0.5单位报时
env.process(clock(env, '慢钟', 1.0))  # 每1.0单位报时

# 运行到指定时间
env.run(until=2)

1
2
3
4
5
6
7
8

env = simpy.Environment()
env.process(clock(env, '调试钟', 1.0))

print("=== 单步调试模式 ===")
for step_num in range(3):
    print(f"\n步骤{step_num+1}前: 当前时间={env.now}, 下一事件在={env.peek()}")
    env.step()  # 仅处理一个事件
    print(f"步骤{step_num+1}后: 当前时间={env.now}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

env = simpy.Environment()

# 类型1: 超时事件（Timeout）—— 让时间流逝
timeout_evt = env.timeout(delay=5, value="延迟结束")
# 该事件在env.now + 5时刻自动触发，携带value

# 类型2: 通用事件（Event）—— 需手动触发
manual_evt = env.event()
# 稍后可调用 manual_evt.succeed(value) 或 .fail(exception)

# 类型3: 进程事件（Process）—— 进程本身即是事件
def my_process(env):
    yield env.timeout(3)
    return "进程结果"

proc_evt = env.process(my_process(env))
# 进程事件在进程结束（return或耗尽）时触发

# 类型4: 条件事件（Condition）—— 组合多个事件
evt1 = env.event()
evt2 = env.event()

# AnyOf：任一事件触发即触发
any_evt = env.any_of([evt1, evt2])

# AllOf：所有事件都触发才触发
all_evt = env.all_of([evt1, evt2])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

def car(env):
    """定义汽车进程的生成器函数"""
    while True:
        print(f"[{env.now}] 汽车开始行驶")
        yield env.timeout(2)  # 行驶2单位时间

        print(f"[{env.now}] 汽车开始停车")
        yield env.timeout(5)  # 停车5单位时间

# 调用car(env)并不执行代码，而是返回生成器对象
env = simpy.Environment()
car_generator = car(env)  # 此时函数体一行都未执行
print(type(car_generator))  # <class 'generator'>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

def lifecycle_demo(env):
    """演示进程完整生命周期的进程"""
    print(f"[{env.now}] 进程启动")
    val1 = yield env.timeout(2, value="第一阶段")
    print(f"[{env.now}] 第一阶段完成，收到值: {val1}")

    val2 = yield env.timeout(3, value="第二阶段")
    print(f"[{env.now}] 第二阶段完成，收到值: {val2}")

    return "进程正常结束"  # 终止并返回结果

env = simpy.Environment()
proc = env.process(lifecycle_demo(env))

# 进程状态检查
print(f"进程是否存活: {proc.is_alive}")  # True
print(f"进程当前等待的事件: {proc.target}")

env.run()

print(f"\n进程是否存活: {proc.is_alive}")  # False
print(f"进程返回值: {proc.value}")  # "进程正常结束"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

def teacher(env):
    """教师进程：讲课45分钟后结束"""
    print(f"[{env.now}] 教师开始讲课")
    yield env.timeout(45)
    print(f"[{env.now}] 教师讲课结束")
    return "知识传授完毕"

def student(env, teacher_process):
    """学生进程：等待教师讲完课"""
    print(f"[{env.now}] 学生就座等待下课")

    # 关键：直接yield进程对象！
    result = yield teacher_process

    print(f"[{env.now}] 学生收到结果: {result}")
    print(f"[{env.now}] 学生离开教室")

env = simpy.Environment()
teacher_proc = env.process(teacher(env))
env.process(student(env, teacher_proc))

env.run()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

env = simpy.Environment()
evt = env.event()

print(f"未触发状态: triggered={evt.triggered}, processed={evt.processed}")
# triggered=False, processed=False

evt.succeed(value="手动触发")
print(f"已触发状态: triggered={evt.triggered}, processed={evt.processed}")
# triggered=True, processed=False（尚未处理）

# 当有进程yield该事件时，它变为processed

1
2
3
4
5
6
7
8
9

# 成功触发
evt.succeed(value="成功结果")

# 失败触发（抛出异常给等待者）
evt.fail(ValueError("出错了"))

# 复制另一事件状态
other_evt = env.event()
evt.trigger(other_evt)  # evt现在与other_evt同状态同值

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

def task(env, name, duration):
    """带名字的任务进程"""
    yield env.timeout(duration)
    return f"{name}完成"

env = simpy.Environment()

# 创建多个任务进程
tasks = [
    env.process(task(env, "任务A", 3)),
    env.process(task(env, "任务B", 5)),
    env.process(task(env, "任务C", 8))
]

# 等待任一任务完成
result = yield env.any_of(tasks)

# result是字典：{已触发的事件: 其值}
completed_event = list(result.keys())[0]
print(f"最快完成的是: {result[completed_event]} 在 {env.now}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

def parallel_workflow(env):
    """并行工作流示例"""
    # 启动多个独立任务
    analysis = env.process(task(env, "需求分析", 2))
    design = env.process(task(env, "系统设计", 4))
    review = env.process(task(env, "评审", 1))

    # 等待全部完成
    results = yield env.all_of([analysis, design, review])

    # results.values()包含所有返回值
    print(f"所有任务在{env.now}完成")
    for val in results.values():
        print(f"  - {val}")

1
2
3
4
5
6

# 简单超时（无返回值）
yield env.timeout(5)  # 暂停5单位时间

# 带值的超时（可作为"定时消息"）
message = yield env.timeout(10, value="时间到！")
print(message)  # "时间到！"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

def customer(env, name, counter):
    """银行客户进程"""
    arrival_time = env.now
    print(f"[{env.now:.1f}] {name} 到达银行")

    # 请求资源（可能等待）
    with counter.request() as request:
        yield request  # 这里会阻塞直到有空闲容量

        # 计算等待时间
        wait = env.now - arrival_time
        print(f"[{env.now:.1f}] {name} 开始服务 (等待了{wait:.1f})")

        # 占用资源进行服务
        service_time = 2.0
        yield env.timeout(service_time)

        print(f"[{env.now:.1f}] {name} 完成服务离开")

# 创建环境和资源
env = simpy.Environment()
counter = simpy.Resource(env, capacity=2)  # 双窗口银行

# 启动多个客户
for i in range(5):
    env.process(customer(env, f"客户{i+1}", counter))
    # 客户间隔到达
    if i < 4:
        yield env.timeout(0.5)

env.run()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

res = simpy.Resource(env, capacity=3)

# 资源基本信息
print(f"总容量: {res.capacity}")        # 3
print(f"当前占用数: {res.count}")       # 当前正在使用数

# 等待队列
print(f"等待者数量: {len(res.queue)}")  # 排队请求数
print(f"等待队列: {res.queue}")         # 列表[Request, ...]

# 正在使用者
print(f"当前用户: {res.users}")         # 列表[Request, ...]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

def priority_customer(env, name, resource, priority):
    """带优先级的客户"""
    with resource.request(priority=priority) as req:
        yield req
        print(f"[{env.now}] {name}(优先级{priority}) 获得服务")
        yield env.timeout(1)

env = simpy.Environment()
res = simpy.PriorityResource(env, capacity=1)

# 故意让低优先级先请求
env.process(priority_customer(env, "普通客户", res, priority=10))
env.process(priority_customer(env, "VIP客户", res, priority=1))  # 高优先级

env.run()
# 输出会显示VIP客户后来但先被服务

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

def preemptible_task(env, name, res, priority, preempt):
    """可被抢占的任务"""
    with res.request(priority=priority, preempt=preempt) as req:
        try:
            yield req
            print(f"[{env.now}] {name} 获得资源")
            yield env.timeout(5)  # 需要5单位时间
            print(f"[{env.now}] {name} 正常完成")
        except simpy.Interrupt as interrupt:
            # 被更高优先级任务抢占
            by = interrupt.cause.by  # 抢占者
            print(f"[{env.now}] {name} 被{by}抢占！")

env = simpy.Environment()
res = simpy.PreemptiveResource(env, capacity=1)

# 长任务（低优先级，可被抢占）
env.process(preemptible_task(env, "长任务", res, priority=10, preempt=True))

# 稍后发高优先级任务（将抢占）
env.process(preemptible_task(env, "紧急任务", res, priority=1, preempt=True))

env.run()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

def gas_station(env, station):
    """加油站管理"""
    while True:
        if station.fuel_tank.level < 100:
            print(f"[{env.now}] 油量低: {station.fuel_tank.level}")
            yield station.fuel_tank.put(500)  # 补给500单位
            print(f"[{env.now}] 补给后: {station.fuel_tank.level}")
        yield env.timeout(15)

def car(env, name, station):
    """汽车加油"""
    print(f"[{env.now}] {name} 到达")
    with station.fuel_pump.request() as req:
        yield req
        yield station.fuel_tank.get(40)  # 加40单位油
        print(f"[{env.now}] {name} 加完油离开")
        yield env.timeout(5)

env = simpy.Environment()

station = type('Station', (), {})()
station.fuel_tank = simpy.Container(env, init=500, capacity=1000)
station.fuel_pump = simpy.Resource(env, capacity=2)

env.process(gas_station(env, station))
for i in range(3):
    env.process(car(env, f"车{i+1}", station))

env.run(until=100)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

def producer(env, store):
    """生产者"""
    for i in range(3):
        item = f"产品{i}"
        yield store.put(item)
        print(f"[{env.now}] 生产了 {item}")
        yield env.timeout(2)

def consumer(env, store):
    """消费者"""
    while True:
        item = yield store.get()
        print(f"[{env.now}] 消费了 {item}")
        yield env.timeout(3)

env = simpy.Environment()
store = simpy.Store(env, capacity=5)

env.process(producer(env, store))
env.process(consumer(env, store))

env.run(until=20)

1
2
3
4
5
6

machine_shop = simpy.FilterStore(env)
machine_shop.put({'type': 'lathe', 'id': 1})
machine_shop.put({'type': 'mill', 'id': 2})

# 只获取车床
lathe = yield machine_shop.get(lambda m: m['type'] == 'lathe')

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

# === Environment ===
env = simpy.Environment(initial_time=0)
env.run(until=10)          # 运行到时间10
env.step()                 # 单步推进
env.peek()                 # 查看下一事件时间
env.now                    # 当前时间
env.process(gen_func(env)) # 创建并启动进程
env.event()                # 创建普通事件
env.timeout(delay)         # 创建超时事件
env.any_of([e1, e2])       # 任一事件
env.all_of([e1, e2])       # 全部事件

# === Process ===
proc = env.process(generator(env))
proc.is_alive              # 是否存活
proc.value                 # 返回值
yield proc                 # 等待进程结束
proc.interrupt(cause)      # 中断进程

# === Event ===
evt = env.event()
evt.succeed(value)         # 成功触发
evt.fail(exception)        # 失败触发
yield evt                  # 等待事件
evt.triggered              # 是否已触发
evt.processed              # 是否已处理

# === Resource ===
res = simpy.Resource(env, capacity=1)
with res.request() as req:
    yield req
    # 使用资源
# 自动释放

# 优先级资源
res = simpy.PriorityResource(env, capacity=1)
with res.request(priority=1) as req:
    yield req

# 抢占式资源
res = simpy.PreemptiveResource(env, capacity=1)
with res.request(priority=1, preempt=True) as req:
    try:
        yield req
    except simpy.Interrupt:
        # 被抢占

# === Container ===
container = simpy.Container(env, init=100, capacity=1000)
yield container.put(50)    # 放入50
yield container.get(30)    # 取出30
container.level            # 当前量

# === Store ===
store = simpy.Store(env, capacity=5)
yield store.put(item)      # 存入
item = yield store.get()   # 取出

1
2
3
4
5

env = simpy.Environment()
env.process(clock(env, '钟', 1.0))
env.run(until=10)
# 内部机制：连续pop事件并执行，直到env.now >= 10
# 执行时间复杂度：O(n_events)，一步到位

1
2
3
4
5
6

env = simpy.Environment()
env.process(clock(env, '钟', 1.0))
for i in range(10):
    env.step()
# 每次仅处理一个事件，精确控制推进
# 可在每步间检查状态、注入外部事件

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

import simpy
import random

def impatient_customer(env, name, counter, patience=5):
    arrival = env.now
    print(f"[{env.now:.1f}] {name} 到达")

    # 方法1：使用条件事件 |
    with counter.request() as req:
        results = yield req | env.timeout(patience)

        if req in results:
            # 获得资源
            wait = env.now - arrival
            print(f"[{env.now:.1f}] {name} 获服务 (等了{wait:.1f})")
            yield env.timeout(random.uniform(2, 5))
            print(f"[{env.now:.1f}] {name} 完成离开")
        else:
            # 超时
            print(f"[{env.now:.1f}] {name} 等不及离开！")

env = simpy.Environment()
counter = simpy.Resource(env, capacity=1)

# 同时到达两位客户
env.process(impatient_customer(env, "客户A", counter))
env.process(impatient_customer(env, "客户B", counter))

env.run()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

def task_preemptive(env, name, res, priority):
    with res.request(priority=priority, preempt=True) as req:
        try:
            yield req
            print(f"[{env.now}] {name} 开始")
            yield env.timeout(5)
            print(f"[{env.now}] {name} 正常完成")
        except simpy.Interrupt:
            print(f"[{env.now}] {name} 被抢断！")

env = simpy.Environment()
res = simpy.PreemptiveResource(env, capacity=1)

env.process(task_preemptive(env, "普通", res, priority=10))
env.process(task_preemptive(env, "紧急", res, priority=1))

env.run()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

def task_direct(env, name, worker_proc):
    yield env.timeout(2)
    worker_proc.interrupt("紧急到来")  # 直接中断另一进程

def worker(env):
    try:
        print(f"[{env.now}] 工作者开始")
        yield env.timeout(5)
        print(f"[{env.now}] 工作者正常完成")
    except simpy.Interrupt as i:
        print(f"[{env.now}] 工作者被{i.cause}中断")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

import simpy
import random
import statistics

# 全局统计容器
wait_times = []       # 每位客户的等待时间
service_times = []    # 每位客户的服务时间
system_times = []     # 每位客户的系统逗留时间

def customer_generator(env, counter, lambda_rate):
    """
    客户生成器进程：按泊松过程持续生成客户

    设计要点：
    - 循环永不终止（while True）
    - 指数分布采样保证泊松过程特性
    - 每次创建独立客户进程（env.process）
    - 到达时刻记录用于后续等待时间计算
    """
    customer_count = 0

    while True:
        # 采样到达间隔时间
        # 指数分布参数λ即为到达率
        interarrival_time = random.expovariate(lambda_rate)

        # 让出控制权，等待间隔时间流逝
        # 这一步至关重要：仿真时钟在这里推进
        yield env.timeout(interarrival_time)

        # 到达时刻（现在时刻）
        arrival_time = env.now
        customer_count += 1

        # 创建客户进程
        # 注意：env.process()将生成器包装为Process对象
        # 并立即调度Initialize事件在当前时刻启动该进程
        env.process(customer_process(
            env,
            f"客户{customer_count}",
            counter,
            arrival_time
        ))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

def customer_process(env, name, counter, arrival_time):
    """
    单个客户的进程定义

    生命周期：
    1. 到达并进入队列（自动，无需显式操作）
    2. 请求柜员资源（可能等待）
    3. 获得资源后接受服务
    4. 服务完成后释放资源并离开

    参数：
        env: 仿真环境
        name: 客户名称（用于追踪输出）
        counter: 柜员资源对象
        arrival_time: 到达时刻（用于计算等待时间）
    """
    # 第一步：打印到达信息
    # 此时env.now应该等于arrival_time
    print(f"[{env.now:6.2f}分] {name} 到达银行")

    # 第二步：请求柜员资源
    # 使用上下文管理器(with)确保资源最终释放
    # request()生成一个Request事件
    with counter.request() as request:
        # 等待请求被满足（可能阻塞）
        # 若资源有空闲容量，立即获得
        # 否则进入counter.queue等待队列
        yield request

        # 到达这里说明已获得柜员
        # 计算等待时间
        wait_time = env.now - arrival_time
        wait_times.append(wait_time)

        print(f"[{env.now:6.2f}分] {name} 开始服务 "
              f"(等待了 {wait_time:.2f} 分钟)")

        # 第三步：接受服务
        # 采样服务时间（指数分布）
        service_rate = 0.15  # μ=0.15客户/分钟
        service_time = random.expovariate(service_rate)
        service_times.append(service_time)

        # 占用柜员服务时间
        yield env.timeout(service_time)

        # 第四步：服务完成离开
        # 计算系统逗留时间
        system_time = env.now - arrival_time
        system_times.append(system_time)

        print(f"[{env.now:6.2f}分] {name} 完成服务离开 "
              f"(逗留了 {system_time:.2f} 分钟)")

    # 离开with块，资源自动释放
    # SimPy会检查counter.queue，唤醒下一个等待的客户

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

def monitor(env, counter, interval=10):
    """
    监控进程：定期报告队列状态

    设计考量：
    - 定期采样提供系统动态视角
    - 打印队列长度、柜员状态、利用率
    - 实际应用中可将数据存入文件用于后续分析
    """
    while True:
        yield env.timeout(interval)

        # 收集当前状态
        queue_length = len(counter.queue)
        counter_status = "繁忙" if counter.count == 1 else "空闲"
        utilization = counter.count / counter.capacity

        print(f"\n{'='*40}")
        print(f"监控报告 @ {env.now:.2f}分钟")
        print(f"  队列长度: {queue_length} 人")
        print(f"  柜员状态: {counter_status}")
        print(f"  利用率: {utilization:.1%}")
        print(f"{'='*40}\n")

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

def run_bank_simulation(sim_time=500, lambda_rate=0.1, mu_rate=0.15, seed=42):
    """
    运行完整的银行仿真

    参数：
        sim_time: 仿真总时长
        lambda_rate: 客户到达率
        mu_rate: 柜员服务率
        seed: 随机数种子（结果可重复）
    """
    # 设置随机种子
    random.seed(seed)

    # 清空统计容器
    global wait_times, service_times, system_times
    wait_times = []
    service_times = []
    system_times = []

    # 创建仿真环境
    env = simpy.Environment()

    # 创建柜员资源
    # capacity=1 表示单服务台
    counter = simpy.Resource(env, capacity=1)

    # 启动客户生成器进程
    env.process(customer_generator(env, counter, lambda_rate))

    # 启动监控进程
    env.process(monitor(env, counter, interval=50))

    # 打印仿真参数
    print("=" * 60)
    print("银行排队系统仿真 - M/M/1模型")
    print("=" * 60)
    print(f"仿真参数：")
    print(f"  到达率 λ = {lambda_rate} 客户/分钟")
    print(f"  服务率 μ = {mu_rate} 客户/分钟")
    print(f"  利用率 ρ = λ/μ = {lambda_rate/mu_rate:.3f}")
    print(f"  仿真时长 = {sim_time} 分钟")
    print("=" * 60)
    print("\n仿真开始...\n")

    # 运行仿真
    env.run(until=sim_time)

    # ========== 仿真结束，统计分析 ==========
    print("\n" + "=" * 60)
    print("仿真结束 - 统计分析")
    print("=" * 60)

    # 等待时间统计
    if wait_times:
        print(f"\n【等待时间统计】")
        print(f"  样本量: {len(wait_times)} 位客户")
        print(f"  平均等待时间: {statistics.mean(wait_times):.2f} 分钟")
        print(f"  标准差: {statistics.stdev(wait_times):.2f} 分钟")
        print(f"  最大等待时间: {max(wait_times):.2f} 分钟")
        print(f"  最小等待时间: {min(wait_times):.2f} 分钟")

    # 服务时间统计
    if service_times:
        print(f"\n【服务时间统计】")
        print(f"  平均服务时间: {statistics.mean(service_times):.2f} 分钟")
        print(f"  理论均值: {1/mu_rate:.2f} 分钟")  # E[S] = 1/μ

    # 系统逗留时间统计
    if system_times:
        print(f"\n【系统逗留时间统计】")
        print(f"  平均逗留时间: {statistics.mean(system_times):.2f} 分钟")

    # ========== 理论对比 ==========
    print("\n" + "=" * 60)
    print("理论指标对比 - M/M/1模型")
    print("=" * 60)

    rho = lambda_rate / mu_rate

    print(f"\n【理论预测值】")
    print(f"  利用率 ρ = {rho:.3f}")
    print(f"  空闲概率 p₀ = {1-rho:.3f}")
    print(f"  平均队列长度 Lq = {rho**2/(1-rho):.3f} 人")
    print(f"  平均系统长度 L = {rho/(1-rho):.3f} 人")
    print(f"  平均等待时间 Wq = {rho/(mu_rate-lambda_rate):.3f} 分钟")
    print(f"  平均逗留时间 W = {1/(mu_rate-lambda_rate):.3f} 分钟")

    # ========== Little定律验证 ==========
    print("\n" + "=" * 60)
    print("Little定律验证")
    print("=" * 60)

    if wait_times and system_times:
        avg_wait = statistics.mean(wait_times)
        avg_system = statistics.mean(system_times)

        # L = λW
        L_theory = rho / (1 - rho)
        L_from_W = lambda_rate * avg_system

        # Lq = λWq
        Lq_theory = rho**2 / (1 - rho)
        Lq_from_W = lambda_rate * avg_wait

        print(f"\n【Little定律: L = λW】")
        print(f"  理论 L = {L_theory:.3f}")
        print(f"  估计 L = λ × W = {L_from_W:.3f}")
        print(f"  相对误差 = {abs(L_theory-L_from_W)/L_theory:.1%}")

        print(f"\n【Little定律: Lq = λWq】")
        print(f"  理论 Lq = {Lq_theory:.3f}")
        print(f"  估计 Lq = λ × Wq = {Lq_from_W:.3f}")
        print(f"  相对误差 = {abs(Lq_theory-Lq_from_W)/Lq_theory:.1%}")

if __name__ == "__main__":
    run_bank_simulation()