使用多线程调用某个方法（请求），计算每个线程消耗时间
import threading
import time
import requests
import json
import concurrent.futures
def input_req():
url = "https://xxxxxxxxxxxx"
approval_content = "nulla in"
payload = json.dumps({
"approval_content": approval_content,
"ac_token": "123"
})
headers = {
'Authorization': 'bear {{ACCESS_TOKEN}}',
'User-Agent': 'Apifox/1.0.0 (https://apifox.com)',
'Content-Type': 'application/json'
}
response = requests.request("POST", url, headers=headers, data=payload)
if response.status_code == 200:
content = json.loads(response.content)
task_id = content.get("item").get('task_id')
return task_id
def output_req(task_id):
url = "https://xxxxxx"
payload = json.dumps({
"task_id": str(task_id),
"ac_token": "123"
})
headers = {
'Authorization': 'bear {{ACCESS_TOKEN}}',
'User-Agent': 'Apifox/1.0.0 (https://apifox.com)',
'Content-Type': 'application/json'
}
response = requests.request("POST", url, headers=headers, data=payload)
if response.status_code == 200:
item = json.loads(response.content).get('item')
if item and isinstance(item, list):
return True
else:
print(f"发送输出请求完毕，返回状态码：{response.status_code}")
def run():
start_time = time.time()
thread_id = threading.current_thread().ident
print("线程", thread_id, "开始执行")
tag = False
task_id = input_req()
print("线程", thread_id, "任务id", task_id)
while not tag:
tag = output_req(task_id)
if not tag:
time.sleep(1)
end_time = time.time()
execution_time = end_time - start_time
print("线程", thread_id, "完成，执行时间：", execution_time, "秒")
return execution_time
def test_time(n=1):
for i in range(n):
execution_times = []
with concurrent.futures.ThreadPoolExecutor() as executor:
futures = [executor.submit(run) for i in range(1, 11)]
for future in concurrent.futures.as_completed(futures):
execution_time = future.result()
execution_times.append(execution_time)
print(f"---轮数：{i + 1}---所有线程已完成-----")
all_time = sum(execution_times)
print(f"---轮数：{i + 1}---总耗时{all_time}-----")
for i, execution_time in enumerate(execution_times):
print("线程", i + 1, "消耗时间：", execution_time, "秒")
if __name__ == '__main__':
test_time(2)