Python多进程操作MySQLdb时连接丢失及线程池写入问题求助

Hey there! Let's dig into why you're hitting those "Lost connection to database" errors with multiprocessing, and why ThreadPool's map/map_async is giving you grief when writing to MySQL. I'll walk through the root causes and concrete fixes below.

Why Multiprocessing Breaks Your DB Connection #

The core issue here is that MySQL database connections are not designed to be shared across processes. Each Python process has its own isolated memory space—if you create a connection in the main process and pass it to a child process (like your check_writing process), the underlying socket connection gets mangled when multiple processes try to use it. MySQL sees this inconsistent state and drops the connection, hence the error.

Even if you don't explicitly pass the connection, if your child processes inherit it (which they do by default in multiprocessing), the same problem occurs—two processes fighting over the same connection handle is a recipe for disaster.

Why ThreadPool.map/map_async Fails #

Thread-based issues usually boil down to one of two things:

• Shared connection conflicts: MySQL connections are single-threaded by design. If all your threads reuse the same connection, concurrent requests will clash, leading to unexpected errors or dropped connections.
• Unmanaged connections: If you create a new connection per thread but don't clean them up properly, you'll hit MySQL's max_connections limit quickly, blocking new writes.

Fixes for Multiprocessing #

1. Create a New Connection in Each Process #

Every process (including your check_writing monitor) needs to initialize its own database connection. Never reuse a connection from the main process. Here's a quick example:

import multiprocessing
import pymysql
import time

def bulk_update(chunk):
    # Initialize connection INSIDE the child process
    conn = pymysql.connect(host="your_host", user="user", password="pass", db="your_db")
    cursor = conn.cursor()
    try:
        # Use bulk update logic here (e.g., CASE WHEN for batch updates)
        update_query = """
            UPDATE your_table 
            SET col = CASE id 
                {cases}
            END
            WHERE id IN ({ids})
        """
        cases = "\n".join([f"WHEN {item['id']} THEN '{item['val']}'" for item in chunk])
        ids = ",".join([str(item['id']) for item in chunk])
        cursor.execute(update_query.format(cases=cases, ids=ids))
        conn.commit()
    except Exception as e:
        conn.rollback()
        print(f"Chunk failed: {e}")
    finally:
        cursor.close()
        conn.close()

def monitor_process():
    # Separate connection for the monitor
    conn = pymysql.connect(host="your_host", user="user", password="pass", db="your_db")
    cursor = conn.cursor()
    while True:
        cursor.execute("SHOW PROCESSLIST")
        print("Current MySQL processes:", cursor.fetchall())
        time.sleep(5)  # Don't spam the DB!
    # Cleanup if you add an exit condition
    cursor.close()
    conn.close()

if __name__ == "__main__":
    # Split your large dataset into chunks
    data_chunks = split_your_data_into_chunks(your_large_dataset)
    
    # Start update pool
    update_pool = multiprocessing.Pool(processes=4)
    update_pool.map(bulk_update, data_chunks)
    
    # Start monitor
    monitor = multiprocessing.Process(target=monitor_process)
    monitor.start()
    monitor.join()
    
    update_pool.close()
    update_pool.join()

2. Avoid Inheriting Connections #

Make sure your main process doesn't create a connection before spawning child processes—if it does, the child processes might inherit it accidentally. Always delay connection creation until inside the child process.

Fixes for ThreadPool #

1. Use a Connection Pool #

The easiest way to handle threading with MySQL is to use a connection pool library like DBUtils. It automatically assigns a unique connection to each thread, preventing conflicts. Here's how to set it up:

from concurrent.futures import ThreadPoolExecutor
from DBUtils.PooledDB import PooledDB
import pymysql

# Initialize connection pool
conn_pool = PooledDB(
    creator=pymysql,
    maxconnections=10,  # Match this to your thread count
    host="your_host",
    user="user",
    password="pass",
    db="your_db"
)

def update_single_item(item):
    conn = conn_pool.connection()
    cursor = conn.cursor()
    try:
        cursor.execute("UPDATE your_table SET col=%s WHERE id=%s", (item['val'], item['id']))
        conn.commit()
    except Exception as e:
        conn.rollback()
        print(f"Update failed for id {item['id']}: {e}")
    finally:
        cursor.close()
        conn.close()

if __name__ == "__main__":
    with ThreadPoolExecutor(max_workers=5) as executor:
        executor.map(update_single_item, your_large_dataset)

2. No Pool? Create a Connection Per Thread #

If you don't want to use a pool, explicitly create a new connection inside each thread's task function, and close it when done. Never share a connection across threads.

Bonus Optimization Tips #

• Batch Updates: Instead of updating one row at a time, use bulk queries (like CASE WHEN or INSERT ... ON DUPLICATE KEY UPDATE) to reduce the number of DB calls—this is way faster and easier on connections.
• Tweak MySQL Config: If connections are getting dropped prematurely, increase wait_timeout and interactive_timeout in your MySQL config to keep connections alive longer. Also, ensure max_connections is high enough for your process/thread count.
• Throttle Monitoring: Your check_writing process doesn't need to query SHOW PROCESSLIST every second—add a 5-10 second sleep to avoid overwhelming the DB.

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