fork download 
  1. import time
  2. import concurrent.futures
  3. import threading
  4. import pandas as pd
  5. import queue
  6.  
  7.  
  8.  
  9.  
  10. class Strategy(object):
  11.     """策略类，处理任务结果"""
  12.  
  13.     def __init__(self, save_path: str):
  14.         self.save_path = save_path
  15.         self.result_queue = queue.Queue()
  16.         self.results: list = []
  17.  
  18.     def _setdata_hreo(self):
  19.         """设置数据"""
  20.         # 创建集合并转为列表
  21.         opt_set = {"弓", "勇武无双","攻击·连击"}
  22.         opts = list(opt_set)
  23.         print("策略选项：\n", opts) 
  24.  
  25.         # 修正字典中列表拼接的语法错误
  26.         dataset = {
  27.             "英雄": ["孙尚香"],
  28.             "兵种": [opts[0], opts[1]], # 用逗号分隔元素
  29.             "定位": [opts[2]],
  30.  
  31.  
  32.         }
  33.  
  34.         print("数据集：\n", dataset)
  35.  
  36.         self.results.append(dataset)
  37.         self.result_queue.put(self.results)
  38.         print("结果列表：\n", self.results)
  39.         self.get_results(dataset)
  40.  
  41.     def get_results(self,dataset):
  42.         if dataset:
  43.             df = pd.DataFrame(self.results)
  44.             print("DataFrame：{}".format(df))
  45.         else:
  46.             print("错误")
  47.  
  48.     def get_queue(self):
  49.         if self.result_queue:
  50.             data = self.result_queue.get()
  51.             print("队列：\n{}".format(data))
  52.  
  53.     def loop(self):
  54.         if self.results:
  55.             index :int = 0
  56.             length :int = len(self.results)
  57.             while index < length:
  58.                 result = self.results[index]
  59.                 print("循环♻️",result)
  60.                 index += 1
  61.  
  62.     def resource(self,*item):
  63.         if item:
  64.  
  65.             self.result_queue.put(item)
  66.             items = self.result_queue.get()
  67.             print("队列：\n",items)
  68.  
  69. def main():
  70.  
  71.     strategy = Strategy(save_path="./")
  72.     strategy._setdata_hreo()
  73.     strategy.get_queue()
  74.     strategy.loop()
  75.  
  76.     strategy.resource({
  77.              "等级":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16]
  78.     })
  79.  
  80.  
  81. # 实例化策略类并调用方法
  82. if __name__ == "__main__":
  83.  
  84.  
  85.     tasks :list = []
  86.     thr = threading.Thread(target=main)
  87.     tasks.append(thr)
  88.  
  89.     index = 0
  90.     length = len(tasks)
  91.  
  92.     start = time.time()
  93.     while index < length:
  94.         task = tasks[index]
  95.         task.start()
  96.         index += 1
  97.     print("线程结束")
  98.     end = time.time()
  99.     print(f"耗时：{end - start:.6f}")
  100.  
  101.     st = time.time()
  102.     while index < length:
  103.         task = tasks[index]
  104.         task.join()
  105.         index += 1
  106.  
  107.     print("任务全部完成")
  108.     ed = time.time()
  109.     print(f"耗时：{ed - st:.6f}")
Success #stdin #stdout 2.66s 79376KB
comments (?)
stdin 
Standard input is empty
stdout 
策略选项：
 ['勇武无双', '攻击·连击', '弓']
数据集：
 {'英雄': ['孙尚香'], '兵种': ['勇武无双', '攻击·连击'], '定位': ['弓']}
结果列表：
 [{'英雄': ['孙尚香'], '兵种': ['勇武无双', '攻击·连击'], '定位': ['弓']}]
线程结束
耗时：0.013062
任务全部完成
耗时：0.000001
DataFrame：      英雄             兵种   定位
0  [孙尚香]  [勇武无双, 攻击·连击]  [弓]
队列：
[{'英雄': ['孙尚香'], '兵种': ['勇武无双', '攻击·连击'], '定位': ['弓']}]
循环♻️ {'英雄': ['孙尚香'], '兵种': ['勇武无双', '攻击·连击'], '定位': ['弓']}
队列：
 ({'等级': [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16]},)
https://ideone.com/ey9Uq4
language:
Python 3 (python  3.12)
created:
6 days ago
visibility:
public

Share or Embed source code

Discover > Sphere Engine API

The brand new service which powers Ideone!

Discover > IDE Widget

Widget for compiling and running the source code in a web browser!