Лаб. работа “Системы массового обслуживания”
Введение в системы массового обслуживания
Системы массового обслуживания (СМО) представляют собой математические модели, которые используются для анализа и оптимизации процессов обслуживания клиентов в различных организациях. Эти системы помогают прогнозировать и улучшать эффективность обслуживания, оптимизировать количество обслуживающего персонала и ресурсов, а также учитывать важные характеристики, такие как время ожидания и уровень обслуживания.
Основные компоненты СМО:
- Поток заявок — клиенты или задачи, поступающие в систему
- Очередь — место ожидания для заявок, если все каналы обслуживания заняты
- Каналы обслуживания — устройства или персонал, обрабатывающий заявки
- Дисциплина обслуживания — правила выбора следующей заявки из очереди
Классификация систем массового обслуживания (нотация Кендалла):
A/B/c/K/N/D
- A — закон распределения интервалов поступления заявок
- B — закон распределения времени обслуживания
- c — количество каналов обслуживания
- K — максимальное количество заявок в системе
- N — количество источников заявок
- D — дисциплина обслуживания
Теоретические основы
Модель M/M/1
Модель M/M/1 описывает систему с одним каналом обслуживания, пуассоновским потоком заявок и экспоненциальным временем обслуживания.
Основные параметры:
- λ (лямбда) — интенсивность поступления заявок (заявок/ед. времени)
- μ (мю) — интенсивность обслуживания (заявок/ед. времени)
- ρ = λ/μ — коэффициент загрузки системы
Условие стационарности: ρ < 1 (интенсивность поступления меньше интенсивности обслуживания)
Основные характеристики:
Вероятность простоя системы: \[ P_0 = 1 - \rho \]
Среднее число заявок в системе: \[ L = \frac{\rho}{1 - \rho} \]
Среднее число заявок в очереди: \[ L_q = \frac{\rho^2}{1 - \rho} \]
Среднее время пребывания в системе: \[ W = \frac{1}{\mu - \lambda} \]
Среднее время ожидания в очереди: \[ W_q = \frac{\lambda}{\mu(\mu - \lambda)} \]
Модель M/M/c
Модель M/M/c описывает систему с c параллельными каналами обслуживания.
Основные характеристики:
Вероятность простоя системы: \[ P_0 = \left[\sum_{k=0}^{c-1} \frac{(c\rho)^k}{k!} + \frac{(c\rho)^c}{c!(1-\rho)}\right]^{-1} \]
Вероятность ожидания: \[ P_{ожид} = \frac{(c\rho)^c}{c!(1-\rho)} \cdot P_0 \]
Среднее число заявок в очереди: \[ L_q = \frac{\rho \cdot P_{ожид}}{1 - \rho} \]
Практическая реализация
Подготовка окружения
Моделирование пуассоновского потока
def poisson_process(lambda_rate, simulation_time):
"""
Генерация пуассоновского потока заявок
Args:
lambda_rate: интенсивность потока (заявок/ед. времени)
simulation_time: время моделирования
Returns:
list: моменты поступления заявок
"""
arrivals = []
current_time = 0
while current_time < simulation_time:
# Генерация интервала между заявками
interval = np.random.exponential(1/lambda_rate)
current_time += interval
if current_time < simulation_time:
arrivals.append(current_time)
return arrivals
# Пример использования
lambda_rate = 2.0 # 2 заявки в единицу времени
simulation_time = 100
arrivals = poisson_process(lambda_rate, simulation_time)
print(f"Сгенерировано {len(arrivals)} заявок за {simulation_time} единиц времени")
print(f"Фактическая интенсивность: {len(arrivals)/simulation_time:.3f}")
# Визуализация потока заявок
plt.figure(figsize=(12, 4))
plt.eventplot(arrivals, lineoffsets=0.5, linelengths=0.8)
plt.title(f'Пуассоновский поток заявок (λ = {lambda_rate})')
plt.xlabel('Время')
plt.ylabel('Заявки')
plt.grid(True, alpha=0.3)
plt.show()Моделирование экспоненциального времени обслуживания
def exponential_service(mu_rate):
"""
Генерация времени обслуживания с экспоненциальным распределением
Args:
mu_rate: интенсивность обслуживания
Returns:
float: время обслуживания
"""
return np.random.exponential(1/mu_rate)
# Проверка распределения
mu_rate = 3.0 # 3 заявки в единицу времени
service_times = [exponential_service(mu_rate) for _ in range(1000)]
plt.figure(figsize=(10, 4))
plt.hist(service_times, bins=30, density=True, alpha=0.7, label='Эмпирическое')
x = np.linspace(0, max(service_times), 100)
plt.plot(x, stats.expon.pdf(x, scale=1/mu_rate), 'r-', linewidth=2, label='Теоретическое')
plt.title(f'Распределение времени обслуживания (μ = {mu_rate})')
plt.xlabel('Время обслуживания')
plt.ylabel('Плотность вероятности')
plt.legend()
plt.grid(True, alpha=0.3)
plt.show()Имитационное моделирование СМО M/M/1
class MM1Queue:
"""
Класс для моделирования системы M/M/1
"""
def __init__(self, lambda_rate, mu_rate):
self.lambda_rate = lambda_rate # интенсивность поступления
self.mu_rate = mu_rate # интенсивность обслуживания
self.rho = lambda_rate / mu_rate # коэффициент загрузки
# Статистика
self.arrivals = []
self.departures = []
self.queue_lengths = []
self.waiting_times = []
self.service_times = []
# Состояние системы
self.queue = deque()
self.server_busy = False
self.current_arrival_index = 0
self.current_service_end_time = 0
def simulate(self, simulation_time):
"""Запуск имитационного моделирования"""
# Генерация потока заявок
self.arrivals = poisson_process(self.lambda_rate, simulation_time)
# Инициализация
time_points = np.linspace(0, simulation_time, 1000)
current_time = 0
next_arrival_idx = 0
while current_time < simulation_time:
# Обработка поступления заявки
if next_arrival_idx < len(self.arrivals) and self.arrivals[next_arrival_idx] <= current_time:
self._handle_arrival(current_time)
next_arrival_idx += 1
# Обработка завершения обслуживания
if self.server_busy and current_time >= self.current_service_end_time:
self._handle_departure(current_time)
# Запись длины очереди
self.queue_lengths.append(len(self.queue))
# Шаг по времени
current_time += 0.01
# Расчет статистики
self._calculate_statistics()
def _handle_arrival(self, current_time):
"""Обработка поступления заявки"""
if not self.server_busy:
# Сервер свободен - immediate service
service_time = exponential_service(self.mu_rate)
self.service_times.append(service_time)
self.current_service_end_time = current_time + service_time
self.server_busy = True
self.waiting_times.append(0)
else:
# Сервер занят - добавление в очередь
self.queue.append(current_time)
def _handle_departure(self, current_time):
"""Обработка завершения обслуживания"""
self.departures.append(current_time)
if self.queue:
# Есть заявки в очереди
arrival_time = self.queue.popleft()
waiting_time = current_time - arrival_time
self.waiting_times.append(waiting_time)
# Начало обслуживания следующей заявки
service_time = exponential_service(self.mu_rate)
self.service_times.append(service_time)
self.current_service_end_time = current_time + service_time
self.server_busy = True
else:
# Очередь пуста
self.server_busy = False
def _calculate_statistics(self):
"""Расчет статистических характеристик"""
if self.waiting_times:
self.avg_waiting_time = np.mean(self.waiting_times)
self.avg_service_time = np.mean(self.service_times)
self.avg_queue_length = np.mean(self.queue_lengths)
else:
self.avg_waiting_time = 0
self.avg_service_time = 0
self.avg_queue_length = 0
def get_theoretical_metrics(self):
"""Расчет теоретических характеристик"""
if self.rho >= 1:
return None
return {
'P0': 1 - self.rho,
'L': self.rho / (1 - self.rho),
'Lq': self.rho**2 / (1 - self.rho),
'W': 1 / (self.mu_rate - self.lambda_rate),
'Wq': self.lambda_rate / (self.mu_rate * (self.mu_rate - self.lambda_rate))
}
def compare_theory_simulation(self):
"""Сравнение теоретических и имитационных результатов"""
theoretical = self.get_theoretical_metrics()
if theoretical is None:
print("Система нестабильна (ρ ≥ 1)")
return
print("=== Сравнение теоретических и имитационных результатов ===")
print(f"Коэффициент загрузки (ρ): {self.rho:.3f}")
print()
print("Характеристика\t\tТеория\t\tСимуляция\tРазница")
print("-" * 60)
print(f"Среднее время ожидания\t{theoretical['Wq']:.3f}\t\t{self.avg_waiting_time:.3f}\t\t{abs(theoretical['Wq'] - self.avg_waiting_time):.3f}")
print(f"Среднее время обслуживания\t{1/self.mu_rate:.3f}\t\t{self.avg_service_time:.3f}\t\t{abs(1/self.mu_rate - self.avg_service_time):.3f}")
print(f"Средняя длина очереди\t{theoretical['Lq']:.3f}\t\t{self.avg_queue_length:.3f}\t\t{abs(theoretical['Lq'] - self.avg_queue_length):.3f}")
# Пример моделирования
lambda_rate = 1.5 # 1.5 заявки в единицу времени
mu_rate = 2.0 # 2 заявки в единицу времени
queue_system = MM1Queue(lambda_rate, mu_rate)
queue_system.simulate(1000)
queue_system.compare_theory_simulation()
# Визуализация динамики очереди
plt.figure(figsize=(12, 6))
plt.plot(queue_system.queue_lengths)
plt.title(f'Динамика длины очереди (λ = {lambda_rate}, μ = {mu_rate})')
plt.xlabel('Время (шаги моделирования)')
plt.ylabel('Длина очереди')
plt.grid(True, alpha=0.3)
plt.show()Исследование влияния загрузки системы
def analyze_system_loading():
"""Исследование влияния коэффициента загрузки на характеристики СМО"""
mu_rate = 2.0 # фиксированная интенсивность обслуживания
lambda_rates = np.linspace(0.2, 1.8, 9) # различные интенсивности поступления
results = {
'lambda': [],
'rho': [],
'avg_waiting_time_sim': [],
'avg_waiting_time_theory': [],
'avg_queue_length_sim': [],
'avg_queue_length_theory': []
}
for lambda_rate in lambda_rates:
queue_system = MM1Queue(lambda_rate, mu_rate)
queue_system.simulate(1000)
theoretical = queue_system.get_theoretical_metrics()
results['lambda'].append(lambda_rate)
results['rho'].append(queue_system.rho)
results['avg_waiting_time_sim'].append(queue_system.avg_waiting_time)
results['avg_queue_length_sim'].append(queue_system.avg_queue_length)
if theoretical:
results['avg_waiting_time_theory'].append(theoretical['Wq'])
results['avg_queue_length_theory'].append(theoretical['Lq'])
else:
results['avg_waiting_time_theory'].append(np.nan)
results['avg_queue_length_theory'].append(np.nan)
# Визуализация результатов
fig, axes = plt.subplots(2, 2, figsize=(15, 10))
# Время ожидания
axes[0, 0].plot(results['rho'], results['avg_waiting_time_sim'], 'bo-', label='Симуляция')
axes[0, 0].plot(results['rho'], results['avg_waiting_time_theory'], 'r--', label='Теория')
axes[0, 0].set_xlabel('Коэффициент загрузки (ρ)')
axes[0, 0].set_ylabel('Среднее время ожидания')
axes[0, 0].set_title('Зависимость времени ожидания от загрузки')
axes[0, 0].legend()
axes[0, 0].grid(True, alpha=0.3)
# Длина очереди
axes[0, 1].plot(results['rho'], results['avg_queue_length_sim'], 'bo-', label='Симуляция')
axes[0, 1].plot(results['rho'], results['avg_queue_length_theory'], 'r--', label='Теория')
axes[0, 1].set_xlabel('Коэффициент загрузки (ρ)')
axes[0, 1].set_ylabel('Средняя длина очереди')
axes[0, 1].set_title('Зависимость длины очереди от загрузки')
axes[0, 1].legend()
axes[0, 1].grid(True, alpha=0.3)
# Гистограмма времени ожидания для разных ρ
rho_values = [0.3, 0.6, 0.9]
for i, rho in enumerate(rho_values):
lambda_rate = rho * mu_rate
queue_system = MM1Queue(lambda_rate, mu_rate)
queue_system.simulate(1000)
axes[1, i].hist(queue_system.waiting_times, bins=30, alpha=0.7)
axes[1, i].set_title(f'Распределение времени ожидания (ρ = {rho})')
axes[1, i].set_xlabel('Время ожидания')
axes[1, i].set_ylabel('Частота')
axes[1, i].grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
return results
# Запуск анализа
results = analyze_system_loading()Моделирование многоканальной системы M/M/c
class MMCQueue:
"""
Класс для моделирования системы M/M/c
"""
def __init__(self, lambda_rate, mu_rate, num_servers):
self.lambda_rate = lambda_rate
self.mu_rate = mu_rate
self.num_servers = num_servers
self.rho = lambda_rate / (num_servers * mu_rate) # коэффициент загрузки
# Статистика
self.arrivals = []
self.departures = []
self.queue_lengths = []
self.waiting_times = []
self.service_times = []
# Состояние системы
self.queue = deque()
self.servers = [None] * num_servers # None - свободен, иначе время окончания обслуживания
def simulate(self, simulation_time):
"""Запуск имитационного моделирования"""
# Генерация потока заявок
self.arrivals = poisson_process(self.lambda_rate, simulation_time)
# Инициализация
current_time = 0
next_arrival_idx = 0
while current_time < simulation_time:
# Обработка поступления заявки
if next_arrival_idx < len(self.arrivals) and self.arrivals[next_arrival_idx] <= current_time:
self._handle_arrival(current_time)
next_arrival_idx += 1
# Обработка завершения обслуживания
self._handle_departures(current_time)
# Запись длины очереди
self.queue_lengths.append(len(self.queue))
# Шаг по времени
current_time += 0.01
# Расчет статистики
self._calculate_statistics()
def _handle_arrival(self, current_time):
"""Обработка поступления заявки"""
# Поиск свободного сервера
free_server = None
for i, server_end_time in enumerate(self.servers):
if server_end_time is None or server_end_time <= current_time:
free_server = i
break
if free_server is not None:
# Есть свободный сервер - immediate service
service_time = exponential_service(self.mu_rate)
self.service_times.append(service_time)
self.servers[free_server] = current_time + service_time
self.waiting_times.append(0)
else:
# Все серверы заняты - добавление в очередь
self.queue.append(current_time)
def _handle_departures(self, current_time):
"""Обработка завершений обслуживания"""
for i in range(self.num_servers):
if self.servers[i] is not None and self.servers[i] <= current_time:
self.departures.append(current_time)
self.servers[i] = None
# Если есть заявки в очереди
if self.queue:
arrival_time = self.queue.popleft()
waiting_time = current_time - arrival_time
self.waiting_times.append(waiting_time)
# Начало обслуживания
service_time = exponential_service(self.mu_rate)
self.service_times.append(service_time)
self.servers[i] = current_time + service_time
def _calculate_statistics(self):
"""Расчет статистических характеристик"""
if self.waiting_times:
self.avg_waiting_time = np.mean(self.waiting_times)
self.avg_service_time = np.mean(self.service_times)
self.avg_queue_length = np.mean(self.queue_lengths)
else:
self.avg_waiting_time = 0
self.avg_service_time = 0
self.avg_queue_length = 0
def get_theoretical_metrics(self):
"""Расчет теоретических характеристик"""
if self.rho >= 1:
return None
# Расчет P0
c = self.num_servers
rho_c = self.lambda_rate / self.mu_rate # трафик
sum_term = sum((rho_c**k) / np.math.factorial(k) for k in range(c))
last_term = (rho_c**c) / (np.math.factorial(c) * (1 - self.rho))
P0 = 1 / (sum_term + last_term)
# Расчет вероятности ожидания
P_wait = (rho_c**c) / (np.math.factorial(c) * (1 - self.rho)) * P0
# Расчет характеристик
Lq = (self.rho * P_wait) / (1 - self.rho)
Wq = Lq / self.lambda_rate
W = Wq + 1/self.mu_rate
L = self.lambda_rate * W
return {
'P0': P0,
'P_wait': P_wait,
'Lq': Lq,
'Wq': Wq,
'W': W,
'L': L
}
# Сравнение одноканальной и многоканальной систем
def compare_single_vs_multi_server():
"""Сравнение систем с разным количеством серверов"""
lambda_rate = 2.0
mu_rate = 1.0
server_counts = [1, 2, 3, 4]
fig, axes = plt.subplots(2, 2, figsize=(15, 10))
metrics = {
'servers': [],
'avg_waiting_time_sim': [],
'avg_waiting_time_theory': [],
'avg_queue_length_sim': [],
'avg_queue_length_theory': []
}
for i, num_servers in enumerate(server_counts):
queue_system = MMCQueue(lambda_rate, mu_rate, num_servers)
queue_system.simulate(1000)
theoretical = queue_system.get_theoretical_metrics()
metrics['servers'].append(num_servers)
metrics['avg_waiting_time_sim'].append(queue_system.avg_waiting_time)
metrics['avg_queue_length_sim'].append(queue_system.avg_queue_length)
if theoretical:
metrics['avg_waiting_time_theory'].append(theoretical['Wq'])
metrics['avg_queue_length_theory'].append(theoretical['Lq'])
else:
metrics['avg_waiting_time_theory'].append(np.nan)
metrics['avg_queue_length_theory'].append(np.nan)
# Визуализация динамики очереди
row = i // 2
col = i % 2
axes[row, col].plot(queue_system.queue_lengths[:500]) # первые 500 точек
axes[row, col].set_title(f'Система M/M/{num_servers} (ρ = {queue_system.rho:.2f})')
axes[row, col].set_xlabel('Время')
axes[row, col].set_ylabel('Длина очереди')
axes[row, col].grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
# Сравнительный график
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 5))
x = np.arange(len(server_counts))
width = 0.35
ax1.bar(x - width/2, metrics['avg_waiting_time_sim'], width, label='Симуляция')
ax1.bar(x + width/2, metrics['avg_waiting_time_theory'], width, label='Теория')
ax1.set_xlabel('Количество серверов')
ax1.set_ylabel('Среднее время ожидания')
ax1.set_title('Сравнение времени ожидания')
ax1.set_xticks(x)
ax1.set_xticklabels(server_counts)
ax1.legend()
ax1.grid(True, alpha=0.3)
ax2.bar(x - width/2, metrics['avg_queue_length_sim'], width, label='Симуляция')
ax2.bar(x + width/2, metrics['avg_queue_length_theory'], width, label='Теория')
ax2.set_xlabel('Количество серверов')
ax2.set_ylabel('Средняя длина очереди')
ax2.set_title('Сравнение длины очереди')
ax2.set_xticks(x)
ax2.set_xticklabels(server_counts)
ax2.legend()
ax2.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
return metrics
# Запуск сравнения
metrics = compare_single_vs_multi_server()Самостоятельные задания
Задание 1: Анализ стабильности системы
Цель: Исследовать условия стабильности систем массового обслуживания.
Задачи:
- Для системы M/M/1 исследуйте поведение при различных значениях коэффициента загрузки ρ ∈ [0.1, 0.9]
- Постройте графики зависимости среднего времени ожидания и средней длины очереди от ρ
- Определите критическое значение ρ, при котором система становится нестабильной
- Сравните результаты имитационного моделирования с теоретическими расчетами
- Проанализируйте дисперсию времени ожидания при разных значениях ρ
Формат отчета:
- Графики зависимостей характеристик от ρ
- Таблица сравнения теоретических и практических результатов
- Анализ дисперсии и распределения времени ожидания
- Выводы об условиях стабильности системы
Задание 2: Оптимизация многоканальной системы
Цель: Определить оптимальное количество каналов обслуживания.
Задачи:
- Для заданной интенсивности поступления λ = 3 заявки/час и интенсивности обслуживания μ = 1 заявка/час
- Исследуйте системы M/M/c для c от 1 до 6
- Рассчитайте стоимость обслуживания, если:
- Стоимость одного канала: 100 ед./час
- Стоимость ожидания клиента: 50 ед./час
- Определите оптимальное количество каналов, минимизирующее общую стоимость
- Проанализируйте как изменится оптимальное решение при изменении стоимости ожидания
Формат отчета:
- Таблица с характеристиками систем для разного количества каналов
- Расчет общей стоимости для каждого варианта
- График зависимости общей стоимости от количества каналов
- Рекомендации по оптимальной конфигурации системы
Задание 3: Моделирование систем с ограниченной очередью
Цель: Исследовать системы с ограниченной вместимостью очереди.
Задачи:
- Модифицируйте класс MM1Queue для моделирования системы M/M/1/K (K - максимальная длина очереди)
- Исследуйте влияние ограничения на вероятность отказа в обслуживании
- Для λ = 2, μ = 3 и значений K от 0 до 10 рассчитайте:
- Вероятность отказа
- Эффективную интенсивность поступления
- Среднее время ожидания
- Сравните результаты с системой неограниченной очередью
- Постройте графики зависимостей характеристик от K
Формат отчета:
- Описание модифицированной модели
- Таблица с характеристиками для разных значений K
- Графики зависимостей от K
- Сравнение с неограниченной системой
- Выводы о влиянии ограничения очереди
Задание 4: Нестандартные распределения
Цель: Исследовать влияние распределений времени обслуживания на характеристики системы.
Задачи:
- Реализуйте систему M/G/1 с различными распределениями времени обслуживания:
- Экспоненциальное (baseline)
- Равномерное на [0, 2/μ]
- Нормальное со средним 1/μ и σ = 0.2/μ
- Гамма-распределение с параметрами (k=2, θ=1/(2μ))
- Сравните характеристики систем при одинаковых λ и μ
- Проанализируйте влияние коэффициента вариации времени обслуживания
- Исследуйте как изменяется дисперсия времени ожидания
- Сделайте выводы о чувствительности системы к форме распределения
Формат отчета:
- Описание реализованных распределений
- Сравнительная таблица характеристик для разных распределений
- Графики распределений времени ожидания
- Анализ влияния коэффициента вариации
- Выводы о робастности системы
Задание 5: Прикладная задача
Цель: Применить теорию массового обслуживания к реальной ситуации.
Выберите одну из задач:
Вариант А: Магазин - Описать работу кассового зала магазина - Собрать статистику (или использовать реалистичные параметры) - Определить оптимальное количество касс - Рассчитать экономическую эффективность
Вариант Б: Колл-центр - Моделировать работу операторов колл-центра - Учесть разные типы звонков с разным временем обслуживания - Определить необходимое количество операторов - Проанализировать уровень обслуживания
Вариант В: Веб-сервер - Моделировать обработку запросов веб-сервером - Учесть пиковые нагрузки - Определить необходимые ресурсы - Рассчитать время отклика
Формат отчета:
- Описание моделируемой системы
- Параметры и допущения
- Результаты моделирования
- Рекомендации по оптимизации
- Экономический анализ
Контрольные вопросы
- Какие основные компоненты входят в систему массового обслуживания?
- Что означает нотация Кендалла A/B/c и какие бывают основные типы распределений?
- Каковы условия стационарности для системы M/M/1 и почему они необходимы?
- В чем различие между теоретическим анализом и имитационным моделированием СМО?
- Как коэффициент загрузки ρ влияет на характеристики системы M/M/1?
- Какие преимущества дает многоканальная система по сравнению с одноканальной?
- Как ограничение длины очереди влияет на характеристики системы?
- Какие практические приложения имеют системы массового обслуживания?
- Какие методы используются для анализа нестационарных систем?
- Какие существуют подходы к оптимизации параметров СМО?
Рекомендации по выполнению работы
- Начните с простых примеров и убедитесь, что базовая модель работает корректно
- Используйте одинаковые random seeds для воспроизводимости результатов
- Проверяйте условия стационарности перед анализом характеристик
- Собирайте достаточную статистику для надежных оценок
- Визуализируйте результаты для лучшего понимания процессов
- Сравнивайте с теорией для валидации имитационных моделей
- Документируйте допущения и ограничения моделей