mysql db单例模式

import mysql.connector

class MySQLSingleton:
    _instance = None

    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            cls._instance = super(MySQLSingleton, cls).__new__(cls)
        return cls._instance

    def __init__(self, host, user, password, database):
        if not hasattr(self, 'conn'):
            self.conn = mysql.connector.connect(
                host=host,
                user=user,
                password=password,
                database=database
            )

    def get_connection(self):
        return self.conn

# 使用示例
db1 = MySQLSingleton('localhost', 'user', 'password', 'database')
db2 = MySQLSingleton('localhost', 'user', 'password', 'database')

print(db1 is db2)  # 输出: True

import threading

class MySQLSingleton:
    _instance = None
    _lock = threading.Lock()

    def __new__(cls, *args, **kwargs):
        with cls._lock:
            if not cls._instance:
                cls._instance = super(MySQLSingleton, cls).__new__(cls)
        return cls._instance

import time

class MySQLSingleton:
    # ... 省略其他代码 ...

    def check_connection(self):
        try:
            cursor = self.conn.cursor()
            cursor.execute("SELECT 1")
            cursor.close()
        except mysql.connector.Error as err:
            self.conn = mysql.connector.connect(
                host=self.host,
                user=self.user,
                password=self.password,
                database=self.database
            )

    def keep_alive(self, interval=60):
        while True:
            time.sleep(interval)
            self.check_connection()