MySQL 8.0单用户多登录限制实现方案咨询（Python操作）

方案1：自定义锁表机制（灵活可控） #

由于MySQL原生的用户连接限制（如MAX_USER_CONNECTIONS）只能控制总连接数，无法区分首个会话与后续会话，你可以通过自定义锁表来实现独占逻辑：

• 创建锁表

CREATE TABLE IF NOT EXISTS session_lock (
    lock_id INT PRIMARY KEY DEFAULT 1,
    holder_session_id VARCHAR(255) NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY (lock_id)
);

• 会话启动时抢占锁
  在Python会话初始化阶段，生成唯一会话ID并尝试抢占锁，同时加入超时机制避免异常锁死：

import mysql.connector
import uuid
from datetime import datetime, timedelta

# 初始化数据库连接
db = mysql.connector.connect(
    host="你的主机地址",
    user="你的MySQL账号",
    password="你的密码",
    database="目标数据库"
)
cursor = db.cursor()
session_id = uuid.uuid4().hex
has_write_perm = False

# 超时阈值：超过5分钟的锁视为失效，允许新会话抢占
expiry_threshold = (datetime.now() - timedelta(minutes=5)).strftime('%Y-%m-%d %H:%M:%S')

try:
    # 尝试插入或更新锁记录
    cursor.execute("""
        INSERT INTO session_lock (lock_id, holder_session_id, created_at)
        VALUES (1, %s, NOW())
        ON DUPLICATE KEY UPDATE
            holder_session_id = IF(created_at < %s, %s, holder_session_id),
            created_at = IF(created_at < %s, NOW(), created_at)
    """, (session_id, expiry_threshold, session_id, expiry_threshold))
    db.commit()

    # 校验是否成功获取锁
    cursor.execute("SELECT holder_session_id FROM session_lock WHERE lock_id = 1")
    current_holder = cursor.fetchone()[0]
    if current_holder == session_id:
        has_write_perm = True
        print("已获取独占写入权限")
    else:
        print("已有活跃会话，仅拥有读取权限")
except mysql.connector.Error as e:
    print(f"锁操作失败: {str(e)}")

# 后续写入操作前检查权限
def execute_write(sql, params=None):
    if not has_write_perm:
        raise PermissionError("无写入权限，已有活跃会话")
    cursor.execute(sql, params)
    db.commit()

• 会话退出时释放锁
  用finally块确保锁能被释放，即使会话异常退出：

finally:
    if has_write_perm:
        cursor.execute("DELETE FROM session_lock WHERE lock_id = 1 AND holder_session_id = %s", (session_id,))
        db.commit()
    cursor.close()
    db.close()

方案2：利用MySQL内置GET_LOCK()函数（轻量便捷） #

MySQL提供的会话级锁函数，无需额外建表，异常断开会自动释放锁：

• 获取锁

cursor.execute("SELECT GET_LOCK('exclusive_write_lock', 0)")
lock_acquired = cursor.fetchone()[0] == 1
has_write_perm = lock_acquired
if lock_acquired:
    print("成功获取写入锁")
else:
    print("无法获取锁，仅支持读取操作")

第二个参数0表示立即返回，不等待锁释放。

• 释放锁

finally:
    if has_write_perm:
        cursor.execute("SELECT RELEASE_LOCK('exclusive_write_lock')")
    cursor.close()
    db.close()

关键注意事项 #

• 两种方案都要确保写入操作前校验has_write_perm状态，避免越权写入。
• 自定义锁表的超时时间可根据业务调整，防止异常导致锁长期占用。
• 若使用GET_LOCK()，注意锁是全局的，其他使用同一账号的会话都会受此锁限制，符合你的需求。

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