58 lines
2.1 KiB
Python
58 lines
2.1 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
'''
|
||
# @Time : 2025/10/10 10:00
|
||
# @Author : reenrr
|
||
# @File : test_read.py
|
||
# @Description : 使用http测试读取的速率
|
||
'''
|
||
import requests
|
||
import time
|
||
import logging
|
||
|
||
# 配置日志（可选，用于记录请求状态）
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s-%(levelname)s-%(message)s"
|
||
)
|
||
|
||
def send_continuous_requests(url, interval=0.5, max_requests=None):
|
||
"""
|
||
持续发送 HTTP GET 请求到指定 URL
|
||
:param url: 请求的目标 URL
|
||
:param interval: 请求间隔（秒），默认 0.5 秒
|
||
:param max_requests: 最大请求次数（None 表示无限循环）
|
||
"""
|
||
request_count = 0
|
||
error_count = 0
|
||
|
||
while max_requests is None or request_count < max_requests:
|
||
try:
|
||
start_time = time.time()
|
||
# 发送 GET 请求
|
||
response = requests.get(url, timeout=5) # 超时时间设为 5 秒，避免请求卡住
|
||
request_count += 1
|
||
|
||
# 检查响应状态码（200 表示成功）
|
||
if response.status_code == 200:
|
||
logging.info(f"请求 {request_count} 成功 | 响应耗时: {time.time()-start_time:.3f} 秒")
|
||
# 可选：打印响应内容（如果需要验证返回数据）
|
||
# print(f"响应内容: {response.text}")
|
||
else:
|
||
logging.warning(f"请求 {request_count} 失败 | 状态码: {response.status_code}")
|
||
error_count += 1
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
logging.error(f"请求 {request_count} 异常: {str(e)}")
|
||
error_count += 1
|
||
|
||
# 控制请求间隔
|
||
time.sleep(interval)
|
||
|
||
logging.info(f"总计请求: {request_count} | 失败/异常: {error_count}")
|
||
|
||
if __name__ == "__main__":
|
||
# 替换为你要测试的 URL
|
||
target_url = "http://127.0.0.1:5000/api/read_generic/PLC1/DB100_Read/1048/bool?count=10"
|
||
send_continuous_requests(target_url, interval=0.05, max_requests=100) # 示例：每秒发 1 次，共发 100 次
|
||
# 若要无限循环，可改为：send_continuous_requests(target_url, interval=1) |