Files
gateway_plc/gateway/test_read.py

58 lines
2.1 KiB
Python
Raw Permalink Normal View History

#!/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)