201 lines
6.5 KiB
Python
201 lines
6.5 KiB
Python
import asyncio
|
|
import random
|
|
import aiohttp
|
|
from time import time, sleep
|
|
import uuid
|
|
from student import generate_student_data
|
|
|
|
|
|
def generate_students():
|
|
student = generate_student_data()
|
|
return student
|
|
|
|
|
|
# 配置信息
|
|
BASE_URL = "http://apply.hxzhxy.cn"
|
|
STATIC_PATHS = [
|
|
"/static/js/chunk-vendors.6a5fca90.js",
|
|
"/static/js/index.5371b2cb.js",
|
|
"/static/js/pages-login-index.b4bd93ca.js",
|
|
"/static/js/pages-login-index.js",
|
|
"http://band.hxzhxy.cn/register/plan/getTeam?teamId=106",
|
|
]
|
|
|
|
# 用户数据生成
|
|
USER_AGENTS = [
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
|
|
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15",
|
|
"Mozilla/5.0 (iPhone; CPU iPhone OS 15_5 like Mac OS X) AppleWebKit/605.1.15",
|
|
"Mozilla/5.0 (Linux; Android 12; SM-G991B) AppleWebKit/537.36",
|
|
"Mozilla/5.0 (iPad; CPU OS 15_5 like Mac OS X) AppleWebKit/605.1.15",
|
|
]
|
|
|
|
|
|
def generate_user_data(user_id):
|
|
"""生成模拟用户数据"""
|
|
return {
|
|
"id": user_id,
|
|
"session_id": str(uuid.uuid4()),
|
|
"user_agent": random.choice(USER_AGENTS),
|
|
"request_count": 0,
|
|
"start_time": time(),
|
|
}
|
|
|
|
|
|
async def async_submit_to_baidu(user_data, session):
|
|
return
|
|
"""异步提交数据"""
|
|
# 构造提交数据
|
|
payload = generate_students()
|
|
a = random.uniform(5, 20)
|
|
await asyncio.sleep(a)
|
|
print(f"🎉 用户 {user_data['id']} 提交数据 ")
|
|
|
|
try:
|
|
# 提交到百度
|
|
start_time = time()
|
|
async with session.post(
|
|
"http://band.hxzhxy.cn/register/student/add",
|
|
json=payload,
|
|
headers={
|
|
"User-Agent": user_data["user_agent"],
|
|
"X-Session-Id": user_data["session_id"],
|
|
"Content-Type": "application/json",
|
|
},
|
|
) as response:
|
|
# 处理响应
|
|
response_text = await response.text()
|
|
elapsed_time = time() - start_time
|
|
|
|
# 记录用户总用时
|
|
total_time = time() - user_data["start_time"]
|
|
user_data["total_time"] = total_time
|
|
|
|
if response.status == 200:
|
|
print(
|
|
f"🎉 用户 {user_data['id']} 数据提交成功! | 提交耗时: {elapsed_time:.3f}秒 | 总用时: {total_time:.3f}秒"
|
|
)
|
|
return True
|
|
else:
|
|
print(
|
|
f"⚠️ 用户 {user_data['id']} 提交失败: 状态码 {response.status} | 总用时: {total_time:.3f}秒"
|
|
)
|
|
return False
|
|
except Exception as e:
|
|
print(f"‼️ 用户 {user_data['id']} 提交异常: {str(e)}")
|
|
return False
|
|
|
|
|
|
# 修改用户会话函数为异步版本
|
|
async def async_simulate_user_session(user_id):
|
|
"""异步模拟一个用户的完整会话"""
|
|
user_data = generate_user_data(user_id)
|
|
print(f"👤 用户 {user_id} 开始会话, Session ID: {user_data['session_id']}")
|
|
|
|
# 创建共享的ClientSession
|
|
async with aiohttp.ClientSession() as session:
|
|
# 使用异步请求所有静态资源
|
|
start = time()
|
|
tasks = []
|
|
for path in STATIC_PATHS:
|
|
url = BASE_URL + path if not path.startswith("http") else path
|
|
tasks.append(async_fetch_resource(url, user_data, session)) # 传入session
|
|
|
|
# 并行执行所有静态资源请求
|
|
results = await asyncio.gather(*tasks, return_exceptions=True)
|
|
for i, result in enumerate(results):
|
|
if isinstance(result, Exception):
|
|
print(f"❌ 用户 {user_id} 资源加载失败: {str(result)}")
|
|
else:
|
|
print(f"✅ 用户 {user_id} 资源加载完成: {result}")
|
|
|
|
end = time()
|
|
print(f"📊 用户 {user_id} 加载所有静态资源耗时: {end - start:.3f}秒")
|
|
|
|
# 表单填写和提交
|
|
print(f"🚀 用户 {user_id} 开始看表单")
|
|
await async_submit_to_baidu(user_data, session) # 传入session
|
|
|
|
return user_data
|
|
|
|
|
|
# 添加异步资源请求函数
|
|
async def async_fetch_resource(url, user_data, session):
|
|
"""异步获取单个资源"""
|
|
# 更新用户请求计数
|
|
user_data["request_count"] += 1
|
|
|
|
# 生成请求头
|
|
headers = {
|
|
"User-Agent": user_data["user_agent"],
|
|
"X-Session-Id": user_data["session_id"],
|
|
"X-Request-Count": str(user_data["request_count"]),
|
|
"Accept-Encoding": "gzip, deflate",
|
|
}
|
|
|
|
# 添加随机延迟模拟网络波动
|
|
await asyncio.sleep(random.uniform(0.1, 0.5))
|
|
|
|
try:
|
|
# 异步执行资源请求
|
|
start_time = time()
|
|
async with session.get(url, headers=headers) as response:
|
|
content = await response.text()
|
|
elapsed_time = time() - start_time
|
|
|
|
if response.status != 200:
|
|
return f"错误状态码: {response.status} | 耗时: {elapsed_time:.3f}秒"
|
|
|
|
# 提取文件名
|
|
file_name = url.split("/")[-1].split("?")[0] # 处理带参数的URL
|
|
|
|
return f"文件: {file_name} | 大小: {len(content)/1024:.1f}KB | 耗时: {elapsed_time:.3f}秒"
|
|
except Exception as e:
|
|
return f"请求异常: {str(e)}"
|
|
|
|
|
|
# 修改多用户模拟函数为异步版本
|
|
async def async_simulate_multiple_users(num_users=10):
|
|
"""异步模拟多个用户并发访问"""
|
|
print(f"🎬 开始模拟 {num_users} 个用户并发访问...")
|
|
print("=" * 70)
|
|
|
|
# 创建所有用户会话任务
|
|
tasks = [async_simulate_user_session(i + 1) for i in range(num_users)]
|
|
|
|
# 并行执行所有用户会话
|
|
results = await asyncio.gather(*tasks, return_exceptions=True)
|
|
|
|
# 收集成功的结果
|
|
completed_results = []
|
|
for result in results:
|
|
if isinstance(result, Exception):
|
|
print(f"用户会话异常: {str(result)}")
|
|
else:
|
|
completed_results.append(result)
|
|
|
|
# 生成性能报告
|
|
print("\n" + "=" * 70)
|
|
print("🏁 所有用户已完成操作! 性能摘要:")
|
|
print("-" * 70)
|
|
|
|
total_requests = 0
|
|
completed_submissions = 0
|
|
|
|
for user in completed_results:
|
|
total_requests += user["request_count"]
|
|
if "total_time" in user:
|
|
print(
|
|
f"用户 {user['id']:2d} | 用时: {user['total_time']:.3f}秒 | 请求数: {user['request_count']}"
|
|
)
|
|
completed_submissions += 1
|
|
|
|
print("\n汇总统计:")
|
|
print(f"- 总请求数: {total_requests}次")
|
|
print(f"- 用户提交率: {completed_submissions}/{len(completed_results)}")
|
|
print("=" * 70)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(async_simulate_multiple_users(num_users=3))
|