如何基于Multiprocessing实现带过滤操作的List Comprehension？

Great question! When you need to combine parallel processing with filtering (like your original l = [i for i in range(10) if fun(i)]), the basic pool.map doesn't handle filtering out of the box—it just returns a result for every input. Here are a few practical, easy-to-implement solutions:

1. Return a Sentinel Value and Post-Process #

The simplest approach is to modify your processing function to return a "sentinel" value (like None) when the filter condition fails, then filter out those sentinels after using pool.map.

Example Code: #

from multiprocessing import Pool

def process_item(i):
    # Replace this with your actual `fun(i)` condition
    if i % 2 == 0:  # Let's say we keep even numbers
        return i
    return None  # Sentinel for filtered-out items

if __name__ == "__main__":
    with Pool() as pool:
        raw_results = pool.map(process_item, range(10))
    # Filter out the sentinel values
    filtered_list = [item for item in raw_results if item is not None]
    print(filtered_list)  # Output: [0, 2, 4, 6, 8]

Pros & Cons: #

• Pros: Super straightforward, minimal code changes, works with all pool methods.
• Cons: Wastes a small amount of memory storing sentinel values if most items are filtered out. Fine for small to medium datasets.

2. Use imap/imap_unordered for Memory-Efficient Filtering #

If you're working with large datasets, pool.imap or pool.imap_unordered are better choices. These return results as an iterator (instead of a full list), so you can filter on-the-fly without storing all raw results in memory.

Example with imap (preserves order): #

from multiprocessing import Pool

def fun(i):
    return i % 2 == 0  # Your original filter condition

def process_item(i):
    return i if fun(i) else None

if __name__ == "__main__":
    with Pool() as pool:
        # Use a generator expression to filter while iterating
        filtered_list = list(
            item for item in pool.imap(process_item, range(10)) 
            if item is not None
        )
    print(filtered_list)

Example with imap_unordered (faster, no order guarantee): #

If you don't care about the order of items in your final list, imap_unordered will return results as soon as they're ready, which can speed up processing:

filtered_list = list(
    item for item in pool.imap_unordered(process_item, range(10)) 
    if item is not None
)

Pros & Cons: #

• Pros: Memory-efficient for large datasets, imap_unordered can be faster.
• Cons: imap_unordered doesn't preserve input order; imap is slightly slower than map but more memory-friendly.

3. Use apply_async with Callbacks (Real-Time Filtering) #

For scenarios where you want to process results as soon as each task finishes (instead of waiting for all tasks), use pool.apply_async with a callback function that handles filtering.

Example Code: #

from multiprocessing import Pool

def fun(i):
    return i % 2 == 0

def process_item(i):
    return i if fun(i) else None

def collect_result(result):
    # Callback runs in the main process, so no race conditions here
    if result is not None:
        filtered_list.append(result)

if __name__ == "__main__":
    filtered_list = []
    with Pool() as pool:
        for i in range(10):
            pool.apply_async(process_item, args=(i,), callback=collect_result)
        pool.close()  # Prevent new tasks from being added
        pool.join()   # Wait for all tasks to complete
    print(filtered_list)

Pros & Cons: #

• Pros: Real-time result collection, useful for long-running tasks where you want to see progress.
• Cons: Slightly more code, and you lose control over the order of results (unless you track it manually).

Which One Should You Choose? #

• Go with Option 1 for small datasets or when simplicity is key.
• Use Option 2 for large datasets where memory is a concern.
• Pick Option 3 if you need to process results as they come in (e.g., logging progress).

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