影刀RPA进阶教程：数据库高级应用——连接池、事务管理与批量性能优化

系列第63篇已经讲过SQLite和MySQL的基础读写操作——建表、插入、查询、批量写入。那篇解决的是“怎么连数据库”，今天这篇解决的是“怎么用好数据库”——

• 多个流程同时读写数据库，怎么避免连接冲突？

• 批量插入10万条数据，怎么从5分钟优化到5秒？
• 数据操作中途失败，怎么保证数据一致性（事务）？
• 不同环境（开发/测试/生产）怎么自动切换数据库？

这些都是在生产环境中必然遇到的问题。今天把数据库的高级应用方案讲清楚。

---

先说核心：三个进阶能力

能力	解决的问题	技术方案
连接池	频繁创建/关闭连接耗时	复用连接，减少开销
事务管理	多步操作要么全成功要么全失败	BEGIN/COMMIT/ROLLBACK

| 批量优化 | 大量数据写入慢 | 批量SQL + 分批提交 |

---

第一部分：连接池管理

1.1 为什么需要连接池？

# ❌ 不好的做法：每次操作都创建新连接

![picture.image](https://p3-volc-community-sign.byteimg.com/tos-cn-i-tlddhu82om/9b2f0535923d4fbd92360e83dbab40c5~tplv-tlddhu82om-image.image?=&rk3s=8031ce6d&x-expires=1783545221&x-signature=%2BeL0bJVRGkiCbQfDvUrU4k1F%2Feo%3D)
def 插入一条数据(数据):
    conn = sqlite3.connect('database.db')  # 每次打开
    cursor = conn.cursor()
    cursor.execute('INSERT INTO ...', 数据)
    conn.commit()
    
![picture.image](https://p3-volc-community-sign.byteimg.com/tos-cn-i-tlddhu82om/a4f8e7cb20384546b5ad09a500c3e1d4~tplv-tlddhu82om-image.image?=&rk3s=8031ce6d&x-expires=1783545221&x-signature=g%2B6iUV%2BB5PDOx4x%2B8iGL3gwkO2s%3D)
    conn.close()  # 每次关闭
# 插入1000条 → 打开/关闭1000次 → 极慢

![picture.image](https://p3-volc-community-sign.byteimg.com/tos-cn-i-tlddhu82om/591b552429e44575aff9f5327f2691a1~tplv-tlddhu82om-image.image?=&rk3s=8031ce6d&x-expires=1783545221&x-signature=UYqqglTJapNMd1YM25%2Fi%2FM5Ssb8%3D)
# ✅ 好的做法：复用连接
# 在流程开始时创建连接，用完再关闭
conn = sqlite3.connect('database.db')
# 执行所有操作...
conn.close()

1.2 SQLite连接池实现

SQLite是单写多读的，多个流程同时写会报错。用连接池管理并发访问：

# ============================================
# SQLite连接池（Python）
# ============================================
import sqlite3
import threading
from queue import Queue

class SQLiteConnectionPool:
    def __init__(self, db_path, max_connections=5):
        self.db_path = db_path
        self.max_connections = max_connections
        self._pool = Queue()
        self._lock = threading.Lock()
        self._created_count = 0
        
![picture.image](https://p3-volc-community-sign.byteimg.com/tos-cn-i-tlddhu82om/232d604f8c714c83a766a2eef07a91ef~tplv-tlddhu82om-image.image?=&rk3s=8031ce6d&x-expires=1783545221&x-signature=PFGsKdwaRz0jHnhPYy%2FzzAAenvY%3D)
        
        # 预创建连接
        for _ in range(max_connections):
            self._create_connection()
    
    def _create_connection(self):
        conn = sqlite3.connect(self.db_path, check_same_thread=False)
        conn.row_factory = sqlite3.Row
        self._pool.put(conn)
        self._created_count += 1
    
    def get_connection(self, timeout=10):
        """获取连接（带超时）"""
        try:
            return self._pool.get(timeout=timeout)
        except:
            # 池中无可用连接，创建新的
            self._create_connection()
            return self._pool.get()
    
    def return_connection(self, conn):
        """归还连接"""
        if conn:
            self._pool.put(conn)
    
    def close_all(self):
        """关闭所有连接"""
        
![picture.image](https://p3-volc-community-sign.byteimg.com/tos-cn-i-tlddhu82om/31a3895c33724ef9b79ccf5f9b8b982b~tplv-tlddhu82om-image.image?=&rk3s=8031ce6d&x-expires=1783545221&x-signature=uMQfUjZmjXQXfd5L1MFhDTfKQ5s%3D)
        while not self._pool.empty():
            conn = self._pool.get()
            conn.close()

# 使用示例
pool = SQLiteConnectionPool('C:/data/app.db', max_connections=5)

def 并发写入数据(数据列表):
    conn = pool.get_connection()
    try:
        cursor = conn.cursor()
        cursor.executemany('INSERT INTO products VALUES (?,?,?)', 数据列表)
        conn.commit()
    finally:
        pool.return_connection(conn)

# 多个流程可以同时调用，共享连接池

1.3 MySQL连接池（使用DBUtils）

# ============================================
# MySQL连接池（使用DBUtils）
# ============================================
from DBUtils.PooledDB import PooledDB
import pymysql

# 安装：pip install DBUtils pymysql

# 创建连接池
pool = PooledDB(
    creator=pymysql,
    maxconnections=10,
    mincached=2,
    maxcached=5,
    blocking=True,
    host='localhost',
    port=3306,
    user='rpa_user',
    password='rpa_password',
    database='rpa_data',
    charset='utf8mb4'
)

def 执行查询(SQL语句, 参数=()):
    """从连接池获取连接执行查询"""
    conn = pool.connection()
    try:
        cursor = conn.cursor()
        cursor.execute(SQL语句, 参数)
        return cursor.fetchall()
    finally:
        conn.close()  # 归还到连接池

def 执行写入(SQL语句, 参数列表):
    """批量写入"""
    conn = pool.connection()
    try:
        cursor = conn.cursor()
        cursor.executemany(SQL语句, 参数列表)
        conn.commit()
        return cursor.rowcount
    finally:
        conn.close()

---

第二部分：事务管理

2.1 事务的核心概念

事务确保一组操作要么全部成功，要么全部失败。适用于多表关联更新、数据一致性要求高的场景。

# ============================================
# 事务管理模板
# ============================================
def 事务操作():
    conn = get_connection()
    try:
        # 开始事务（默认自动开始）
        cursor = conn.cursor()
        
        # 操作1：扣减库存
        cursor.execute('UPDATE products SET stock = stock - 1 WHERE id = ?', (product_id,))
        
        # 操作2：创建订单
        cursor.execute('INSERT INTO orders (product_id, quantity) VALUES (?,?)', (product_id, 1))
        
        # 操作3：更新销量统计
        cursor.execute('UPDATE stats SET sales = sales + 1 WHERE product_id = ?', (product_id,))
        
        # 全部成功 → 提交
        conn.commit()
        return {"status": "success"}
        
    except Exception as e:
        # 任何步骤失败 → 全部回滚
        conn.rollback()
        return {"status": "failed", "error": str(e)}
    finally:
        conn.close()

2.2 影刀中的事务处理

# ============================================
# 影刀调用带事务的数据库操作
# ============================================

# 在影刀流程中，调用Python执行事务
执行Python代码(
    代码="""
import pymysql

def 转账(from_account, to_account, amount):
    conn = pymysql.connect(host='localhost', user='root', password='xxx', database='bank')
    try:
        cursor = conn.cursor()
        
        # 检查余额
        cursor.execute('SELECT balance FROM accounts WHERE id=%s', (from_account,))
        balance = cursor.fetchone()[0]
        if balance < amount:
            return {'status': 'failed', 'message': '余额不足'}
        
        # 扣款
        cursor.execute('UPDATE accounts SET balance = balance - %s WHERE id=%s', (amount, from_account))
        
        # 加款
        cursor.execute('UPDATE accounts SET balance = balance + %s WHERE id=%s', (amount, to_account))
        
        # 记录转账日志
        cursor.execute('INSERT INTO transfer_log (from_id, to_id, amount, time) VALUES (%s,%s,%s,NOW())',
                       (from_account, to_account, amount))
        
        conn.commit()
        return {'status': 'success', 'message': '转账成功'}
        
    except Exception as e:
        conn.rollback()
        return {'status': 'failed', 'message': str(e)}
    finally:
        conn.close()

result = 转账('A001', 'B002', 1000)
"""
)

解析JSON(转账结果, 存入变量=转账结果)
输出日志("转账结果：" + 转账结果["status"])

2.3 保存点（Savepoint）——部分回滚

当一个事务中有多个独立步骤，只想回滚其中某一步时，使用保存点。

# ============================================
# 保存点：部分回滚
# ============================================
def 批量处理订单(订单列表):
    conn = get_connection()
    cursor = conn.cursor()
    
    try:
        # 创建保存点
        cursor.execute('SAVEPOINT batch_start')
        
        for idx, order in enumerate(订单列表):
            try:
                # 处理单个订单
                cursor.execute('INSERT INTO orders VALUES (?,?,?)', order)
            except Exception as e:
                # 单个订单失败 → 回滚到保存点（不影响已处理的其他订单）
                cursor.execute(f'ROLLBACK TO SAVEPOINT batch_start')
                print(f"订单{idx}失败，已回滚：{e}")
                continue
        
        conn.commit()
        return {"status": "partial_success", "processed": len(订单列表)}
        
    except Exception as e:
        conn.rollback()  # 致命错误，全部回滚
        return {"status": "failed", "error": str(e)}
    finally:
        conn.close()

---

第三部分：批量性能优化

3.1 批量插入速度对比

方式	1万条耗时	10万条耗时
逐条插入	~8秒	~80秒
批量插入（executemany）	~0.5秒	~4秒
批量插入 + 分批提交	~0.5秒	~3秒

3.2 批量插入最佳实践

# ============================================
# 高性能批量插入（MySQL）
# ============================================
import pymysql

def 高性能批量插入(数据列表, 批次大小=1000):
    """分批批量插入，避免单次过大"""
    conn = pymysql.connect(host='localhost', user='root', password='xxx', database='rpa_data')
    cursor = conn.cursor()
    
    total = len(数据列表)
    inserted = 0
    
    for i in range(0, total, 批次大小):
        batch = 数据列表[i:i+批次大小]
        
        # 使用executemany批量插入
        cursor.executemany('''
            INSERT INTO products (product_id, name, price, sales, collect_time)
            VALUES (%s, %s, %s, %s, %s)
        ''', batch)
        
        conn.commit()  # 分批提交
        inserted += len(batch)
        print(f"已插入 {inserted}/{total}")
    
    conn.close()
    return inserted

# 使用
结果 = 高性能批量插入(采集数据, 批次大小=2000)
输出日志(f"批量插入完成，共{结果}条")

3.3 SQLite批量插入优化

# ============================================
# SQLite批量插入性能优化
# ============================================
import sqlite3

def 优化SQLite批量插入(数据列表):
    conn = sqlite3.connect('C:/data/products.db')
    cursor = conn.cursor()
    
    # 关键优化1：关闭事务自动提交
    conn.isolation_level = None
    cursor.execute('BEGIN TRANSACTION')
    
    # 关键优化2：使用executemany
    cursor.executemany('''
        INSERT OR REPLACE INTO products (product_id, name, price, sales, collect_time)
        VALUES (?, ?, ?, ?, ?)
    ''', 数据列表)
    
    # 关键优化3：提交事务
    conn.commit()
    
    # 关键优化4：维护数据库（释放空间）
    cursor.execute('VACUUM')
    
    conn.close()
    return len(数据列表)

---

第四部分：环境隔离与配置管理

4.1 多环境数据库配置

# ============================================
# 环境隔离配置
# ============================================

# 配置表：database_config.xlsx
# | 环境 | 数据库类型 | 主机 | 端口 | 数据库名 | 用户 | 密码 |
# |------|-----------|------|------|---------|------|------|
# | dev | mysql | localhost | 3306 | rpa_dev | dev_user | dev_pass |
# | test | mysql | 192.168.1.10 | 3306 | rpa_test | test_user | test_pass |
# | prod | mysql | 192.168.1.100 | 3306 | rpa_prod | prod_user | prod_pass |

def get_db_config(环境变量):
    """根据环境获取数据库配置"""
    配置读取 = {
        "dev": {
            "host": "localhost",
            "user": "dev_user",
            "password": "dev_pass",
            "database": "rpa_dev"
        },
        "test": {
            "host": "192.168.1.10",
            "user": "test_user",
            "password": "test_pass",
            "database": "rpa_test"
        },
        "prod": {
            "host": "192.168.1.100",
            "user": "prod_user",
            "password": "prod_pass",
            "database": "rpa_prod"
        }
    }
    return 配置读取.get(环境变量, 配置读取["dev"])

4.2 自动创建表结构

# ============================================
# 自动建表（确保表存在）
# ============================================
def 确保表存在(conn):
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS products (
            id INTEGER PRIMARY KEY AUTO_INCREMENT,
            product_id VARCHAR(50) UNIQUE,
            name VARCHAR(255),
            price DECIMAL(10,2),
            sales INT,
            collect_time DATETIME,
            INDEX idx_product_id (product_id),
            INDEX idx_collect_time (collect_time)
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    ''')
    conn.commit()

---

第五部分：数据库健康检查

# ============================================
# 数据库健康检查
# ============================================
def 健康检查():
    """检查数据库连接是否正常"""
    try:
        conn = get_connection()
        cursor = conn.cursor()
        cursor.execute('SELECT 1')
        cursor.fetchone()
        conn.close()
        return {"status": "healthy"}
    except Exception as e:
        return {"status": "unhealthy", "error": str(e)}

def 数据量检查(表名, 天数限制=30):
    """检查最近N天的数据量"""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(f'''
        SELECT COUNT(*) FROM {表名}
        WHERE collect_time > DATE_SUB(NOW(), INTERVAL {天数限制} DAY)
    ''')
    count = cursor.fetchone()[0]
    conn.close()
    return count

# 在监控流程中使用
健康状态 = 健康检查()
if 健康状态["status"] != "healthy":
    调用子流程(发送告警, 输入: "数据库连接异常")

数据量 = 数据量检查("products")
if 数据量 == 0:
    输出日志("【告警】最近30天无数据写入")

---

常见问题速查

问题	原因	解决方法
数据库连接被占用	连接未释放	用连接池，确保finally释放
SQLite写入冲突	多线程同时写	用连接池排队，或用MySQL
事务执行了一半失败	未用事务包裹	用BEGIN/COMMIT/ROLLBACK
批量插入速度慢	逐条提交	用executemany + 分批提交
连接超时	长时间空闲	设置连接超时时间，或定期保活
表不存在	未初始化	流程启动时执行建表语句

---

推荐资源

• DBUtils文档（连接池）
• SQLite事务文档
• MySQL批量插入最佳实践
• 本系列第63篇《SQLite/MySQL数据库读写与批量数据存储》

---

#影刀RPA #RPA自动化 #数据库 #连接池 #事务 #批量优化

作者：林焱

本文为《影刀RPA学习手册》系列文章之一，内容源于实操经验的整理与分享。