Python并发实现咨询：匹配特定模式多文件单词计数的并发方案

Hey there! Nice to see you digging into Python concurrency for this word-counting task. Since you've already worked with multiprocessing, let's break down several other effective approaches to implement your count_words('/foo/bar/*.txt') function, complete with code examples and use-case notes.

1. Thread-Based Concurrency with concurrent.futures.ThreadPoolExecutor #

File I/O is IO-bound (most time is spent waiting for data to read/write), which makes threading a great fit here—Python's GIL doesn't block us much for IO tasks. The ThreadPoolExecutor simplifies managing worker threads.

Here's how you can implement it:

import glob
from concurrent.futures import ThreadPoolExecutor
from collections import defaultdict

def count_words_in_file(file_path):
    word_counts = defaultdict(int)
    with open(file_path, 'r', encoding='utf-8') as f:
        for line in f:
            for word in line.strip().split():
                word_counts[word] += 1
    return word_counts

def merge_counts(all_counts):
    final_counts = defaultdict(int)
    for counts in all_counts:
        for word, count in counts.items():
            final_counts[word] += count
    return final_counts

def count_words(pattern):
    files = glob.glob(pattern)
    with ThreadPoolExecutor() as executor:
        results = executor.map(count_words_in_file, files)
    return merge_counts(results)

When to use this: Great for large numbers of files where most time is spent reading from disk. It's lighter on memory than multiprocessing since threads share the same memory space.

2. AsyncIO with asyncio and aiofiles #

For high-concurrency IO tasks, asyncio is another powerful option. It uses a single thread with non-blocking IO, which can be more efficient than threading for very large numbers of files.

You'll need aiofiles (install with pip install aiofiles) for async file operations:

import glob
import asyncio
import aiofiles
from collections import defaultdict

async def count_words_in_file(file_path):
    word_counts = defaultdict(int)
    async with aiofiles.open(file_path, 'r', encoding='utf-8') as f:
        async for line in f:
            for word in line.strip().split():
                word_counts[word] += 1
    return word_counts

async def gather_counts(files):
    tasks = [count_words_in_file(file) for file in files]
    return await asyncio.gather(*tasks)

def count_words(pattern):
    files = glob.glob(pattern)
    all_counts = asyncio.run(gather_counts(files))
    final_counts = defaultdict(int)
    for counts in all_counts:
        for word, count in counts.items():
            final_counts[word] += count
    return final_counts

When to use this: Ideal when you have hundreds or thousands of small files—asyncio's non-blocking model minimizes overhead from thread management.

3. Simplified Multiprocessing with concurrent.futures.ProcessPoolExecutor #

If you already used multiprocessing, you might appreciate the cleaner API of ProcessPoolExecutor (it's a wrapper around multiprocessing). This is perfect if your word-counting logic becomes CPU-bound (e.g., adding complex word normalization like stemming/lemmatization).

import glob
from concurrent.futures import ProcessPoolExecutor
from collections import defaultdict

def count_words_in_file(file_path):
    # Same as the thread-based version, or add CPU-heavy processing here
    word_counts = defaultdict(int)
    with open(file_path, 'r', encoding='utf-8') as f:
        for line in f:
            for word in line.strip().split():
                # Example CPU-heavy step: normalize to lowercase and strip punctuation
                word = word.lower().strip('.,!?')
                word_counts[word] += 1
    return word_counts

def merge_counts(all_counts):
    final_counts = defaultdict(int)
    for counts in all_counts:
        for word, count in counts.items():
            final_counts[word] += count
    return final_counts

def count_words(pattern):
    files = glob.glob(pattern)
    with ProcessPoolExecutor() as executor:
        results = executor.map(count_words_in_file, files)
    return merge_counts(results)

When to use this: Use this if your word processing is CPU-intensive (not just reading files). Processes bypass the GIL, so they can utilize multiple CPU cores fully.

4. Hybrid Approach: Processes + Threads #

For a mix of IO-bound file reading and CPU-bound word processing, you can combine both models: use processes to handle the CPU-heavy work, and threads within each process to handle file IO. This is more advanced but can optimize performance for mixed workloads.

import glob
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
from collections import defaultdict

def process_file(file_path):
    # Threaded file reading within a process
    def read_file(fp):
        with open(fp, 'r', encoding='utf-8') as f:
            return f.read()
    
    with ThreadPoolExecutor(max_workers=1) as thread_exec:
        content = thread_exec.submit(read_file, file_path).result()
    
    # CPU-heavy word processing here
    word_counts = defaultdict(int)
    for word in content.strip().split():
        word = word.lower().strip('.,!?')  # Example normalization
        word_counts[word] += 1
    return word_counts

def count_words(pattern):
    files = glob.glob(pattern)
    with ProcessPoolExecutor() as proc_exec:
        results = proc_exec.map(process_file, files)
    final_counts = defaultdict(int)
    for counts in results:
        for word, count in counts.items():
            final_counts[word] += count
    return final_counts

When to use this: Best for workloads where you spend significant time both reading files (IO-bound) and processing words (CPU-bound).

Quick Decision Guide #

• IO-bound only (simple word count): ThreadPoolExecutor or AsyncIO
• CPU-bound word processing: ProcessPoolExecutor
• Mixed IO/CPU: Hybrid approach

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