面试题：Python复杂代码片段深度分析与改进（专家难度）

Question

以下是一段用于处理文本文件中单词频率统计的Python代码，并且使用了多线程来提高处理效率： ```python import threading import concurrent.futures import collections class WordCounter: def __init__(self, file_path): self.file_path = file_path self.lock = threading.Lock() self.word_count = collections.Counter() def count_words(self, start, end): with open(self.file_path, 'r') as file: file.seek(start) data = file.read(end - start) words = data.split() local_count = collections.Counter(words) with self.lock: self.word_count += local_count def process_file(self): file_size = 0 with open(self.file_path, 'r') as file: file.seek(0, 2) file_size = file.tell() num_threads = 4 part_size = file_size // num_threads threads = [] for i in range(num_threads): start = i * part_size end = (i + 1) * part_size if i < num_threads - 1 else file_size thread = threading.Thread(target=self.count_words, args=(start, end)) threads.append(thread) thread.start() for thread in threads: thread.join() return self.word_count if __name__ == '__main__': counter = WordCounter('large_text_file.txt') result = counter.process_file() print(result.most_common(10)) ``` 1. 分析这段代码可能存在的潜在问题，包括但不限于线程安全、资源竞争、文件读取效率等方面。 2. 提出改进方案，使得代码在正确性、性能和可维护性上都得到提升，可考虑使用 `concurrent.futures` 模块中的其他特性进行优化。

Answer 1

1. 潜在问题分析

• 线程安全：
  • 当前代码使用了 threading.Lock 来确保 self.word_count 的线程安全，在一定程度上解决了资源竞争问题。但是在读取文件部分，虽然按字节范围读取看似独立，但如果文件编码不是简单的ASCII编码（如UTF - 8可能存在字节跨边界问题），可能会导致单词解析错误。
• 资源竞争：
  • 整体上通过锁机制对 self.word_count 的更新操作进行了保护，避免了直接的资源竞争。然而，如果在实际应用中，有其他部分也对 self.word_count 进行操作且没有正确使用锁，仍可能出现资源竞争。
• 文件读取效率：
  • 按固定字节数划分文件读取范围可能会导致单词截断。例如，一个单词正好跨两个线程的读取范围，就会被错误地分割，导致统计不准确。另外，file.seek 和 file.read 操作在多线程环境下频繁调用，可能存在一定的性能开销。

2. 改进方案

import concurrent.futures
import collections


class WordCounter:
    def __init__(self, file_path):
        self.file_path = file_path
        self.word_count = collections.Counter()

    def count_words(self, lines):
        local_count = collections.Counter()
        for line in lines:
            words = line.split()
            local_count.update(words)
        return local_count

    def process_file(self):
        with open(self.file_path, 'r') as file:
            lines = file.readlines()
        num_threads = 4
        parts = [lines[i::num_threads] for i in range(num_threads)]
        with concurrent.futures.ThreadPoolExecutor(max_workers=num_threads) as executor:
            futures = [executor.submit(self.count_words, part) for part in parts]
            for future in concurrent.futures.as_completed(futures):
                self.word_count += future.result()
        return self.word_count


if __name__ == '__main__':
    counter = WordCounter('large_text_file.txt')
    result = counter.process_file()
    print(result.most_common(10))

• 正确性提升：
  • 通过 readlines 读取文件所有行，然后按行划分给不同线程处理，避免了单词截断问题，保证了单词统计的准确性。
• 性能提升：
  • 使用 concurrent.futures.ThreadPoolExecutor 简化了多线程的创建和管理。同时减少了 file.seek 和 file.read 的频繁调用，提高了文件读取效率。
• 可维护性提升：
  • 代码结构更加清晰，将单词统计逻辑封装在 count_words 方法中，process_file 方法专注于任务分配和结果汇总，提高了代码的可读性和可维护性。