pip install pyqt5 aiohttp

import sys
import asyncio
import aiohttp
from PyQt5.QtWidgets import (
    QApplication, QWidget, QLabel, QLineEdit, QTextEdit, QVBoxLayout, QHBoxLayout, QPushButton, QSpinBox, QTableWidget, QTableWidgetItem
)
from PyQt5.QtCore import Qt
from PyQt5.QtGui import QFont

class AsyncHttpTester(QWidget):
    def __init__(self):
        super().__init__()
        self.init_ui()

    def init_ui(self):
        """初始化用戶界面"""
        self.setWindowTitle("接口并發(fā)測試工具")
        self.setGeometry(100, 100, 800, 600)
        self.setFont(QFont("Arial", 10))

        # === 接口配置區(qū) ===
        url_label = QLabel("請(qǐng)求 URL:")
        self.url_input = QLineEdit()
        self.url_input.setPlaceholderText("請(qǐng)輸入接口 URL")

        headers_label = QLabel("請(qǐng)求頭 (JSON 格式):")
        self.headers_input = QTextEdit()
        self.headers_input.setPlaceholderText('例如：{"Content-Type": "application/json"}')

        params_label = QLabel("請(qǐng)求參數(shù) (JSON 格式):")
        self.params_input = QTextEdit()
        self.params_input.setPlaceholderText('例如：{"key": "value"}')

        times_label = QLabel("發(fā)送次數(shù):")
        self.times_input = QSpinBox()
        self.times_input.setRange(1, 1000)
        self.times_input.setValue(1)

        # === 開始按鈕 ===
        self.start_button = QPushButton("開始測試")
        self.start_button.clicked.connect(self.start_test)

        # === 結(jié)果展示區(qū) ===
        results_label = QLabel("測試結(jié)果:")
        self.results_table = QTableWidget()
        self.results_table.setColumnCount(3)
        self.results_table.setHorizontalHeaderLabels(["請(qǐng)求序號(hào)", "狀態(tài)碼", "響應(yīng)時(shí)間 (秒)"])
        self.results_table.setColumnWidth(0, 100)
        self.results_table.setColumnWidth(1, 100)
        self.results_table.setColumnWidth(2, 150)

        # === 布局 ===
        layout = QVBoxLayout()

        # 接口配置布局
        config_layout = QVBoxLayout()
        config_layout.addWidget(url_label)
        config_layout.addWidget(self.url_input)
        config_layout.addWidget(headers_label)
        config_layout.addWidget(self.headers_input)
        config_layout.addWidget(params_label)
        config_layout.addWidget(self.params_input)
        config_layout.addWidget(times_label)
        config_layout.addWidget(self.times_input)

        # 添加開始按鈕
        config_layout.addWidget(self.start_button)

        # 結(jié)果展示布局
        results_layout = QVBoxLayout()
        results_layout.addWidget(results_label)
        results_layout.addWidget(self.results_table)

        # 整合布局
        layout.addLayout(config_layout)
        layout.addLayout(results_layout)
        self.setLayout(layout)

    async def send_request(self, session, url, headers, params, index):
        """發(fā)送單個(gè) HTTP 請(qǐng)求"""
        try:
            async with session.post(url, json=params, headers=headers) as response:
                elapsed = response.elapsed.total_seconds() if response.elapsed else 0
                return index, response.status, elapsed
        except Exception as e:
            return index, f"Error: {str(e)}", 0

    async def start_async_requests(self, url, headers, params, times):
        """啟動(dòng)并發(fā)請(qǐng)求"""
        tasks = []
        async with aiohttp.ClientSession() as session:
            for i in range(times):
                tasks.append(self.send_request(session, url, headers, params, i + 1))
            return await asyncio.gather(*tasks)

    def start_test(self):
        """開始測試按鈕事件"""
        url = self.url_input.text().strip()
        try:
            headers = eval(self.headers_input.toPlainText().strip()) if self.headers_input.toPlainText().strip() else {}
            params = eval(self.params_input.toPlainText().strip()) if self.params_input.toPlainText().strip() else {}
        except Exception as e:
            self.results_table.setRowCount(0)
            self.results_table.setRowCount(1)
            self.results_table.setItem(0, 0, QTableWidgetItem("Error"))
            self.results_table.setItem(0, 1, QTableWidgetItem(f"Invalid headers/params: {str(e)}"))
            return

        times = self.times_input.value()
        if not url:
            self.results_table.setRowCount(0)
            self.results_table.setRowCount(1)
            self.results_table.setItem(0, 0, QTableWidgetItem("Error"))
            self.results_table.setItem(0, 1, QTableWidgetItem("URL 不能為空"))
            return

        # 清空結(jié)果表
        self.results_table.setRowCount(0)

        # 啟動(dòng)異步任務(wù)
        loop = asyncio.get_event_loop()
        results = loop.run_until_complete(self.start_async_requests(url, headers, params, times))

        # 更新結(jié)果表
        self.results_table.setRowCount(len(results))
        for i, (index, status, elapsed) in enumerate(results):
            self.results_table.setItem(i, 0, QTableWidgetItem(str(index)))
            self.results_table.setItem(i, 1, QTableWidgetItem(str(status)))
            self.results_table.setItem(i, 2, QTableWidgetItem(f"{elapsed:.2f}"))

if __name__ == "__main__":
    app = QApplication(sys.argv)
    tester = AsyncHttpTester()
    tester.show()
    sys.exit(app.exec_())