python 多并发

多并发实现

 1 #!/usr/bin/python
 2 # -*- coding: utf-8 -*-
 3 
 4 """
 5 @File    :   .py
 6 @Description: 
 7 @Time    :   2024/04/22 10:00:00
 8 @Author  :   
 9 @Version :   1.0
10 @Contact :   [email protected]
11 """
12 
13 import os
14 import re
15 import json
16 import concurrent.futures
17 
18 
19 class Test(object):
20     def __init__(self):
21         self.new_ret = []
22         self.token = "xxx"
23 
24     def load_json(self, file_path):
25         with open(file_path, 'r') as file:
26             case_list = json.loads(file.read())
27         return case_list
28 
29     def process_data(self, every_task):
30         print("[%s] 开始..." % str(every_task))
31         task_query = every_task.get("what_query", None)
32 
33         if not task_query:
34             return False
35         chat_res = Chat().chat(token=self.token, messages_content=task_query)
36         if chat_res.status_code != 200:
37             return False
38         chat_msg = chat_res.json()["a1"]
39         self.new_ret.append({"what": task_query, "response": chat_msg})
40         print("[%s] 结束..." % str(every_task))
41 
42     def get_chat_response(self, json_file):
43         json_case_list = self.load_json(file_path=json_file)
44 
45         with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
46             results = executor.map(self.process_data, json_case_list)  # 并发处理数据集
47 
48         print("完成.")
49 
50 
51 if __name__ == "__main__":
52     json_file = "data/mm.json"
53     TestPipeline().get_chat_response(json_file=json_file)