如何在Python中将生成的XML文件加入队列供线程池执行并确保队列含8000个文件？

Alright, let's break this down and fix/extend your code to meet your goal: generating 8000 XML files via a thread pool with proper task queuing. First, let's clean up the small issues in your existing code, then build out the thread pool logic.

First, Fix the Minor Code Issues #

• Your frange function has a typo: tmp += ste should be tmp += step
• Windows file paths need careful handling—use raw strings (prefix with r) or double backslashes to avoid escape character errors. For example:

  os.chdir(r'C:\Users\Aravind_Sampathkumar\Desktop\IMLC')
  tree = ET.parse(r'C:\Users\Aravind_Sampathkumar\Desktop\IMLC\BO\IMLC_v4p8_Aravind.xml')
  

Core Implementation: Thread Pool with Task Queuing #

We'll use Python's built-in concurrent.futures.ThreadPoolExecutor to manage parallel XML generation. This module handles the task queue internally, but we'll also cover explicit queue control if you need it.

Step 1: Define the XML Generation Task Function #

Wrap your XML modification/save logic into a reusable function that each thread will execute:

from lxml import etree as ET
import os
import concurrent.futures

def generate_xml(base_tree, file_number, output_dir):
    """Generate a unique XML file based on the base template."""
    try:
        # Make a deep copy of the base tree to avoid thread safety conflicts
        tree = ET.ElementTree(ET.fromstring(ET.tostring(base_tree.getroot())))
        root = tree.getroot()

        # --- Add your XML modification logic here ---
        # Example: Update a unique identifier element
        # id_elem = root.find('.//unique_id')
        # if id_elem is not None:
        #     id_elem.text = f"FILE_{file_number}"

        # Save the new XML file
        output_path = os.path.join(output_dir, f'generated_{file_number}.xml')
        tree.write(output_path, encoding='utf-8', xml_declaration=True)
        return True, file_number
    except Exception as e:
        print(f"Failed to generate file {file_number}: {str(e)}")
        return False, file_number

Step 2: Set Up Thread Pool and Queue Tasks #

Initialize the thread pool, prepare 8000 generation tasks, and submit them to the pool:

def main():
    # Configuration
    total_files = 8000
    output_dir = r'C:\Users\Aravind_Sampathkumar\Desktop\IMLC\generated_xmls'
    base_xml_path = r'C:\Users\Aravind_Sampathkumar\Desktop\IMLC\BO\IMLC_v4p8_Aravind.xml'
    
    # Create output directory if it doesn't exist
    os.makedirs(output_dir, exist_ok=True)
    
    # Load the base XML template once (outside threads for efficiency)
    base_tree = ET.parse(base_xml_path)

    # Use ThreadPoolExecutor to parallelize tasks
    # Adjust max_workers based on your system's capacity (10-20 is a safe start)
    with concurrent.futures.ThreadPoolExecutor(max_workers=15) as executor:
        # Map the generate_xml function to 8000 unique file numbers
        results = executor.map(
            generate_xml,
            [base_tree]*total_files,
            range(1, total_files+1),
            [output_dir]*total_files
        )

        # Track successful/failed generations
        failed_files = []
        for success, file_num in results:
            if not success:
                failed_files.append(file_num)

    # Final verification
    if failed_files:
        print(f"Warning: {len(failed_files)} files failed to generate. File numbers: {failed_files[:10]}...")
    else:
        generated_count = len([f for f in os.listdir(output_dir) if f.endswith('.xml')])
        print(f"Success! Generated {generated_count} XML files (target: {total_files})")

if __name__ == "__main__":
    main()

Key Details: #

• Thread Safety: We make a deep copy of the base XML tree in each task to avoid race conditions (lxml's ElementTree isn't thread-safe for modifications).
• Automatic Queuing: ThreadPoolExecutor manages the task queue internally—when you call executor.map, it queues all 8000 tasks and distributes them to available threads.
• Error Resilience: Each task catches exceptions individually, so a single failed file won't crash the entire process.

Explicit Queue Implementation (For Manual Control) #

If you need direct control over the task queue (e.g., for dynamic task addition), use queue.Queue with worker threads:

import queue
import threading

def worker(task_queue, base_tree, output_dir):
    while not task_queue.empty():
        file_number = task_queue.get()
        try:
            generate_xml(base_tree, file_number, output_dir)
        finally:
            task_queue.task_done()

def main_with_explicit_queue():
    total_files = 8000
    output_dir = r'C:\Users\Aravind_Sampathkumar\Desktop\IMLC\generated_xmls'
    base_xml_path = r'C:\Users\Aravind_Sampathkumar\Desktop\IMLC\BO\IMLC_v4p8_Aravind.xml'
    
    os.makedirs(output_dir, exist_ok=True)
    base_tree = ET.parse(base_xml_path)

    # Create and populate the task queue
    task_queue = queue.Queue()
    for num in range(1, total_files+1):
        task_queue.put(num)

    # Start worker threads
    num_workers = 15
    threads = []
    for _ in range(num_workers):
        t = threading.Thread(target=worker, args=(task_queue, base_tree, output_dir))
        t.start()
        threads.append(t)

    # Wait for all tasks to complete
    task_queue.join()

    # Verify results
    generated_count = len([f for f in os.listdir(output_dir) if f.endswith('.xml')])
    print(f"Generated {generated_count} XML files (target: {total_files})")

if __name__ == "__main__":
    main_with_explicit_queue()

This approach gives you full control over the task queue, which is useful if you need to add tasks dynamically during execution.

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