多线程目录爆破：用map显式传参并隐式传递列表元素的方法

这问题我之前帮不少人捋清楚过，其实核心就是把固定的hostname参数提前绑定到函数上，让线程只需要接收动态的directory参数就行。下面分几种常用场景给你具体实现方案：

方法1：用functools.partial绑定固定参数 #

这是最清晰、最不容易踩坑的方式，partial可以帮你把函数的某个参数固定住，生成一个新的“简化版”函数。

from functools import partial
import threading

def check_directory(hostname, directory):
    # 这里写你的目录检查逻辑，比如发送HTTP请求判断是否存在
    print(f"正在检查: {hostname}/{directory}")

# 你的目标主机和待爆破目录列表
target_host = "http://test.example.com"
dir_list = ["admin", "api", "uploads", "docs", "backup"]

# 把hostname绑定到check_directory的第一个参数，生成新函数
bound_check_func = partial(check_directory, target_host)

# 遍历目录列表创建线程
threads = []
for dir_path in dir_list:
    # 现在只需要给线程传入directory参数就行
    thread = threading.Thread(target=bound_check_func, args=(dir_path,))
    threads.append(thread)
    thread.start()

# 等待所有线程执行完成
for t in threads:
    t.join()

解释：partial(check_directory, target_host)会返回一个新函数，这个函数被调用时，会自动把target_host作为第一个参数传给check_directory，你只需要传入剩下的directory参数即可。

方法2：用Lambda表达式包装函数 #

如果不想导入额外模块，用lambda也能实现，但要注意循环变量的延迟绑定问题——必须把当前的dir_path作为参数传给lambda，不然所有线程都会用循环最后一个元素。

import threading

def check_directory(hostname, directory):
    print(f"正在检查: {hostname}/{directory}")

target_host = "http://test.example.com"
dir_list = ["admin", "api", "uploads", "docs", "backup"]

threads = []
for dir_path in dir_list:
    # 显式把dir_path传给lambda，避免延迟绑定坑
    thread = threading.Thread(target=lambda d: check_directory(target_host, d), args=(dir_path,))
    threads.append(thread)
    thread.start()

for t in threads:
    t.join()

方法3：用线程池（ThreadPoolExecutor）直接传参 #

如果用concurrent.futures的线程池，写法会更简洁，直接在提交任务时传入两个参数即可：

用submit方法 #

from concurrent.futures import ThreadPoolExecutor

def check_directory(hostname, directory):
    print(f"正在检查: {hostname}/{directory}")

target_host = "http://test.example.com"
dir_list = ["admin", "api", "uploads", "docs", "backup"]

with ThreadPoolExecutor(max_workers=5) as executor:
    # 遍历目录，每个任务直接传入hostname和当前目录
    futures = [executor.submit(check_directory, target_host, dir_path) for dir_path in dir_list]
    # 等待所有任务完成（如果需要获取返回值可以在这里处理）
    for future in futures:
        future.result()

用map方法配合partial #

from concurrent.futures import ThreadPoolExecutor
from functools import partial

def check_directory(hostname, directory):
    print(f"正在检查: {hostname}/{directory}")

target_host = "http://test.example.com"
dir_list = ["admin", "api", "uploads", "docs", "backup"]

bound_check_func = partial(check_directory, target_host)

with ThreadPoolExecutor(max_workers=5) as executor:
    # map会自动把dir_list的每个元素传给bound_check_func
    executor.map(bound_check_func, dir_list)

内容的提问来源于stack exchange，提问作者user9232663