Python多线程邮箱验证报错求助：参数数量不符TypeError

问题描述 #

我是Python新手，需要对CSV文件中的3000条邮箱列表进行多线程验证，参考相关文章后，邮箱验证逻辑本身正常，但多线程实现失败并报错，以下是我的代码及报错信息：

原始代码 #

pool_sema = threading.Semaphore(value=maxthreads)

# make empty list
fields = []
emails_list = []
deliverable_list = []
hostexists_list = []

csvFilename = r'test2.csv'

with open(csvFilename, 'r', encoding="utf-8") as csvFile:
    csvreader = csv.reader(csvFile)
    # extracting field names through first row
    fields = next(csvreader)  # skip the header
    for row in csvreader:
        x,y = [str(value)for value in row] 
        inputAddress = str(y)
        emails_list.append(inputAddress)

def task(i):
    pool_sema.acquire()
    try:
        normal_verifier = Verifier(source_addr='example@gmail.com') 
        results = normal_verifier.verify(i)  # i will be email to be verify
        deliverable_list.append(results["deliverable"])
        hostexists_list.append(results["host_exists"])
        time.sleep(3)
    except Exception as e:
        print("Error")
    finally:
        pool_sema.release()

def create_threads(number_of_threads):
    try:
        threads = [threading.Thread(target=task, args=(i)) for i in emails_list]
        [t.start() for t in threads]
    except Exception as e:
        print("Error")
    finally:
        [t.join() for t in threads]

create_threads((len(emails_list)))

# Create Data frame
# Making number of columns
validating_Email = pd.DataFrame(columns=['Email', 'Deliverable', 'Host exists'])
#Dictionary
data_dictionary = {'Email':emails_list, 'Deliverable':deliverable_list, 'Host exists' :hostexists_list}
# Dictionary to Pandas DataFrame
df = pd.DataFrame.from_dict(data_dictionary)
# Save Data in EXCEL
df.to_excel('gmail_multi_thread_1.xlsx', index = False) 

报错信息 #

Exception in thread Thread-505:
Traceback (most recent call last):
  File "C:\Users\Aaditri\anaconda3\lib\threading.py", line 973, in _bootstrap_inner
Exception in thread Thread-506:
Traceback (most recent call last):
  File "C:\Users\Aaditri\anaconda3\lib\threading.py", line 973, in _bootstrap_inner
Exception in thread Thread-507:
Exception in thread     self.run()
  File "C:\Users\Aaditri\anaconda3\lib\threading.py", line 910, in run

File "C:\Users\Aaditri\anaconda3\lib\threading.py", line 910, in run
    self._target(*self._args, **self._kwargs)
TypeError: task() takes 1 positional argument but 28 were given
Exception in thread Thread-530:
Traceback (most recent call last):
  File "C:\Users\Aaditri\anaconda3\lib\threading.py", line 973, in _bootstrap_inner
Exception in thread     Thread-531:

错误原因分析 #

1. 线程参数传递错误：创建threading.Thread时，args参数要求传入元组。你写的args=(i)会被Python解析成单个变量而非元组，当i是邮箱字符串时，会被拆分成单个字符作为多个参数传递给task函数，导致task()只接受1个参数却收到多个的报错。正确写法是args=(i,)（末尾加逗号明确是元组）。

2. 线程安全问题：deliverable_list和hostexists_list是普通列表，多线程环境下调用append方法不是线程安全的，可能导致数据混乱或丢失，需要用锁来保护列表操作。

3. 未定义maxthreads：代码中pool_sema = threading.Semaphore(value=maxthreads)里的maxthreads变量未定义，需要先设置合理的并发线程数（比如10、20，避免创建3000个线程耗尽系统资源）。

修复后的代码 #

import threading
import csv
import time
import pandas as pd
# 根据实际路径导入Verifier类
from email_verifier import Verifier  

# 设置最大并发线程数
maxthreads = 10
pool_sema = threading.Semaphore(value=maxthreads)
# 定义锁保护列表操作
list_lock = threading.Lock()

# 初始化列表
fields = []
emails_list = []
deliverable_list = []
hostexists_list = []

csvFilename = r'test2.csv'

with open(csvFilename, 'r', encoding="utf-8") as csvFile:
    csvreader = csv.reader(csvFile)
    fields = next(csvreader)  # 跳过表头
    for row in csvreader:
        x, y = [str(value) for value in row]
        inputAddress = str(y)
        emails_list.append(inputAddress)

def task(i):
    pool_sema.acquire()
    try:
        normal_verifier = Verifier(source_addr='example@gmail.com')
        results = normal_verifier.verify(i)
        # 加锁保护列表append操作
        with list_lock:
            deliverable_list.append(results["deliverable"])
            hostexists_list.append(results["host_exists"])
        time.sleep(3)
    except Exception as e:
        print(f"验证邮箱{i}时出错: {str(e)}")
    finally:
        pool_sema.release()

def create_threads():
    threads = []
    try:
        # 修正args参数为元组
        threads = [threading.Thread(target=task, args=(i,)) for i in emails_list]
        for t in threads:
            t.start()
    except Exception as e:
        print(f"创建线程时出错: {str(e)}")
    finally:
        for t in threads:
            t.join()

create_threads()

# 生成DataFrame并保存
df = pd.DataFrame({
    'Email': emails_list,
    'Deliverable': deliverable_list,
    'Host exists': hostexists_list
})
df.to_excel('gmail_multi_thread_1.xlsx', index=False)

额外注意事项 #

• 并发数控制：maxthreads不要设置太大，建议根据网络情况和系统资源调整（比如10-20），避免触发邮件服务器的频率限制或耗尽本地资源。
• Verifier复用：如果Verifier实例可以复用，建议把它的初始化放到task函数外面，避免重复创建实例，提升性能。
• 异常处理细化：原代码的异常处理太笼统，修复后的代码增加了具体邮箱和错误信息，方便排查问题。

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