[已解决]多线程存储文件导致文件缺失问题

Jack来学c

import json
import re
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from queue import Queue
from urllib.parse import urljoin
import jieba
from spellchecker import SpellChecker
import jieba.analyse
from bs4 import BeautifulSoup

lock = threading.Lock()
queue = Queue()

def load_stopwords(file_path):
    with open(file_path, 'r', encoding='utf-8-sig') as file:
        stop_words = [line.strip() for line in file]
    return stop_words


stop_words = load_stopwords('A:/搜索引擎系统/停用词表.txt')


def get_all_hrefs(text: str, current_url):
    all_href = []
    # 使用BeautifulSoup来解析网页源码
    soup = BeautifulSoup(text, 'lxml')
    # 获取所有a标签下的所有href属性
    all_a_tags = soup.find_all('a')
    for a_tag in all_a_tags:
        href = a_tag.get('href')
        if href and not href.startswith('#') and not href.startswith('javascript:'):
            absolute_url = urljoin(current_url, href)
            # 将https替换为http避免重复
            new_url = absolute_url.replace("https", "http")
            all_href.append(new_url)
    # 将得到的链接返回
    return all_href



def is_chinese(word):
    for char in word:
        if '\u4e00' <= char <= '\u9fff':  # 中文字符的Unicode范围
            return True
    return False


def remove_gibberish_words(words):
    spell = SpellChecker()
    filtered_words = []
    for word in words:
        if not is_chinese(word):  # 检查单词是否为中文
            if spell.known([word]):  # 对于非中文单词进行拼写检查
                filtered_words.append(word)
        else:
            filtered_words.append(word)
    return filtered_words


def Separate_words(content):
    # 使用正则表达式的 sub 方法将标点符号替换为空格
    content = re.sub("[^\w\s]+", "", content)
    content = list(jieba.cut(content, cut_all=False, use_paddle=True))
    content = remove_gibberish_words(content)
    content = [element for element in content if element.strip() != '']
    content = [word for word in content if word.lower() not in stop_words]

    return content


def to_text(num, current_url, title, content, links):
    with lock:
        data = {
            "id": num,
            "url": current_url,
            "title": title,
            "content": content,
            "links": links
        }
        with open(f'A:/搜索引擎系统/Test/web{page}.txt', 'w', encoding='utf-8-sig') as file:
            json.dump(data, file, ensure_ascii=False)


def get_title_and_content(current_url, html):
    soup = BeautifulSoup(html, 'html.parser')

    title = []
    if soup.get('title') != None:
        title = soup.find('title').get_text()

    content = soup.get_text()
    content = content.replace('\n', '')

    links = get_all_hrefs(html, current_url)
    return title, content, links


def do_work(page):
    try:
        with open(f'A:/搜索引擎系统/all_information/web{page}.txt', 'r', encoding='utf-8-sig') as file:

            text = file.read()
            num = text.split('\n')[0]
            current_url = text.split('\n')[1]
            html = str(text.split('\n')[2:])

            title, content, links = get_title_and_content(current_url, html)
            content = Separate_words(content)
            to_text(num, current_url, title, content, links)
            print(num)


    except FileNotFoundError:
        # 文件不存在时的处理
        print("文件不存在，请检查文件路径！")


if __name__ == '__main__':

    page = 1
    target = 100
    for num in range(1, target + 1):
        queue.put(num)

    start_time = time.time()

    with ThreadPoolExecutor(36) as pool:
        while not queue.empty():
            page = queue.get()
            pool.submit(do_work, page)

            time.sleep(0.1)
    pool.shutdown(cancel_futures=True)


    end_time = time.time()
    print(end_time - start_time)




我使用多线程处理得到的html数据并存储，遇到了文件缺失的问题

最佳答案

月排行榜 / 总排行榜

陶远航

2023-9-23 20:43:12

通过阅读您提供的代码，我注意到在多线程处理的过程中，您使用了一个共享的队列 queue 来存储待处理的任务，并且每个线程从队列中获取任务并进行处理。然而，在您的代码中，并没有对文件操作进行加锁，这可能导致多个线程同时访问同一个文件，从而导致文件缺失问题。

为了解决这个问题，您可以考虑使用文件锁来确保同一时间只有一个线程可以访问文件。在 to_text 函数中，您可以将文件写入操作放在一个临界区域内，然后使用 lock 对象对该临界区域进行加锁。这样可以确保每次只有一个线程进行文件写入操作，避免多个线程同时写入同一个文件。

下面是修改后的代码示例：

1. import json
   
2. import re
   
3. import threading
   
4. import time
   
5. from concurrent.futures import ThreadPoolExecutor
   
6. from queue import Queue
   
7. from urllib.parse import urljoin
   
8. import jieba
   
9. from spellchecker import SpellChecker
   
10. import jieba.analyse
    
11. from bs4 import BeautifulSoup
    
12. 
    
13. lock = threading.Lock()
    
14. queue = Queue()
    
15. 
    
16. def load_stopwords(file_path):
    
17.     with open(file_path, 'r', encoding='utf-8-sig') as file:
    
18.         stop_words = [line.strip() for line in file]
    
19.     return stop_words
    
20. 
    
21. # 其他函数定义...
    
22. 
    
23. def to_text(num, current_url, title, content, links):
    
24.     with lock:
    
25.         data = {"id": num, "url": current_url, "title": title, "content": content, "links": links}
    
26.         with open(f'A:/搜索引擎系统/Test/web{num}.txt', 'w', encoding='utf-8-sig') as file:
    
27.             json.dump(data, file, ensure_ascii=False)
    
28. 
    
29. # 其他函数定义...
    
30. 
    
31. def do_work(page):
    
32.     try:
    
33.         with open(f'A:/搜索引擎系统/all_information/web{page}.txt', 'r', encoding='utf-8-sig') as file:
    
34.             text = file.read()
    
35.             num = text.split('\n')[0]
    
36.             current_url = text.split('\n')[1]
    
37.             html = str(text.split('\n')[2:])
    
38.             title, content, links = get_title_and_content(current_url, html)
    
39.             content = Separate_words(content)
    
40.             to_text(num, current_url, title, content, links)
    
41.             print(num)
    
42.     except FileNotFoundError:
    
43.         print("文件不存在，请检查文件路径！")
    
44. 
    
45. if __name__ == '__main__':
    
46.     # 设置线程数
    
47.     num_threads = 36
    
48.     page = 1
    
49.     target = 100
    
50. 
    
51.     for num in range(1, target + 1):
    
52.         queue.put(num)
    
53. 
    
54.     start_time = time.time()
    
55. 
    
56.     with ThreadPoolExecutor(num_threads) as pool:
    
57.         while not queue.empty():
    
58.             page = queue.get()
    
59.             pool.submit(do_work, page)
    
60.             time.sleep(0.1)
    
61. 
    
62.     pool.shutdown(cancel_futures=True)
    
63. 
    
64.     end_time = time.time()
    
65.     print(end_time - start_time)
    

复制代码

在修改后的代码中，我对 to_text 函数进行了修改，在写入文件之前使用 lock 进行加锁，确保每次只有一个线程能够执行写入操作。

希望这些修改能够解决您遇到的文件缺失问题。如果问题仍然存在，请提供更多详细信息，以便我能够给出进一步的建议。
如果问题已经解决，请设置最佳答案

跳转到最佳答案楼层

isdkz

可能是由于多个线程同时访问同一个文件，导致文件被覆盖或者同时读取的问题。可以使用文件锁来解决这个问题。在写文件之前，使用 with lock: 语句获取锁，确保同一时刻只有一个线程在访问文件。另外，建议使用 with open 语句来打开文件，可以自动关闭文件，避免文件被占用的情况。

陶远航

通过阅读您提供的代码，我注意到在多线程处理的过程中，您使用了一个共享的队列 queue 来存储待处理的任务，并且每个线程从队列中获取任务并进行处理。然而，在您的代码中，并没有对文件操作进行加锁，这可能导致多个线程同时访问同一个文件，从而导致文件缺失问题。

为了解决这个问题，您可以考虑使用文件锁来确保同一时间只有一个线程可以访问文件。在 to_text 函数中，您可以将文件写入操作放在一个临界区域内，然后使用 lock 对象对该临界区域进行加锁。这样可以确保每次只有一个线程进行文件写入操作，避免多个线程同时写入同一个文件。

下面是修改后的代码示例：

1. import json
   
2. import re
   
3. import threading
   
4. import time
   
5. from concurrent.futures import ThreadPoolExecutor
   
6. from queue import Queue
   
7. from urllib.parse import urljoin
   
8. import jieba
   
9. from spellchecker import SpellChecker
   
10. import jieba.analyse
    
11. from bs4 import BeautifulSoup
    
12. 
    
13. lock = threading.Lock()
    
14. queue = Queue()
    
15. 
    
16. def load_stopwords(file_path):
    
17.     with open(file_path, 'r', encoding='utf-8-sig') as file:
    
18.         stop_words = [line.strip() for line in file]
    
19.     return stop_words
    
20. 
    
21. # 其他函数定义...
    
22. 
    
23. def to_text(num, current_url, title, content, links):
    
24.     with lock:
    
25.         data = {"id": num, "url": current_url, "title": title, "content": content, "links": links}
    
26.         with open(f'A:/搜索引擎系统/Test/web{num}.txt', 'w', encoding='utf-8-sig') as file:
    
27.             json.dump(data, file, ensure_ascii=False)
    
28. 
    
29. # 其他函数定义...
    
30. 
    
31. def do_work(page):
    
32.     try:
    
33.         with open(f'A:/搜索引擎系统/all_information/web{page}.txt', 'r', encoding='utf-8-sig') as file:
    
34.             text = file.read()
    
35.             num = text.split('\n')[0]
    
36.             current_url = text.split('\n')[1]
    
37.             html = str(text.split('\n')[2:])
    
38.             title, content, links = get_title_and_content(current_url, html)
    
39.             content = Separate_words(content)
    
40.             to_text(num, current_url, title, content, links)
    
41.             print(num)
    
42.     except FileNotFoundError:
    
43.         print("文件不存在，请检查文件路径！")
    
44. 
    
45. if __name__ == '__main__':
    
46.     # 设置线程数
    
47.     num_threads = 36
    
48.     page = 1
    
49.     target = 100
    
50. 
    
51.     for num in range(1, target + 1):
    
52.         queue.put(num)
    
53. 
    
54.     start_time = time.time()
    
55. 
    
56.     with ThreadPoolExecutor(num_threads) as pool:
    
57.         while not queue.empty():
    
58.             page = queue.get()
    
59.             pool.submit(do_work, page)
    
60.             time.sleep(0.1)
    
61. 
    
62.     pool.shutdown(cancel_futures=True)
    
63. 
    
64.     end_time = time.time()
    
65.     print(end_time - start_time)
    

复制代码

在修改后的代码中，我对 to_text 函数进行了修改，在写入文件之前使用 lock 进行加锁，确保每次只有一个线程能够执行写入操作。

希望这些修改能够解决您遇到的文件缺失问题。如果问题仍然存在，请提供更多详细信息，以便我能够给出进一步的建议。
如果问题已经解决，请设置最佳答案

Mike_python小

问题描述：

我使用多线程处理获取到的HTML数据，并进行存储，但是在运行过程中出现了文件缺失的问题。我希望得到解决该问题的方法。

解决方案：

文件缺失的问题可能是由于多线程同时写入文件导致的。在多线程环境下，多个线程同时访问同一个文件，可能会导致文件写入冲突，进而导致文件丢失或损坏。

为了解决这个问题，你可以考虑以下两个方面的改进：

1. 使用文件锁：你已经使用了 threading.Lock() 创建了一个锁对象 lock，但是在写入文件时没有使用该锁进行保护。在 to_text() 函数中，你可以使用锁来保证同一时间只有一个线程可以访问写入文件的代码块。具体做法是，在写入文件之前，调用 lock.acquire() 获取锁，写入文件之后，调用 lock.release() 释放锁。这样可以确保每个线程在写入文件时不会被其他线程打断。

  

2.    def to_text(num, current_url, title, content, links):
3.        with lock:
4.            # 获取锁
5.            data = {
6.                "id": num,
7.                "url": current_url,
8.                "title": title,
9.                "content": content,
10.                "links": links
11.            }
12.            with open(f'A:/搜索引擎系统/Test/web{page}.txt', 'w', encoding='utf-8-sig') as file:
13.                json.dump(data, file, ensure_ascii=False)
14.            # 释放锁
15.    

复制代码

2. 合理设置线程池大小：在使用 ThreadPoolExecutor 创建线程池时，可以通过参数 max_workers 来指定线程池的大小。你目前将 max_workers 设置为了 36，这意味着最多同时执行 36 个线程。如果你的机器资源有限，可能会导致同时写入文件的线程过多，进而增加文件冲突的概率。你可以根据系统资源和具体需求来调整线程池的大小，以避免文件缺失问题。

  

2.    with ThreadPoolExecutor(max_workers=10) as pool:
3.        # 线程池的其他代码...
4.    

复制代码

通过以上两个改进，你可以尝试解决文件缺失的问题。希望能帮助到你！如果还有其他问题，请随时追问。

球一个最佳答案谢谢啦！这对我非常重要！

Jack来学c

陶远航 发表于 2023-9-23 20:43
通过阅读您提供的代码，我注意到在多线程处理的过程中，您使用了一个共享的队列 queue 来存储待处理的任务 ...

好的，谢谢