多线程不处理完整列表

Question

我正在使用多线程访问从 csv 读取的链接，奇怪的是，不管 max-workers 是什么，甚至当我删除多线程部分时，代码运行的 url 数量也比列表中的任意少。我打印列表以验证计数。例如，如果列表有 5000 个 url，则代码在 4084 处停止，如果链接为 13,000，它将在 9200 处停止，即使只有 130 个链接，它也会在 80 处停止。我在这里做错了什么？

import requests
import xlrd
import concurrent.futures
from bs4 import BeautifulSoup
import csv


header_added = False
file_location = "Urls.xlsx"
workbook = xlrd.open_workbook(file_location)
sheet = workbook.sheet_by_index(0)
all_links = []
for row in range(1, 11000):
     all_links.append(sheet.cell_value(row,0))

print(len(all_links))
i = 0
def get_solution(url):
    global header_added, i
    page = requests.get(url).text
    soup = BeautifulSoup(page, 'html.parser')
    ques_div = soup.find('p', class_='header-description')
    ques = ques_div.find('span').text
    ans_divs = soup.findAll('div', class_='puzzle-solution')
    ans = ans_divs[0].text
    print("Solution ", i)
    i += 1
    dict1 ={"Words": ques, "Solution": ans}
    with open('Results10k.csv', 'a+', encoding='utf-8') as f:
        w = csv.DictWriter(f, dict1.keys())
        if not header_added:
            w.writeheader()
            header_added = True
        w.writerow(dict1)

with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
    result = executor.map(get_solution, all_links)

Answer 1

这是对您的代码进行的一种不需要锁的修改——相反，只有一个进程写入文件。

此外，由于 GIL，使用 ThreadPool 会比使用进程支持的 Pool 慢。

import csv
import multiprocessing

import requests
import xlrd
from bs4 import BeautifulSoup

sess = requests.Session()


def get_solution(url):
    try:
        resp = sess.get(url)
        resp.raise_for_status()
        page = resp.text
        soup = BeautifulSoup(page, "html.parser")
        ques_div = soup.find("p", class_="header-description")
        ques = ques_div.find("span").text.strip()
        ans_divs = soup.findAll("div", class_="puzzle-solution")
        ans = ans_divs[0].text.strip()
        return {"URL": url, "Words": ques, "Solution": ans, "Error": ""}
    except Exception as exc:
        print(url, "Error:", exc)
        return {"URL": url, "Words": "", "Solution": "", "Error": str(exc)}


def read_links(file_location):
    workbook = xlrd.open_workbook(file_location)
    sheet = workbook.sheet_by_index(0)
    all_links = []
    for row in range(1, 11000):
        all_links.append(sheet.cell_value(row, 0))
    return all_links


def main():
    links = read_links("./Urls.xlsx")
    with open("Results10k.csv", "w", encoding="utf-8") as f:
        with multiprocessing.Pool() as p:  # (or multiprocessing.pool.ThreadPool)
            for i, result in enumerate(p.imap_unordered(get_solution, links, chunksize=16)):
                if i == 0:
                    writer = csv.DictWriter(f, result.keys())
                    writer.writeheader()
                writer.writerow(result)
                f.flush()  # Ensure changes are written immediately
                if i % 100 == 0:  # Progress indicator
                    print(i)


if __name__ == "__main__":
    main()

Answer 2

可能是，get_solution() 对于某些 URL 崩溃。您可以在函数体中添加 try/except 并将所有崩溃的 URL 写入不同的文件。

def get_solution(url):
    try:
        ...
    except:
        with open('errors.txt','a+') as f:
            f.write(url+'\n')

如果这是问题，则数字应加起来为总数。 open() 也可能不是线程安全的。

file_lock = threading.Lock()
def get_solution(url):
    with file_lock:
        with open('Results10k.csv', 'a+', encoding='utf-8') as f:
            w = csv.DictWriter(f, dict1.keys())
            ...