数据库表实现账号池管理

使用数据库表实现账号池管理是解决“多脚本挤账号”问题的可靠方案，核心是通过数据库的事务和锁机制实现账号的原子性申请与释放，避免并发冲突。以下是详细设计步骤（以 MySQL 为例，其他数据库可类比调整）：

一、核心数据库表设计

需要至少两张表：

1. 账号资源表（account_pool）：存储账号基本信息及实时状态；
2. 账号操作日志表（account_operation_log）：记录账号申请/释放记录，用于问题排查。

1. 账号资源表（account_pool）

字段名	类型	说明
id	INT	自增主键，账号唯一标识
username	VARCHAR(50)	账号用户名（唯一，不可重复）
password	VARCHAR(255)	账号密码（必须加密存储，如AES加密，避免明文泄露）
status	ENUM	账号状态：'IDLE'（空闲）、'USING'（使用中）、'BANNED'（封禁）
last_used_time	DATETIME	最近一次被使用的时间（用于检测“僵尸占用”）
script_id	VARCHAR(100)	当前使用该账号的脚本标识（如脚本进程ID、任务ID，便于追踪）
proxy_ip	VARCHAR(50)	绑定的代理IP（可选，若需网络隔离）
created_at	DATETIME	账号录入时间
updated_at	DATETIME	账号状态更新时间（自动更新）

创建表SQL（MySQL）：

CREATE TABLE `account_pool` (
  `id` INT PRIMARY KEY AUTO_INCREMENT,
  `username` VARCHAR(50) NOT NULL UNIQUE COMMENT '账号用户名',
  `password` VARCHAR(255) NOT NULL COMMENT '加密后的密码',
  `status` ENUM('IDLE', 'USING', 'BANNED') NOT NULL DEFAULT 'IDLE' COMMENT '账号状态',
  `last_used_time` DATETIME NULL COMMENT '最近使用时间',
  `script_id` VARCHAR(100) NULL COMMENT '当前使用的脚本标识',
  `proxy_ip` VARCHAR(50) NULL COMMENT '绑定代理IP',
  `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  INDEX idx_status (`status`) COMMENT '按状态查询索引'
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT '账号资源池表';

2. 账号操作日志表（account_operation_log）

用于记录账号的申请、释放、异常等操作，便于问题排查（如“账号长期未释放”“频繁申请失败”等）。

字段名	类型	说明
id	INT	自增主键
account_id	INT	关联account_pool.id，操作的账号ID
username	VARCHAR(50)	账号用户名（冗余存储，便于直接查看）
operation_type	ENUM	操作类型：'APPLY'（申请）、'RELEASE'（释放）、'FORCE_RELEASE'（强制释放）
script_id	VARCHAR(100)	操作的脚本标识（如进程ID、任务ID）
operation_time	DATETIME	操作时间
ip_address	VARCHAR(50)	脚本执行的IP地址（便于定位问题脚本）
status	ENUM	操作结果：'SUCCESS'（成功）、'FAIL'（失败）
remark	TEXT	备注（如失败原因：“账号已被占用”“网络超时”等）

创建表SQL（MySQL）：

CREATE TABLE `account_operation_log` (
  `id` INT PRIMARY KEY AUTO_INCREMENT,
  `account_id` INT NOT NULL COMMENT '账号ID',
  `username` VARCHAR(50) NOT NULL COMMENT '账号用户名',
  `operation_type` ENUM('APPLY', 'RELEASE', 'FORCE_RELEASE') NOT NULL COMMENT '操作类型',
  `script_id` VARCHAR(100) NOT NULL COMMENT '脚本标识',
  `operation_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '操作时间',
  `ip_address` VARCHAR(50) NULL COMMENT '执行IP',
  `status` ENUM('SUCCESS', 'FAIL') NOT NULL COMMENT '操作结果',
  `remark` TEXT NULL COMMENT '备注',
  INDEX idx_account_id (`account_id`),
  INDEX idx_operation_time (`operation_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT '账号操作日志表';

二、核心逻辑实现（以Python为例）

1. 账号申请流程（原子操作，避免并发冲突）

脚本启动时，从account_pool中申请一个状态为IDLE 的账号，并将其状态改为USING。关键是通过数据库事务+行锁确保操作原子性，避免多个脚本同时申请到同一个账号。

步骤：

1. 开启数据库事务；
2. 查询status='IDLE'的账号（按last_used_time升序，优先使用最久未用的账号，避免资源闲置）；
3. 对查询到的账号加行锁（FOR UPDATE），防止其他事务修改；
4. 更新账号状态为USING，记录script_id和last_used_time；
5. 提交事务，返回账号信息；
6. 若申请失败（如无空闲账号），记录日志并等待重试。

Python代码示例（使用pymysql库）：

import pymysql
from datetime import datetime
import uuid  # 用于生成唯一script_id


def apply_account(db_config, script_id=None):
    """
    从数据库申请空闲账号
    :param db_config: 数据库连接配置（host, user, password, db）
    :param script_id: 脚本标识（可选，默认生成UUID）
    :return: 账号信息（dict）或None（无空闲账号）
    """
    script_id = script_id or str(uuid.uuid4())[:10]  # 生成简短唯一标识
    conn = None
    cursor = None
    try:
        # 1. 连接数据库
        conn = pymysql.connect(**db_config, autocommit=False)  # 关闭自动提交，手动控制事务
        cursor = conn.cursor(pymysql.cursors.DictCursor)

        # 2. 开启事务，查询并锁定空闲账号（原子操作）
        # 按last_used_time升序，优先使用最久未用的账号；加FOR UPDATE锁定行
        query_sql = """
            SELECT id, username, password, proxy_ip 
            FROM account_pool 
            WHERE status = 'IDLE' 
            ORDER BY last_used_time ASC 
            LIMIT 1 
            FOR UPDATE SKIP LOCKED  # MySQL 8.0+支持SKIP LOCKED，跳过已被其他事务锁定的行
        """
        cursor.execute(query_sql)
        account = cursor.fetchone()
        if not account:
            # 无空闲账号，记录日志
            log_operation(
                db_config, account_id=None, username=None, 
                operation_type='APPLY', script_id=script_id, 
                status='FAIL', remark='No idle accounts available'
            )
            return None

        # 3. 更新账号状态为USING
        update_sql = """
            UPDATE account_pool 
            SET status = 'USING', 
                script_id = %s, 
                last_used_time = %s 
            WHERE id = %s
        """
        cursor.execute(update_sql, (script_id, datetime.now(), account['id']))
        conn.commit()  # 提交事务，完成锁定

        # 4. 记录申请成功日志
        log_operation(
            db_config, account_id=account['id'], username=account['username'],
            operation_type='APPLY', script_id=script_id,
            status='SUCCESS', remark='Account applied successfully'
        )
        return account  # 返回账号信息（包含username, password等）

    except Exception as e:
        if conn:
            conn.rollback()  # 异常时回滚事务
        log_operation(
            db_config, account_id=account['id'] if 'account' in locals() else None,
            username=account['username'] if 'account' in locals() else None,
            operation_type='APPLY', script_id=script_id,
            status='FAIL', remark=f'Apply failed: {str(e)}'
        )
        raise e
    finally:
        if cursor:
            cursor.close()
        if conn:
            conn.close()

2. 账号释放流程

脚本执行完毕（或异常退出）时，将账号状态从USING改回IDLE，供其他脚本使用。

步骤：

1. 开启事务；
2. 更新账号状态为IDLE，清空script_id；
3. 提交事务，记录释放日志。

Python代码示例：

def release_account(db_config, account_id, script_id):
    """释放账号（状态改回IDLE）"""
    conn = None
    cursor = None
    try:
        conn = pymysql.connect(**db_config, autocommit=False)
        cursor = conn.cursor(pymysql.cursors.DictCursor)

        # 查询账号当前状态（确认是否被当前脚本占用）
        cursor.execute("SELECT username, status FROM account_pool WHERE id = %s", (account_id,))
        account = cursor.fetchone()
        if not account:
            log_operation(db_config, account_id, None, 'RELEASE', script_id, 'FAIL', 'Account not found')
            return False
        if account['status'] != 'USING':
            log_operation(db_config, account_id, account['username'], 'RELEASE', script_id, 'FAIL', 'Account not in USING status')
            return False

        # 更新账号状态为IDLE
        update_sql = """
            UPDATE account_pool 
            SET status = 'IDLE', 
                script_id = NULL 
            WHERE id = %s
        """
        cursor.execute(update_sql, (account_id,))
        conn.commit()

        # 记录释放日志
        log_operation(
            db_config, account_id=account_id, username=account['username'],
            operation_type='RELEASE', script_id=script_id,
            status='SUCCESS', remark='Account released successfully'
        )
        return True

    except Exception as e:
        if conn:
            conn.rollback()
        log_operation(db_config, account_id, account['username'] if account else None, 'RELEASE', script_id, 'FAIL', f'Release failed: {str(e)}')
        raise e
    finally:
        if cursor:
            cursor.close()
        if conn:
            conn.close()

3. 账号操作日志记录函数

统一记录账号申请/释放的日志到account_operation_log表：

def log_operation(db_config, account_id, username, operation_type, script_id, status, remark):
    """记录账号操作日志"""
    conn = None
    cursor = None
    try:
        conn = pymysql.connect(**db_config)
        cursor = conn.cursor()
        sql = """
            INSERT INTO account_operation_log 
            (account_id, username, operation_type, script_id, ip_address, status, remark)
            VALUES (%s, %s, %s, %s, %s, %s, %s)
        """
        # 获取本地IP（用于记录脚本执行IP）
        ip_address = pymysql._auth.get_local_ip()  # 或使用其他方式获取IP
        cursor.execute(sql, (
            account_id, username, operation_type, script_id,
            ip_address, status, remark[:500]  # 限制remark长度，避免过长
        ))
        conn.commit()
    except Exception as e:
        print(f"Log failed: {e}")  # 日志记录失败不影响主流程
    finally:
        if cursor:
            cursor.close()
        if conn:
            conn.close()

三、异常处理：僵尸账号自动释放（定时任务）

若脚本崩溃或异常退出（如被kill），可能导致账号一直处于USING状态（僵尸账号），无法被其他脚本使用。需通过定时任务检测并释放这类账号。

实现：
创建一个定时任务（如用crontab或Celery Beat），每隔5分钟执行一次以下SQL，释放“USING状态且last_used_time超过30分钟”的账号（时间阈值根据脚本平均执行时长调整）：

-- 释放僵尸账号（状态改回IDLE，记录强制释放日志）
UPDATE account_pool 
SET status = 'IDLE', 
    script_id = NULL,
    last_used_time = NOW()  -- 更新释放时间，便于后续跟踪
WHERE status = 'USING' 
  AND last_used_time < DATE_SUB(NOW(), INTERVAL 30 MINUTE);  -- 超过30分钟未更新视为僵尸账号

同时，在account_operation_log中记录operation_type='FORCE_RELEASE'的日志，便于排查脚本异常退出的原因。

四、密码加密存储（关键！）

数据库中禁止明文存储密码，需加密存储。推荐使用AES对称加密（需妥善保管密钥，如用环境变量或密钥管理服务）。

Python加密/解密示例（使用cryptography库）：

from cryptography.fernet import Fernet

# 生成密钥（仅首次运行，密钥需保存，丢失则无法解密）
# key = Fernet.generate_key()  # 如 b'abcdefghijklmnopqrstuvwxyz123456'
# 实际使用时从环境变量获取密钥
key = bytes(os.getenv('ACCOUNT_ENCRYPT_KEY'), 'utf-8')
cipher_suite = Fernet(key)

def encrypt_password(plaintext_password):
    """加密密码"""
    return cipher_suite.encrypt(plaintext_password.encode()).decode()

def decrypt_password(encrypted_password):
    """解密密码"""
    return cipher_suite.decrypt(encrypted_password.encode()).decode()

# 存储到数据库时：
encrypted_pwd = encrypt_password("user123456")  # 加密后存储
# 脚本使用时：
decrypted_pwd = decrypt_password(account['password'])  # 从数据库读取后解密使用

五、测试验证步骤

1. 初始化账号：向account_pool插入21个账号，密码用encrypt_password加密，状态设为IDLE；
2. 模拟多脚本并发申请：启动5个脚本同时调用apply_account，观察是否每个脚本申请到不同账号（通过account_operation_log日志确认）；
3. 模拟脚本崩溃：手动kill一个正在运行的脚本，等待定时任务执行后，检查该脚本占用的账号是否被释放为IDLE；
4. 无空闲账号场景：启动22个脚本（超过账号总数21），确认第22个脚本申请失败并记录“无空闲账号”日志。

总结

通过数据库表实现账号池的核心是：用account_pool管理账号状态，通过事务+行锁确保原子申请/释放，用定时任务处理异常释放，用日志表追踪操作。该方案可靠性高，支持多脚本分布式部署，且便于监控和问题排查。需注意密码加密和数据库权限控制（仅允许脚本服务器访问账号表），确保账号安全。