优化多线程代码以提高检测速度：使用多进程、会话维持和队列

{ "title": "有哪些方式可以优化下面的代码，提高多线程检测速度，功能不变", "description": "本文介绍了如何优化多线程代码，提高多线程检测速度，主要方法包括使用多进程替代多线程、使用 requests.Session 进行会话维持、使用 ThreadPoolExecutor.map 进行任务分发、使用 Queue 进行结果收集、使用 concurrent.futures.as_completed 获取任务完成顺序。", "keywords": "多线程, 多进程, Python, 代码优化, 检测速度, Google搜索, 会话维持, 队列, 任务分发, 异步", "content": "import requests\nimport time\nimport os\nimport urllib3\nimport sys\nimport random\nfrom bs4 import BeautifulSoup\nfrom concurrent.futures import ProcessPoolExecutor, as_completed\nfrom threading import Lock\nfrom colorama import Fore, init\nfrom queue import Queue\n\nnow_time = time.strftime('%Y-%m-%d %H-%M')\n\n\n# 读取Dorks\ndef work(dorks):\n with open(dorks, mode='r', encoding='utf-8') as file:\n read_content = file.readlines()\n # 将内容加入列表\n content = [result.strip() for result in read_content]\n # 返回数量丢给任务池\n return len(read_content), content\n\n\n# Google搜索\ndef google_serach(query, locks, filename, result_queue):\n try:\n # 关闭HTTPS报错信息\n urllib3.disable_warnings()\n filename = os.path.join(os.getcwd(), f'{filename}.txt')\n domains = ['fr','it','ca','co.uk','ru','co,jp','co.kr','com.au','co.in','com.br','com.ar','co.za','co.nz','es','se','nl','ch','at','dk','be','pl','fi','ie','pt','gr', 'tw', 'com', 'uk', 'de', 'br', 'ca', 'kr', 'mx', 'au', 'za']\n random_domain = random.choice(domains)\n url = f'https://www.google.{random_domain}/search?q={query}&num=100'\n # 请求头\n headers = {\n 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36',\n 'accept-language': 'zh-CN,zh;q=0.9',\n 'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,/;q=0.8,application/signed-exchange;v=b3;q=0.7',\n 'referer': 'https://www.google.com/',\n 'origin': 'https://www.google.com',\n 'Sec-Fetch-Site': 'same-origin',\n 'Sec-Fetch-Mode': 'navigate',\n 'Sec-Fetch-User': '?1',\n 'Sec-Fetch-Dest': 'document'\n }\n # 代理\n proxies = {'http': 'http://127.0.0.1:7890', 'https': 'http://127.0.0.1:7890'}\n response = requests.get(url=url, headers=headers, proxies=proxies, verify=False, timeout=5)\n soup = BeautifulSoup(response.content, 'html.parser')\n # 查找全部div标签\n find_div = soup.find_all('div', {'class': 'yuRUbf'})\n # 开启线程锁\n locks.acquire()\n # 加入列表\n get_url = [url.findNext('a')['href'] + '\n' for url in find_div if 'google.com' not in url.findNext('a')['href']]\n global url_num, dork_finish_num\n url_num += len(get_url)\n dork_finish_num += 1\n print(Fore.GREEN + f'\r{now_time}[INFO]{ "-" * 10}>get Urlnumber：{url_num} Dorsk number：{dork_finish_num} / {dork_total_num}', end='' + Fore.RESET)\n # 写入文件\n write_info(filename, get_url)\n # 释放线程锁\n locks.release()\n\n except TimeoutError:\n pass\n\n\n# 写入文件函数\ndef write_info(filename, get_url):\n\n with open(filename, mode='a+', encoding='utf-8') as file:\n\n file.writelines(get_url)\n\n\nif name == 'main':\n while True:\n try:\n init() # 初始化颜色模块\n dorks_file = input(Fore.YELLOW + f'\n{now_time}[INFO]{ "-" * 10}>input file:' + Fore.RESET)\n print('')\n filename = input(Fore.YELLOW + f'\n{now_time}[INFO]{ "-" * 10}>output file:' + Fore.RESET)\n # 接受work函数返回的元组\n dork_total_num, query_list = work(dorks_file)\n # 定义全局变量完成数量/URL数量\n dork_finish_num = url_num = 0\n\n # 定义进程池数量\n processes = ProcessPoolExecutor(max_workers=20)\n\n # 定义全局锁\n threads_lock = Lock()\n\n # 定义队列\n result_queue = Queue()\n\n # 分配进程池任务\n futures = [processes.submit(google_serach, dokr_list, threads_lock, filename, result_queue) for dokr_list in query_list]\n\n for future in as_completed(futures):\n future.result()\n\n processes.shutdown()\n\n if len(sys.argv) == 1:\n pass\n\n input(Fore.YELLOW + f'\n\n{now_time}[INFO]{"-" * 10}>final huiche' + Fore.RESET)\n break\n # 文件为空\n except FileNotFoundError:\n print(Fore.RED + f'{now_time}[Error]{"-" * 10}>fiel not find' + Fore.RESET)\n # 中断异常\n except KeyboardInterrupt:\n sys.exit(1)

使用多进程替代多线程：可以使用concurrent.futures.ProcessPoolExecutor代替ThreadPoolExecutor，这样可以利用多个进程并行执行任务，提高速度。

from concurrent.futures import ProcessPoolExecutor
...

# 定义进程池数量
processes = ProcessPoolExecutor(max_workers=20)

# 分配进程池任务
for dokr_list in query_list:
    processes.submit(google_serach, dokr_list, threads_lock, filename)

processes.shutdown()

使用requests.Session进行会话维持：可以使用requests.Session对象来发送请求，这样可以维持会话，避免重复建立连接和发送相同的请求头。

session = requests.Session()

...

response = session.get(url=url, headers=headers, proxies=proxies, verify=False, timeout=5)

使用ThreadPoolExecutor.map进行任务分发：可以使用ThreadPoolExecutor.map方法代替手动分发任务，简化代码。

...

with ThreadPoolExecutor(max_workers=20) as executor:
    executor.map(lambda dork: google_serach(dork, threads_lock, filename), query_list)

...

使用Queue进行结果收集：可以使用queue.Queue来收集线程/进程返回的结果，避免全局变量的使用。

from queue import Queue

...

# 定义全局变量完成数量/URL数量
dork_finish_num = url_num = 0

# 定义队列
result_queue = Queue()

...

# Google搜索
def google_serach(query, locks, filename, result_queue):
    try:
        ...

        # 加入队列
        result_queue.put(get_url)

        ...

使用concurrent.futures.as_completed获取任务完成顺序：可以使用concurrent.futures.as_completed方法获取任务的完成顺序，避免依赖全局变量的顺序。

from concurrent.futures import as_completed

...

# 分配进程池任务
futures = [processes.submit(google_serach, dokr_list, threads_lock, filename) for dokr_list in query_list]

for future in as_completed(futures):
    future.result()

...

综上所述，优化后的代码如下：

import requests
import time
import os
import urllib3
import sys
import random
from bs4 import BeautifulSoup
from concurrent.futures import ProcessPoolExecutor, as_completed
from threading import Lock
from colorama import Fore, init
from queue import Queue

now_time = time.strftime('%Y-%m-%d %H-%M')


# 读取Dorks
def work(dorks):
    with open(dorks, mode='r', encoding='utf-8') as file:
        read_content = file.readlines()\n        # 将内容加入列表\n        content = [result.strip() for result in read_content]\n        # 返回数量丢给任务池\n        return len(read_content), content\n\n\n# Google搜索
def google_serach(query, locks, filename, result_queue):
    try:\n        # 关闭HTTPS报错信息\n        urllib3.disable_warnings()\n        filename = os.path.join(os.getcwd(), f'{filename}.txt')\n        domains = ['fr','it','ca','co.uk','ru','co,jp','co.kr','com.au','co.in','com.br','com.ar','co.za','co.nz','es','se','nl','ch','at','dk','be','pl','fi','ie','pt','gr', 'tw', 'com', 'uk', 'de', 'br', 'ca', 'kr', 'mx', 'au', 'za']\n        random_domain = random.choice(domains)\n        url = f'https://www.google.{random_domain}/search?q={query}&num=100'\n        # 请求头\n        headers = {\n                   'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36',\n                   'accept-language': 'zh-CN,zh;q=0.9',\n                   'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',\n                   'referer': 'https://www.google.com/',\n                   'origin': 'https://www.google.com',\n                   'Sec-Fetch-Site': 'same-origin',\n                   'Sec-Fetch-Mode': 'navigate',\n                   'Sec-Fetch-User': '?1',\n                   'Sec-Fetch-Dest': 'document'\n        }\n        # 代理\n        proxies = {'http': 'http://127.0.0.1:7890', 'https': 'http://127.0.0.1:7890'}\n        response = requests.get(url=url, headers=headers, proxies=proxies, verify=False, timeout=5)\n        soup = BeautifulSoup(response.content, 'html.parser')\n        # 查找全部div标签\n        find_div = soup.find_all('div', {'class': 'yuRUbf'})\n        # 开启线程锁\n        locks.acquire()\n        # 加入列表\n        get_url = [url.findNext('a')['href'] + '\n' for url in find_div if 'google.com' not in url.findNext('a')['href']]\n        global url_num, dork_finish_num\n        url_num += len(get_url)\n        dork_finish_num += 1\n        print(Fore.GREEN + f'\r{now_time}[INFO]{ "-" * 10}>get Urlnumber：{url_num}  Dorsk number：{dork_finish_num} / {dork_total_num}', end='' + Fore.RESET)\n        # 写入文件\n        write_info(filename, get_url)\n        # 释放线程锁\n        locks.release()\n\n    except TimeoutError:\n        pass\n\n\n# 写入文件函数\ndef write_info(filename, get_url):\n\n    with open(filename, mode='a+', encoding='utf-8') as file:\n\n        file.writelines(get_url)\n\n\nif __name__ == '__main__':\n    while True:\n        try:\n            init()  # 初始化颜色模块\n            dorks_file = input(Fore.YELLOW + f'\n{now_time}[INFO]{ "-" * 10}>input file:' + Fore.RESET)\n            print('')\n            filename = input(Fore.YELLOW + f'\n{now_time}[INFO]{ "-" * 10}>output file:' + Fore.RESET)\n            # 接受work函数返回的元组\n            dork_total_num, query_list = work(dorks_file)\n            # 定义全局变量完成数量/URL数量\n            dork_finish_num = url_num = 0\n\n            # 定义进程池数量\n            processes = ProcessPoolExecutor(max_workers=20)\n\n            # 定义全局锁\n            threads_lock = Lock()\n\n            # 定义队列\n            result_queue = Queue()\n\n            # 分配进程池任务\n            futures = [processes.submit(google_serach, dokr_list, threads_lock, filename, result_queue) for dokr_list in query_list]\n\n            for future in as_completed(futures):\n                future.result()\n\n            processes.shutdown()\n\n            if len(sys.argv) == 1:\n                pass\n\n            input(Fore.YELLOW + f'\n\n{now_time}[INFO]{"-" * 10}>final huiche' + Fore.RESET)\n            break\n        # 文件为空\n        except FileNotFoundError:\n            print(Fore.RED + f'{now_time}[Error]{"-" * 10}>fiel not find' + Fore.RESET)\n        # 中断异常\n        except KeyboardInterrupt:\n            sys.exit(1)