feat: 优化并发任务处理逻辑以提高搜索效率

main.py

@@ -358,27 +358,20 @@ def integrate_data_parallel():
         if not tasks:
             time.sleep(10)
             continue
-        for payload, kitem in tasks:
-            proxname = proxiesdict.get(kitem['rn'])
-            print(proxname)
-            h = gettoken(proxname)
-            print(h)
-            v_list = get_searchInfo(kitem['keyword'], kitem['level'], h, proxname)
 
-        # futures = [
-        #     executor.submit(search_worker, payload, kitem, flag)
-        #     for payload, kitem in tasks
-        # ]
-        #
-        # # 统计回滚
-        # rollback = {0: [], 1: [], 2: []}
-        #
-        # for fut in concurrent.futures.as_completed(futures):
-        #     ok, f_flag, payload, kitem, v_list = fut.result()
-        #
-        #     if not ok:
-        #         rollback[f_flag].append(payload)
-        #         continue
+        futures = [
+            executor.submit(search_worker, payload, kitem, flag)
+            for payload, kitem in tasks
+        ]
+        # 统计回滚
+        rollback = {0: [], 1: [], 2: []}
+
+        for fut in concurrent.futures.as_completed(futures):
+            ok, f_flag, payload, kitem, v_list = fut.result()
+
+            if not ok:
+                rollback[f_flag].append(payload)
+                continue
 
             for item in v_list:
                 if not v_list: