262 lines
9.0 KiB
Python
262 lines
9.0 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
获取密集库位处理器并发问题修复演示
|
||
展示优化前后的区别
|
||
"""
|
||
|
||
import asyncio
|
||
import time
|
||
import random
|
||
from typing import Dict, Any, List
|
||
from datetime import datetime
|
||
|
||
|
||
class MockDatabase:
|
||
"""模拟数据库"""
|
||
def __init__(self):
|
||
# 模拟5个空闲库位
|
||
self.storage_sites = {
|
||
f"A00{i}": {
|
||
"is_locked": False,
|
||
"is_occupied": False,
|
||
"last_access": datetime.now()
|
||
}
|
||
for i in range(1, 6)
|
||
}
|
||
self.operation_log = []
|
||
|
||
async def query_available_sites(self) -> List[str]:
|
||
"""查询可用库位"""
|
||
await asyncio.sleep(0.01) # 模拟数据库查询延迟
|
||
available = [
|
||
site_id for site_id, info in self.storage_sites.items()
|
||
if not info["is_locked"] and not info["is_occupied"]
|
||
]
|
||
return sorted(available)
|
||
|
||
async def lock_site_old_way(self, site_id: str, task_id: str) -> bool:
|
||
"""旧方式：分别查询后锁定（有竞争条件）"""
|
||
# 先查询
|
||
available_sites = await self.query_available_sites()
|
||
if site_id not in available_sites:
|
||
return False
|
||
|
||
# 模拟网络延迟和处理时间
|
||
await asyncio.sleep(random.uniform(0.01, 0.05))
|
||
|
||
# 尝试锁定（可能已被其他请求抢占）
|
||
if self.storage_sites[site_id]["is_locked"]:
|
||
return False
|
||
|
||
self.storage_sites[site_id]["is_locked"] = True
|
||
self.operation_log.append(f"[OLD] Task {task_id} locked {site_id}")
|
||
return True
|
||
|
||
async def lock_site_new_way(self, site_id: str, task_id: str) -> bool:
|
||
"""新方式：原子性锁定（无竞争条件）"""
|
||
# 原子操作：检查并锁定
|
||
if not self.storage_sites[site_id]["is_locked"] and not self.storage_sites[site_id]["is_occupied"]:
|
||
self.storage_sites[site_id]["is_locked"] = True
|
||
self.operation_log.append(f"[NEW] Task {task_id} locked {site_id}")
|
||
return True
|
||
return False
|
||
|
||
def reset(self):
|
||
"""重置数据库状态"""
|
||
for site_info in self.storage_sites.values():
|
||
site_info["is_locked"] = False
|
||
self.operation_log.clear()
|
||
|
||
|
||
async def simulate_old_approach(db: MockDatabase, num_tasks: int) -> Dict[str, Any]:
|
||
"""模拟旧的实现方式"""
|
||
print(f"\n=== 模拟旧方式：查询后锁定（存在竞争条件）===")
|
||
|
||
async def old_task(task_id: int):
|
||
# 查询可用库位
|
||
available_sites = await db.query_available_sites()
|
||
if not available_sites:
|
||
return {"task_id": task_id, "success": False, "reason": "no_sites"}
|
||
|
||
# 选择第一个可用库位
|
||
target_site = available_sites[0]
|
||
|
||
# 尝试锁定
|
||
success = await db.lock_site_old_way(target_site, f"task_{task_id}")
|
||
return {
|
||
"task_id": task_id,
|
||
"success": success,
|
||
"site_id": target_site if success else None,
|
||
"reason": "success" if success else "race_condition"
|
||
}
|
||
|
||
# 重置数据库
|
||
db.reset()
|
||
start_time = time.time()
|
||
|
||
# 并发执行任务
|
||
tasks = [old_task(i) for i in range(num_tasks)]
|
||
results = await asyncio.gather(*tasks)
|
||
|
||
end_time = time.time()
|
||
|
||
# 分析结果
|
||
successful = [r for r in results if r["success"]]
|
||
failed = [r for r in results if not r["success"]]
|
||
|
||
# 检查是否有重复分配
|
||
allocated_sites = [r["site_id"] for r in successful if r["site_id"]]
|
||
duplicates = len(allocated_sites) - len(set(allocated_sites))
|
||
|
||
print(f"执行时间: {end_time - start_time:.3f}s")
|
||
print(f"成功任务: {len(successful)}")
|
||
print(f"失败任务: {len(failed)}")
|
||
print(f"重复分配: {duplicates} 个库位被重复分配")
|
||
|
||
if duplicates > 0:
|
||
print("❌ 发现并发竞争问题：有库位被重复分配！")
|
||
else:
|
||
print("✓ 未发现重复分配")
|
||
|
||
print("操作日志:")
|
||
for log in db.operation_log[:10]: # 只显示前10条
|
||
print(f" {log}")
|
||
|
||
return {
|
||
"total_tasks": num_tasks,
|
||
"successful": len(successful),
|
||
"failed": len(failed),
|
||
"duplicates": duplicates,
|
||
"execution_time": end_time - start_time
|
||
}
|
||
|
||
|
||
async def simulate_new_approach(db: MockDatabase, num_tasks: int) -> Dict[str, Any]:
|
||
"""模拟新的实现方式"""
|
||
print(f"\n=== 模拟新方式：原子性获取（无竞争条件）===")
|
||
|
||
async def new_task(task_id: int):
|
||
# 查询可用库位
|
||
available_sites = await db.query_available_sites()
|
||
if not available_sites:
|
||
return {"task_id": task_id, "success": False, "reason": "no_sites"}
|
||
|
||
# 按优先级尝试原子性锁定每个库位
|
||
for site_id in available_sites:
|
||
success = await db.lock_site_new_way(site_id, f"task_{task_id}")
|
||
if success:
|
||
return {
|
||
"task_id": task_id,
|
||
"success": True,
|
||
"site_id": site_id,
|
||
"reason": "success"
|
||
}
|
||
|
||
return {"task_id": task_id, "success": False, "reason": "all_occupied"}
|
||
|
||
# 重置数据库
|
||
db.reset()
|
||
start_time = time.time()
|
||
|
||
# 并发执行任务
|
||
tasks = [new_task(i) for i in range(num_tasks)]
|
||
results = await asyncio.gather(*tasks)
|
||
|
||
end_time = time.time()
|
||
|
||
# 分析结果
|
||
successful = [r for r in results if r["success"]]
|
||
failed = [r for r in results if not r["success"]]
|
||
|
||
# 检查是否有重复分配
|
||
allocated_sites = [r["site_id"] for r in successful if r["site_id"]]
|
||
duplicates = len(allocated_sites) - len(set(allocated_sites))
|
||
|
||
print(f"执行时间: {end_time - start_time:.3f}s")
|
||
print(f"成功任务: {len(successful)}")
|
||
print(f"失败任务: {len(failed)}")
|
||
print(f"重复分配: {duplicates} 个库位被重复分配")
|
||
|
||
if duplicates > 0:
|
||
print("❌ 仍然存在并发竞争问题！")
|
||
else:
|
||
print("✓ 成功解决并发竞争问题")
|
||
|
||
print("操作日志:")
|
||
for log in db.operation_log[:10]: # 只显示前10条
|
||
print(f" {log}")
|
||
|
||
return {
|
||
"total_tasks": num_tasks,
|
||
"successful": len(successful),
|
||
"failed": len(failed),
|
||
"duplicates": duplicates,
|
||
"execution_time": end_time - start_time
|
||
}
|
||
|
||
|
||
async def run_comparison_demo():
|
||
"""运行对比演示"""
|
||
print("🔧 获取密集库位处理器并发问题修复演示")
|
||
print("=" * 60)
|
||
|
||
db = MockDatabase()
|
||
num_concurrent_tasks = 10 # 10个并发任务竞争5个库位
|
||
|
||
print(f"测试场景: {num_concurrent_tasks} 个并发任务竞争 {len(db.storage_sites)} 个库位")
|
||
|
||
# 测试旧方式
|
||
old_results = await simulate_old_approach(db, num_concurrent_tasks)
|
||
|
||
# 测试新方式
|
||
new_results = await simulate_new_approach(db, num_concurrent_tasks)
|
||
|
||
# 对比结果
|
||
print(f"\n=== 对比结果 ===")
|
||
print(f"旧方式 - 成功: {old_results['successful']}, 失败: {old_results['failed']}, 重复分配: {old_results['duplicates']}")
|
||
print(f"新方式 - 成功: {new_results['successful']}, 失败: {new_results['failed']}, 重复分配: {new_results['duplicates']}")
|
||
|
||
if old_results['duplicates'] > 0 and new_results['duplicates'] == 0:
|
||
print("🎉 优化成功！新方式完全解决了并发竞争问题")
|
||
elif old_results['duplicates'] == new_results['duplicates'] == 0:
|
||
print("ℹ️ 在当前测试条件下，两种方式都没有出现竞争问题")
|
||
print(" 可能需要增加并发压力或调整延迟来复现问题")
|
||
else:
|
||
print("⚠️ 优化可能不够完善，需要进一步检查")
|
||
|
||
# 性能对比
|
||
performance_diff = ((new_results['execution_time'] - old_results['execution_time']) / old_results['execution_time']) * 100
|
||
if performance_diff > 0:
|
||
print(f"📊 性能: 新方式比旧方式慢 {performance_diff:.1f}%")
|
||
else:
|
||
print(f"📊 性能: 新方式比旧方式快 {abs(performance_diff):.1f}%")
|
||
|
||
|
||
async def stress_test():
|
||
"""压力测试"""
|
||
print(f"\n=== 压力测试 ===")
|
||
print("高并发场景下测试并发安全性...")
|
||
|
||
db = MockDatabase()
|
||
num_tasks = 50 # 50个并发任务
|
||
|
||
print(f"压力测试: {num_tasks} 个高并发任务")
|
||
|
||
# 只测试新方式的压力承受能力
|
||
results = await simulate_new_approach(db, num_tasks)
|
||
|
||
success_rate = (results['successful'] / results['total_tasks']) * 100
|
||
print(f"成功率: {success_rate:.1f}%")
|
||
print(f"平均每任务耗时: {results['execution_time']/results['total_tasks']*1000:.2f}ms")
|
||
|
||
if results['duplicates'] == 0:
|
||
print("✅ 压力测试通过：无并发竞争问题")
|
||
else:
|
||
print("❌ 压力测试失败：仍存在并发竞争问题")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(run_comparison_demo())
|
||
asyncio.run(stress_test()) |