企业官网建设流程全解析

用Python模拟银行排队：手把手教你写一个离散事件模拟（DES）小项目

银行大厅里此起彼伏的叫号声、不断变化的等待人数显示屏、柜台工作人员忙碌的身影——这些日常场景背后隐藏着一个有趣的计算机科学概念：离散事件模拟（DES）。不同于连续系统模拟需要跟踪每一时刻的状态变化，DES只关注那些改变系统状态的关键时刻点，就像银行系统中客户到达、开始服务、结束服务这些离散事件。

今天我们将用Python构建一个完整的银行排队模拟器，从零开始实现事件调度、队列管理和性能分析。这个项目特别适合已经掌握Python基础语法，想通过实战理解系统建模的开发者。不需要高等数学基础，我们会用最直观的代码展现DES的核心思想。

1. 环境准备与基础建模

开始前确保已安装Python 3.6+环境，我们需要用到以下几个标准库：

import heapq # 事件堆管理 import random # 随机数生成 from collections import defaultdict # 数据统计

1.1 定义系统核心组件

银行排队系统可以抽象为三个关键要素：

• 客户：携带到达时间、服务时长等属性
• 服务窗口：有限资源，同一时间只能服务一个客户
• 事件引擎：驱动整个模拟过程的时间轴

首先创建客户类：

class Customer: def __init__(self, arrival_time): self.arrival_time = arrival_time self.service_duration = random.expovariate(1.0/AVG_SERVICE_TIME) self.service_start = None # 将在被服务时记录

1.2 初始化模拟参数

这些参数可以根据实际情况调整：

SIMULATION_TIME = 480 # 8小时工作制(分钟) NUM_WINDOWS = 4 # 开放窗口数量 AVG_ARRIVAL_INTERVAL = 2.5 # 平均每2.5分钟来一位客户 AVG_SERVICE_TIME = 5.0 # 平均服务时长(分钟)

提示：指数分布(expovariate)非常适合模拟客户到达间隔和服务时长，它符合泊松过程的特点——事件独立发生且平均速率恒定。

2. 事件调度系统实现

DES的核心是事件调度算法，我们需要一个优先级队列来管理未来事件。

2.1 事件类型定义

银行系统中主要有两类事件：

class Event: def __init__(self, event_type, time, customer=None): self.type = event_type # 'ARRIVAL' 或 'DEPARTURE' self.time = time # 事件发生时间 self.customer = customer # 关联客户对象 def __lt__(self, other): return self.time < other.time # heapq比较用

2.2 事件循环引擎

这是整个模拟的"心脏"：

def run_simulation(): event_queue = [] current_time = 0 windows = [None] * NUM_WINDOWS # 跟踪窗口状态 queue = [] # 等待队列 stats = defaultdict(list) # 收集统计数据 # 生成第一个到达事件 first_arrival = random.expovariate(1.0/AVG_ARRIVAL_INTERVAL) heapq.heappush(event_queue, Event('ARRIVAL', first_arrival)) while event_queue and current_time < SIMULATION_TIME: event = heapq.heappop(event_queue) current_time = event.time if event.type == 'ARRIVAL': handle_arrival(event, current_time, windows, queue, event_queue, stats) # 安排下一个到达事件 next_arrival = current_time + random.expovariate(1.0/AVG_ARRIVAL_INTERVAL) heapq.heappush(event_queue, Event('ARRIVAL', next_arrival)) elif event.type == 'DEPARTURE': handle_departure(event, current_time, windows, queue, event_queue, stats)

3. 事件处理逻辑实现

3.1 处理客户到达

当新客户到达时：

def handle_arrival(event, current_time, windows, queue, event_queue, stats): customer = Customer(current_time) # 检查是否有空闲窗口 free_window = next((i for i, w in enumerate(windows) if w is None), None) if free_window is not None: # 直接服务 windows[free_window] = customer customer.service_start = current_time departure_time = current_time + customer.service_duration heapq.heappush(event_queue, Event('DEPARTURE', departure_time, customer)) else: # 加入队列 queue.append(customer) # 记录队列长度 stats['queue_length'].append((current_time, len(queue)))

3.2 处理服务完成

当客户完成服务时：

def handle_departure(event, current_time, windows, queue, event_queue, stats): # 记录客户等待数据 customer = event.customer wait_time = customer.service_start - customer.arrival_time stats['wait_times'].append(wait_time) # 释放窗口 window_idx = windows.index(customer) windows[window_idx] = None # 检查队列中是否有等待客户 if queue: next_customer = queue.pop(0) next_customer.service_start = current_time windows[window_idx] = next_customer departure_time = current_time + next_customer.service_duration heapq.heappush(event_queue, Event('DEPARTURE', departure_time, next_customer)) # 更新窗口利用率统计 busy_windows = sum(1 for w in windows if w is not None) stats['utilization'].append((current_time, busy_windows/NUM_WINDOWS))

4. 结果分析与可视化

模拟结束后，我们可以提取关键指标并进行可视化展示。

4.1 计算核心指标

def analyze_results(stats): avg_wait = sum(stats['wait_times'])/len(stats['wait_times']) max_wait = max(stats['wait_times']) # 窗口利用率 = 总忙碌时间 / (窗口数 × 模拟时长) total_busy_time = sum( (t2-t1)*util for (t1, util), (t2, _) in zip(stats['utilization'][:-1], stats['utilization'][1:]) ) avg_utilization = total_busy_time / (NUM_WINDOWS * SIMULATION_TIME) print(f"平均等待时间: {avg_wait:.1f} 分钟") print(f"最长等待时间: {max_wait:.1f} 分钟") print(f"窗口平均利用率: {avg_utilization*100:.1f}%")

4.2 使用Matplotlib可视化

import matplotlib.pyplot as plt def plot_results(stats): # 等待时间分布 plt.figure(figsize=(12, 4)) plt.subplot(1, 2, 1) plt.hist(stats['wait_times'], bins=20, edgecolor='black') plt.title('客户等待时间分布') plt.xlabel('分钟') # 队列长度变化 times, lengths = zip(*stats['queue_length']) plt.subplot(1, 2, 2) plt.step(times, lengths, where='post') plt.title('实时队列长度') plt.xlabel('模拟时间(分钟)') plt.ylabel('等待客户数') plt.tight_layout() plt.show()

5. 模型优化与扩展建议

基础模型运行后，可以考虑以下增强方向：

5.1 多队列 vs 单队列

现实银行通常采用单队列多窗口模式，而超市多是多队列单窗口。我们可以修改模型比较两种策略：

# 在初始化时改为多个队列 queues = [[] for _ in range(NUM_WINDOWS)] # 客户到达时选择最短队列 shortest = min(queues, key=len) shortest.append(customer)

5.2 动态窗口管理

模拟午间高峰时段的窗口调整策略：

def dynamic_windows(current_time, windows): # 11:30-13:30开放更多窗口 if 210 <= current_time <= 330: # 11:30-13:30 return windows + [None] * 2 # 增加2个临时窗口 return windows

5.3 VIP客户优先级

添加5%的VIP客户优先服务：

class VIPCustomer(Customer): def __init__(self, arrival_time): super().__init__(arrival_time) self.is_vip = True # 在队列处理时优先服务VIP queue.sort(key=lambda c: getattr(c, 'is_vip', False), reverse=True)

这个项目最有趣的部分是调整参数观察系统行为变化。当我将AVG_ARRIVAL_INTERVAL设为2.0而保持服务时间不变时，系统很快出现排队积压——这正是DES帮助我们预测系统瓶颈的价值所在。