PHP小码农

Appearance

SQLite3 事务处理 ​

概述 ​

事务是数据库操作的逻辑单元，确保一组操作要么全部成功提交，要么全部失败回滚。SQLite 完整支持 ACID 事务特性，结合 WAL 模式可以提供良好的并发读写性能。

事务核心原则

  • 原子性（Atomicity） — 事务中的操作要么全部执行，要么全部不执行
  • 一致性（Consistency） — 事务前后数据库状态一致
  • 隔离性（Isolation） — 并发事务互不干扰
  • 持久性（Durability） — 提交后的数据永久保存

基础概念 ​

SQLite 事务模式 ​

SQLite 支持不同级别的事务锁定：

语句锁类型说明
BEGIN DEFERRED不获取锁（默认）首次读写时才获取锁
BEGIN IMMEDIATERESERVED 锁阻止其他写入者，允许其他读取者
BEGIN EXCLUSIVEEXCLUSIVE 锁独占访问，阻止所有其他操作

隔离级别 ​

SQLite 使用串行化（Serializable）隔离级别 — 这是最高级别的隔离，确保事务之间完全独立。但 SQLite 的串行化与 MySQL 的有所不同：

  • SQLite 通过文件级锁实现隔离
  • WAL 模式下读者不会阻塞写者
  • 写者仍然需要排他访问

语法与代码 ​

基本事务操作 ​

php
<?php
declare(strict_types=1);

$db = new SQLite3(__DIR__ . '/app.db');
$db->enableExceptions(true);

// 方式1: 使用 exec()
$db->exec('BEGIN IMMEDIATE');
try {
    $db->exec("UPDATE accounts SET balance = balance - 100 WHERE id = 1");
    $db->exec("UPDATE accounts SET balance = balance + 100 WHERE id = 2");
    $db->exec('COMMIT');
    echo "转账成功\n";
} catch (Exception $e) {
    $db->exec('ROLLBACK');
    echo "转账失败: " . $e->getMessage() . "\n";
}

WAL 模式与并发事务 ​

php
<?php
declare(strict_types=1);

$db = new SQLite3(__DIR__ . '/app.db');
$db->enableExceptions(true);

// 检查并开启 WAL
$currentMode = $db->querySingle('PRAGMA journal_mode');
if ($currentMode !== 'wal') {
    $db->exec('PRAGMA journal_mode = WAL;');
    echo "已切换到 WAL 模式\n";
}

// 设置繁忙超时
$db->busyTimeout(5000);

// 在 WAL 模式下：
// - 多个进程可以同时读取
// - 只有一个进程可以写入
// - 写入不会阻塞读取
// - 读取不会阻塞写入
echo "当前模式: " . $db->querySingle('PRAGMA journal_mode') . "\n";

WAL 模式的工作原理

WAL 模式将新数据写入 WAL 文件而非直接修改数据库文件。读取操作从数据库文件读取，写入操作追加到 WAL 文件。定期将 WAL 文件合并回主数据库（称为 checkpoint）。

保存点（Savepoint） ​

php
<?php
declare(strict_types=1);

$db = new SQLite3(__DIR__ . '/app.db');
$db->enableExceptions(true);

$db->exec('BEGIN IMMEDIATE');

try {
    $db->exec("INSERT INTO orders (user_id, total) VALUES (1, 500)");
    $db->exec("UPDATE accounts SET balance = balance - 500 WHERE id = 1");

    // 创建保存点 — 用于部分回滚
    $db->exec('SAVEPOINT order_created');

    try {
        $db->exec("INSERT INTO shipments (order_id, address) VALUES (1, '北京')");
        $db->exec("INSERT INTO notifications (user_id, message) VALUES (1, '已发货')");
    } catch (Exception $e) {
        // 回滚到保存点，保留之前的操作
        $db->exec('ROLLBACK TO order_created');
        echo "发货步骤失败，但订单已创建\n";
    }

    $db->exec('RELEASE order_created');
    $db->exec('COMMIT');
    echo "事务提交成功\n";

} catch (Exception $e) {
    $db->exec('ROLLBACK');
    echo "事务完全回滚: " . $e->getMessage() . "\n";
}

用 SAVEPOINT 模拟嵌套事务 ​

php
<?php
declare(strict_types=1);

function transferMoney(SQLite3 $db, int $from, int $to, float $amount): void
{
    $db->exec('SAVEPOINT transfer');

    try {
        $balance = $db->querySingle(
            "SELECT balance FROM accounts WHERE id = {$from}"
        );

        if ($balance < $amount) {
            throw new RuntimeException('余额不足');
        }

        $db->exec("UPDATE accounts SET balance = balance - {$amount} WHERE id = {$from}");
        $db->exec("UPDATE accounts SET balance = balance + {$amount} WHERE id = {$to}");

        $db->exec('RELEASE transfer');
    } catch (Exception $e) {
        $db->exec('ROLLBACK TO transfer');
        $db->exec('RELEASE transfer');
        throw $e;
    }
}

// 外层事务
$db->exec('BEGIN IMMEDIATE');
try {
    transferMoney($db, 1, 2, 100);
    transferMoney($db, 2, 3, 50);
    $db->exec('COMMIT');
} catch (Exception $e) {
    $db->exec('ROLLBACK');
    echo "批量转账失败: " . $e->getMessage() . "\n";
}

实战示例 ​

事务管理器类 ​

php
<?php
declare(strict_types=1);

class TransactionManager
{
    private SQLite3 $db;
    private int $transactionLevel = 0;

    public function __construct(SQLite3 $db)
    {
        $this->db = $db;
    }

    public function begin(): void
    {
        if ($this->transactionLevel === 0) {
            $this->db->exec('BEGIN IMMEDIATE');
        } else {
            $this->db->exec('SAVEPOINT sp_' . $this->transactionLevel);
        }
        $this->transactionLevel++;
    }

    public function commit(): void
    {
        $this->transactionLevel--;

        if ($this->transactionLevel === 0) {
            $this->db->exec('COMMIT');
        } else {
            $this->db->exec('RELEASE sp_' . $this->transactionLevel);
        }
    }

    public function rollback(): void
    {
        $this->transactionLevel--;

        if ($this->transactionLevel === 0) {
            $this->db->exec('ROLLBACK');
        } else {
            $this->db->exec('ROLLBACK TO sp_' . $this->transactionLevel);
            $this->db->exec('RELEASE sp_' . $this->transactionLevel);
        }
    }

    /**
     * 在事务中执行闭包
     */
    public function transaction(callable $callback): mixed
    {
        $this->begin();
        try {
            $result = $callback($this->db);
            $this->commit();
            return $result;
        } catch (Exception $e) {
            $this->rollback();
            throw $e;
        }
    }

    public function getTransactionLevel(): int
    {
        return $this->transactionLevel;
    }
}

电商订单处理示例 ​

php
<?php
declare(strict_types=1);

class OrderService
{
    private SQLite3 $db;
    private TransactionManager $txManager;

    public function __construct(SQLite3 $db)
    {
        $this->db = $db;
        $this->db->enableExceptions(true);
        $this->db->busyTimeout(5000);
        $this->txManager = new TransactionManager($db);
    }

    /**
     * 创建订单 — 包含库存扣减、余额扣减、记录创建
     */
    public function createOrder(int $userId, array $items): int
    {
        return $this->txManager->transaction(function () use ($userId, $items) {
            // 1. 计算总金额
            $totalAmount = 0;
            foreach ($items as $item) {
                $price = $this->db->querySingle(
                    "SELECT price FROM products WHERE id = {$item['product_id']}"
                );
                $totalAmount += $price * $item['quantity'];
            }

            // 2. 检查并扣减库存
            foreach ($items as $item) {
                $stock = $this->db->querySingle(
                    "SELECT stock FROM products WHERE id = {$item['product_id']}"
                );
                if ($stock < $item['quantity']) {
                    throw new RuntimeException("产品 {$item['product_id']} 库存不足");
                }
                $this->db->exec(
                    "UPDATE products SET stock = stock - {$item['quantity']} WHERE id = {$item['product_id']}"
                );
            }

            // 3. 扣减用户余额
            $balance = $this->db->querySingle(
                "SELECT balance FROM accounts WHERE user_id = {$userId}"
            );
            if ($balance < $totalAmount) {
                throw new RuntimeException("余额不足");
            }
            $this->db->exec(
                "UPDATE accounts SET balance = balance - {$totalAmount} WHERE user_id = {$userId}"
            );

            // 4. 创建订单
            $this->db->exec("
                INSERT INTO orders (user_id, total_amount, status)
                VALUES ({$userId}, {$totalAmount}, 'pending')
            ");
            $orderId = $this->db->lastInsertRowID();

            // 5. 创建订单明细
            foreach ($items as $item) {
                $this->db->exec("
                    INSERT INTO order_items (order_id, product_id, quantity, price)
                    VALUES ({$orderId}, {$item['product_id']}, {$item['quantity']}, {$price})
                ");
            }

            return $orderId;
        });
    }

    /**
     * 取消订单 — 恢复库存和余额
     */
    public function cancelOrder(int $orderId): void
    {
        $this->txManager->transaction(function () use ($orderId) {
            $order = $this->db->querySingle(
                "SELECT user_id, total_amount, status FROM orders WHERE id = {$orderId}",
                true
            );

            if (!$order || $order['status'] === 'cancelled' || $order['status'] === 'shipped') {
                throw new RuntimeException("无法取消该订单");
            }

            // 恢复余额
            $this->db->exec("
                UPDATE accounts SET balance = balance + {$order['total_amount']}
                WHERE user_id = {$order['user_id']}
            ");

            // 恢复库存
            $items = $this->db->query("
                SELECT product_id, quantity FROM order_items WHERE order_id = {$orderId}
            ");
            while ($item = $items->fetchArray(SQLITE3_ASSOC)) {
                $this->db->exec("
                    UPDATE products SET stock = stock + {$item['quantity']}
                    WHERE id = {$item['product_id']}
                ");
            }

            $this->db->exec("UPDATE orders SET status = 'cancelled' WHERE id = {$orderId}");
        });
    }
}

批量插入优化 ​

php
<?php
declare(strict_types=1);

class BulkInserter
{
    private SQLite3 $db;

    public function __construct(SQLite3 $db)
    {
        $this->db = $db;
        $this->db->enableExceptions(true);
    }

    /**
     * 高性能批量插入 — 使用单事务 + 逐条预处理
     */
    public function batchInsert(string $table, array $rows): int
    {
        if (empty($rows)) {
            return 0;
        }

        $columns = array_keys($rows[0]);
        $placeholders = implode(', ', array_fill(0, count($columns), '?'));
        $sql = sprintf(
            'INSERT INTO %s (%s) VALUES (%s)',
            $table,
            implode(', ', $columns),
            $placeholders
        );

        $this->db->exec('BEGIN IMMEDIATE');
        try {
            $stmt = $this->db->prepare($sql);
            $count = 0;
            foreach ($rows as $row) {
                foreach (array_values($row) as $i => $value) {
                    $stmt->bindValue($i + 1, $value);
                }
                $stmt->execute();
                $stmt->reset();
                $count++;
            }
            $this->db->exec('COMMIT');
            return $count;
        } catch (Exception $e) {
            $this->db->exec('ROLLBACK');
            throw $e;
        }
    }
}

注意事项 ​

SQLite 事务限制 ​

重要限制

  1. 写锁粒度 — SQLite 使用数据库文件级锁，一次只能有一个写者
  2. 嵌套事务 — 不支持真正的嵌套事务，需使用 SAVEPOINT 模拟
  3. DDL 与事务 — SQLite 的 DDL 语句可以参与事务
  4. 内存使用 — 大事务会占用大量临时文件空间

死锁预防 ​

php
<?php
// 预防措施:
// 1. 设置合理的 busy_timeout
$db->busyTimeout(5000);

// 2. 保持事务简短
$db->exec('BEGIN IMMEDIATE');
try {
    // 只做必要操作
    $db->exec('COMMIT');
} catch (Exception $e) {
    $db->exec('ROLLBACK');
}

// 3. 按固定顺序访问资源（如果涉及多表）

WAL Checkpoint 管理 ​

php
<?php
// WAL 文件会持续增长，需要定期 checkpoint
// 默认当 WAL 文件达到 1000 页时自动 checkpoint

// 手动触发 checkpoint
$db->exec('PRAGMA wal_checkpoint(TRUNCATE);');

// Checkpoint 模式:
// PASSIVE — 不阻塞，将部分 WAL 数据写入主数据库
// FULL    — 等待所有写入者完成后合并
// RESTART — 等待所有写入者完成后重建 WAL 文件
// TRUNCATE — 截断 WAL 文件为零字节

最佳实践 ​

1. 事务配置模板 ​

php
<?php
function configureSqliteDatabase(SQLite3 $db): void
{
    $db->enableExceptions(true);
    $db->busyTimeout(5000);
    $db->exec('PRAGMA journal_mode = WAL;');
    $db->exec('PRAGMA synchronous = NORMAL;');
    $db->exec('PRAGMA foreign_keys = ON;');
    $db->exec('PRAGMA cache_size = -8000;');
    $db->exec('PRAGMA temp_store = MEMORY;');
    $db->exec('PRAGMA wal_autocheckpoint = 1000;');
}

2. 事务使用原则 ​

php
<?php
// 原则1: 事务要尽可能短
// 原则2: 始终使用异常处理
// 原则3: WAL 模式下用 BEGIN IMMEDIATE 开始写事务
// 原则4: 设置 busy_timeout 避免立即失败
// 原则5: 高频写入考虑使用消息队列缓冲

3. 并发替代方案 ​

php
<?php
// 写入缓冲类 — 减少数据库写入频率
class WriteBuffer
{
    private array $buffer = [];
    private int $threshold;

    public function __construct(int $threshold = 100)
    {
        $this->threshold = $threshold;
    }

    public function push(string $sql, array $params = []): void
    {
        $this->buffer[] = [$sql, $params];

        if (count($this->buffer) >= $this->threshold) {
            // 触发 flush — 需要在外部调用 flush()
            $this->buffer = []; // 简化处理
        }
    }

    public function flush(SQLite3 $db): void
    {
        if (empty($this->buffer)) {
            return;
        }

        $db->exec('BEGIN IMMEDIATE');
        try {
            foreach ($this->buffer as [$sql, $params]) {
                $stmt = $db->prepare($sql);
                foreach ($params as $key => $value) {
                    $stmt->bindValue($key, $value);
                }
                $stmt->execute();
            }
            $db->exec('COMMIT');
            $this->buffer = [];
        } catch (Exception $e) {
            $db->exec('ROLLBACK');
            throw $e;
        }
    }
}

参考链接 ​