Python多线程操作MySQL出现无响应挂起问题求助

使用Python threading模块同时调用两个函数，函数触发后执行MySQL查询，程序陷入无限挂起状态，无任何报错信息。

运行main.py后控制台输出：

a
b
f
test1
test2

后续的USD/EUR guncellendi.始终未打印，程序持续运行但无法完成预期功能。单独运行siparis_kontrolu.py时功能正常。

涉及代码 #

main.py #

import threading
import siparis_kontrolu
import dolar_euro_guncelle

def dd1():
    print("a")
    siparis_kontrolu.siparis_kontrolu()

def dd2():
    print("b")
    dolar_euro_guncelle.dolar_euro_guncelle()

t1 = threading.Thread(target=dd1)
t2 = threading.Thread(target=dd2)

t1.start()
t2.start()

siparis_kontrolu.py #

import mysql_db
def siparis_kontrolu():
    try:
        while True:
            print("test1")
            tum_kullanicilar = mysql_db.mysql_get_all("select * from users")
            print("test2")

dolar_euro_guncelle.py #

import urllib.request
import json
import mysql_db
import time
import logla

def dolar_euro_guncelle():
    while True:
        try:
            print("f")
            data = urllib.request.urlopen(
                "https://finans.truncgil.com/today.json")
            for line in data:
                line = json.loads(line.decode('utf-8'))
                USD = round(float(line['USD']["Satış"].replace(",", ".")), 2)
                EUR = round(float(line['EUR']["Satış"].replace(",", ".")), 2)
                mysql_db.mysql_update("update variables set EUR='" +
                                      str(EUR)+"', USD='"+str(USD)+"' where id='4'")
            time.sleep(10)
            print("USD/EUR guncellendi.")
        except Exception as e:
            logla.logla(e, "dolar_euro_guncelle")
            print(e)

mysql_db.py #

from configparser import ConfigParser
import mysql.connector

config_file = 'config.ini'
config = ConfigParser()
config.read(config_file)

mydb = mysql.connector.connect(host=config['MYSQL']['DB_HOST'],
                               user=config['MYSQL']['DB_USERNAME'], passwd=config['MYSQL']['DB_PASSWORD'], database=config['MYSQL']['DB_DATABASE'])
mycursor = mydb.cursor(buffered=True, dictionary=True)

def mysql_update(sorgu):
    try:
        mycursor.execute(sorgu)
        mydb.commit()
    except Exception as e:
        print(e)

def mysql_get_all(sorgu):
    try:
        mycursor.execute(sorgu)
        return mycursor.fetchall()
    except Exception as e:
        print(e)

问题原因 #

mysql_db.py中全局创建了单个数据库连接mydb和游标mycursor，多线程环境下两个线程会共用这一组连接和游标。但mysql.connector的游标并非线程安全的，当一个线程正在使用游标执行操作时，另一个线程尝试操作同一个游标会引发资源竞争，最终导致程序挂起且无报错输出。

解决方案 #

方案1：每个数据库操作创建独立连接和游标 #

修改mysql_db.py，让每个函数在执行时创建专属的连接和游标，用完后及时关闭，避免共享冲突：

from configparser import ConfigParser
import mysql.connector

config_file = 'config.ini'
config = ConfigParser()
config.read(config_file)

def get_db_connection():
    return mysql.connector.connect(
        host=config['MYSQL']['DB_HOST'],
        user=config['MYSQL']['DB_USERNAME'],
        passwd=config['MYSQL']['DB_PASSWORD'],
        database=config['MYSQL']['DB_DATABASE']
    )

def mysql_update(sorgu):
    conn = None
    cursor = None
    try:
        conn = get_db_connection()
        cursor = conn.cursor(buffered=True, dictionary=True)
        cursor.execute(sorgu)
        conn.commit()
    except Exception as e:
        print(e)
        if conn:
            conn.rollback()
    finally:
        if cursor:
            cursor.close()
        if conn:
            conn.close()

def mysql_get_all(sorgu):
    conn = None
    cursor = None
    result = []
    try:
        conn = get_db_connection()
        cursor = conn.cursor(buffered=True, dictionary=True)
        cursor.execute(sorgu)
        result = cursor.fetchall()
    except Exception as e:
        print(e)
    finally:
        if cursor:
            cursor.close()
        if conn:
            conn.close()
    return result

方案2：使用线程本地存储分配独立资源 #

通过threading.local为每个线程分配专属的连接和游标，避免跨线程共享：

from configparser import ConfigParser
import mysql.connector
import threading

config_file = 'config.ini'
config = ConfigParser()
config.read(config_file)

# 线程本地存储，每个线程拥有独立的连接和游标实例
thread_local = threading.local()

def get_db_connection():
    if not hasattr(thread_local, 'conn'):
        thread_local.conn = mysql.connector.connect(
            host=config['MYSQL']['DB_HOST'],
            user=config['MYSQL']['DB_USERNAME'],
            passwd=config['MYSQL']['DB_PASSWORD'],
            database=config['MYSQL']['DB_DATABASE']
        )
    return thread_local.conn

def get_db_cursor():
    if not hasattr(thread_local, 'cursor'):
        conn = get_db_connection()
        thread_local.cursor = conn.cursor(buffered=True, dictionary=True)
    return thread_local.cursor

def mysql_update(sorgu):
    try:
        cursor = get_db_cursor()
        cursor.execute(sorgu)
        get_db_connection().commit()
    except Exception as e:
        print(e)
        get_db_connection().rollback()

def mysql_get_all(sorgu):
    try:
        cursor = get_db_cursor()
        cursor.execute(sorgu)
        return cursor.fetchall()
    except Exception as e:
        print(e)
        return []

额外优化：参数化查询防SQL注入 #

当前dolar_euro_guncelle.py使用字符串拼接生成SQL语句，存在SQL注入风险，建议修改为参数化查询：

1. 先修改mysql_db.py的mysql_update函数支持参数：

def mysql_update(sorgu, params=None):
    conn = None
    cursor = None
    try:
        conn = get_db_connection()
        cursor = conn.cursor(buffered=True, dictionary=True)
        cursor.execute(sorgu, params or ())
        conn.commit()
    except Exception as e:
        print(e)
        if conn:
            conn.rollback()
    finally:
        if cursor:
            cursor.close()
        if conn:
            conn.close()

2. 再修改dolar_euro_guncelle.py中的调用逻辑：

# 替换原字符串拼接的SQL调用
mysql_db.mysql_update(
    "update variables set EUR=%s, USD=%s where id=%s",
    (str(EUR), str(USD), '4')
)

内容的提问来源于stack exchange，提问作者kaann.gunerr