PostgreSQL 事务

概述

事务是作为单个工作单元执行的一个或多个 SQL 操作的序列。事务确保数据库操作的数据完整性和一致性。

ACID 特性

PostgreSQL 中的事务遵循 ACID 原则：

1. 原子性 (Atomicity) - 所有操作成功或全部失败
2. 一致性 (Consistency) - 数据库保持有效状态
3. 隔离性 (Isolation) - 并发事务不互相干扰
4. 持久性 (Durability) - 已提交的更改持久保存

事务基础

开始事务

-- 开始事务
BEGIN;

-- 或使用 START TRANSACTION
START TRANSACTION;

提交事务

BEGIN;
UPDATE accounts SET balance = balance - 100 WHERE id = 1;
UPDATE accounts SET balance = balance + 100 WHERE id = 2;
COMMIT;  -- 保存所有更改

回滚事务

BEGIN;
UPDATE accounts SET balance = balance - 100 WHERE id = 1;
UPDATE accounts SET balance = balance + 100 WHERE id = 2;
ROLLBACK;  -- 撤销所有更改

完整事务示例

-- 创建示例表
CREATE TABLE accounts (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100),
    balance NUMERIC(10,2)
);

INSERT INTO accounts (name, balance) VALUES
    ('张三', 1000.00),
    ('李四', 500.00);

-- 账户间转账
BEGIN;

-- 从张三扣款
UPDATE accounts 
SET balance = balance - 200 
WHERE name = '张三';

-- 给李四加款
UPDATE accounts 
SET balance = balance + 200 
WHERE name = '李四';

-- 提交事务
COMMIT;

-- 验证转账
SELECT * FROM accounts;

事务隔离级别

PostgreSQL 支持四种隔离级别来控制事务行为：

1. 读未提交 (Read Uncommitted)

BEGIN TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;
-- 在 PostgreSQL 中与 Read Committed 相同
COMMIT;

2. 读已提交 (Read Committed) - 默认

BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED;
-- 只能看到已提交的数据
-- 每个语句看到最新的已提交数据
COMMIT;

3. 可重复读 (Repeatable Read)

BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ;
-- 看到第一个查询时的快照
-- 防止不可重复读
COMMIT;

4. 可串行化 (Serializable)

BEGIN TRANSACTION ISOLATION LEVEL SERIALIZABLE;
-- 最严格的隔离
-- 防止幻读
COMMIT;

隔离级别比较

-- 会话 1: 读已提交
BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED;
SELECT * FROM accounts WHERE name = '张三';
-- 可以看到其他已提交事务的更改
COMMIT;

-- 会话 2: 可重复读
BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ;
SELECT * FROM accounts WHERE name = '张三';
-- 始终看到相同的数据（快照）
SELECT * FROM accounts WHERE name = '张三';
COMMIT;

保存点

保存点允许您部分回滚事务。

BEGIN;

UPDATE accounts SET balance = balance - 100 WHERE id = 1;

SAVEPOINT my_savepoint;

UPDATE accounts SET balance = balance + 100 WHERE id = 2;

-- 哎呀，回滚到保存点
ROLLBACK TO SAVEPOINT my_savepoint;

-- 第一个更新保留，第二个被撤销
COMMIT;

多个保存点

BEGIN;

INSERT INTO accounts (name, balance) VALUES ('王五', 1000);
SAVEPOINT sp1;

UPDATE accounts SET balance = balance + 500 WHERE name = '王五';
SAVEPOINT sp2;

UPDATE accounts SET balance = balance - 200 WHERE name = '王五';
SAVEPOINT sp3;

-- 回滚到 sp2
ROLLBACK TO SAVEPOINT sp2;

-- 释放保存点（删除它）
RELEASE SAVEPOINT sp1;

COMMIT;

事务控制

检查事务状态

-- 检查是否在事务中
SELECT current_setting('transaction_isolation');

-- 显示事务状态
SHOW transaction_isolation;

设置默认隔离级别

-- 对当前会话
SET SESSION CHARACTERISTICS AS TRANSACTION ISOLATION LEVEL SERIALIZABLE;

-- 在 postgresql.conf 中对所有会话
-- default_transaction_isolation = 'repeatable read'

常见事务模式

模式 1：安全转账

CREATE OR REPLACE FUNCTION transfer_money(
    from_account INTEGER,
    to_account INTEGER,
    amount NUMERIC
) RETURNS BOOLEAN AS $$
BEGIN
    -- 检查余额是否充足
    IF (SELECT balance FROM accounts WHERE id = from_account) < amount THEN
        RAISE EXCEPTION '余额不足';
    END IF;
    
    -- 从源账户扣款
    UPDATE accounts 
    SET balance = balance - amount 
    WHERE id = from_account;
    
    -- 给目标账户加款
    UPDATE accounts 
    SET balance = balance + amount 
    WHERE id = to_account;
    
    RETURN TRUE;
EXCEPTION
    WHEN OTHERS THEN
        RAISE NOTICE '事务失败: %', SQLERRM;
        RETURN FALSE;
END;
$$ LANGUAGE plpgsql;

-- 使用函数
SELECT transfer_money(1, 2, 100);

模式 2：带验证的批量插入

BEGIN;

-- 创建临时暂存表
CREATE TEMP TABLE staging_data (
    id INTEGER,
    value TEXT,
    is_valid BOOLEAN DEFAULT FALSE
);

-- 加载数据
INSERT INTO staging_data (id, value) VALUES
    (1, '有效数据'),
    (2, '无效'),
    (3, '有效数据');

-- 验证数据
UPDATE staging_data 
SET is_valid = TRUE 
WHERE LENGTH(value) > 5;

-- 只插入有效数据
INSERT INTO main_table (id, value)
SELECT id, value 
FROM staging_data 
WHERE is_valid = TRUE;

-- 检查是否所有数据都有效
DO $$
BEGIN
    IF EXISTS (SELECT 1 FROM staging_data WHERE NOT is_valid) THEN
        RAISE EXCEPTION '发现无效数据';
    END IF;
END $$;

COMMIT;

模式 3：审计跟踪

CREATE TABLE audit_log (
    id SERIAL PRIMARY KEY,
    table_name VARCHAR(100),
    operation VARCHAR(10),
    old_data JSONB,
    new_data JSONB,
    changed_at TIMESTAMPTZ DEFAULT NOW(),
    changed_by VARCHAR(100)
);

BEGIN;

-- 带审计的更新
WITH old_values AS (
    SELECT to_jsonb(accounts.*) AS data
    FROM accounts
    WHERE id = 1
)
UPDATE accounts 
SET balance = balance + 500
WHERE id = 1
RETURNING (SELECT data FROM old_values) AS old_data, 
          to_jsonb(accounts.*) AS new_data;

-- 记录更改
INSERT INTO audit_log (table_name, operation, old_data, new_data, changed_by)
VALUES ('accounts', 'UPDATE', 
    (SELECT to_jsonb(accounts.*) FROM accounts WHERE id = 1),
    (SELECT to_jsonb(accounts.*) FROM accounts WHERE id = 1),
    CURRENT_USER);

COMMIT;

事务错误和处理

事务中的错误处理

DO $$
BEGIN
    BEGIN
        -- 开始嵌套块
        UPDATE accounts SET balance = balance - 1000 WHERE id = 1;
        
        -- 如果余额为负则失败
        IF (SELECT balance FROM accounts WHERE id = 1) < 0 THEN
            RAISE EXCEPTION '不允许负余额';
        END IF;
        
    EXCEPTION
        WHEN OTHERS THEN
            RAISE NOTICE '发生错误: %', SQLERRM;
            -- 事务被回滚
    END;
END $$;

重试逻辑

CREATE OR REPLACE FUNCTION retry_transaction(max_attempts INTEGER)
RETURNS BOOLEAN AS $$
DECLARE
    attempt INTEGER := 0;
BEGIN
    WHILE attempt < max_attempts LOOP
        BEGIN
            -- 尝试事务
            UPDATE accounts 
            SET balance = balance - 100 
            WHERE id = 1;
            
            -- 如果成功，返回
            RETURN TRUE;
            
        EXCEPTION
            WHEN serialization_failure THEN
                attempt := attempt + 1;
                IF attempt >= max_attempts THEN
                    RAISE EXCEPTION '事务在 % 次尝试后失败', max_attempts;
                END IF;
                -- 重试前等待
                PERFORM pg_sleep(0.1);
        END;
    END LOOP;
    
    RETURN FALSE;
END;
$$ LANGUAGE plpgsql;

并发问题

脏读

-- PostgreSQL 防止脏读
-- 事务 1
BEGIN;
UPDATE accounts SET balance = 999 WHERE id = 1;
-- 尚未提交

-- 事务 2（在另一个会话中）
BEGIN;
SELECT balance FROM accounts WHERE id = 1;
-- 看到旧值（不是 999）
COMMIT;

-- 事务 1
ROLLBACK;  -- 更改被撤销

丢失更新

-- 解决方案：使用 FOR UPDATE
BEGIN;
SELECT * FROM accounts WHERE id = 1 FOR UPDATE;
-- 锁定行
UPDATE accounts SET balance = balance + 100 WHERE id = 1;
COMMIT;

幻读

-- 使用 SERIALIZABLE 防止
BEGIN TRANSACTION ISOLATION LEVEL SERIALIZABLE;
SELECT COUNT(*) FROM accounts WHERE balance > 500;
-- 其他事务无法插入会改变此计数的行
COMMIT;

最佳实践

1. 保持事务简短

   • 最小化持有锁的时间
   • 减少冲突机会

2. 使用适当的隔离级别

   • 大多数情况使用 READ COMMITTED
   • 需要一致性时使用 REPEATABLE READ
   • 严格要求时使用 SERIALIZABLE

3. 正确处理错误

   • 始终使用异常处理
   • 对序列化失败实现重试逻辑

4. 对复杂逻辑使用保存点

   • 部分回滚而不丢失整个事务
   • 更好的错误恢复

5. 避免长时间运行的事务

   • 可能导致阻塞
   • 增加死锁风险
   • 影响 vacuum 性能

事务性能提示

-- 1. 批量操作
BEGIN;
INSERT INTO logs (message) VALUES ('msg1'), ('msg2'), ('msg3');
COMMIT;

-- 2. 对批量加载使用 COPY
BEGIN;
COPY data FROM '/path/to/file.csv' CSV;
COMMIT;

-- 3. 对批量操作禁用自动提交
-- 在 psql 中: \set AUTOCOMMIT off

-- 4. 使用预处理语句
PREPARE insert_account AS
    INSERT INTO accounts (name, balance) VALUES ($1, $2);

BEGIN;
EXECUTE insert_account('用户1', 100);
EXECUTE insert_account('用户2', 200);
COMMIT;

DEALLOCATE insert_account;

监控事务

-- 查看活动事务
SELECT 
    pid,
    usename,
    state,
    query,
    xact_start,
    query_start,
    state_change
FROM pg_stat_activity
WHERE state = 'active';

-- 检查长时间运行的事务
SELECT 
    pid,
    now() - xact_start AS duration,
    query
FROM pg_stat_activity
WHERE xact_start IS NOT NULL
ORDER BY duration DESC;

-- 终止事务
SELECT pg_terminate_backend(pid);

总结

PostgreSQL 事务提供：

• ACID 保证 - 数据完整性和一致性
• 隔离级别 - 控制并发访问
• 保存点 - 部分回滚能力
• 错误处理 - 强大的故障恢复
• 并发控制 - 防止数据冲突

关键命令：

• BEGIN / START TRANSACTION - 开始事务
• COMMIT - 保存更改
• ROLLBACK - 撤销更改
• SAVEPOINT - 创建检查点
• ROLLBACK TO SAVEPOINT - 部分回滚

理解事务对于维护数据完整性和构建可靠的数据库应用程序至关重要。