当 SpringBoot、Redisson、PostgreSQL 一起给我上了一课

🧩 业务背景¶

我们有一个 ETL 数据刷新功能，负责将清洗完的结果数据写回数据库。这个看似简单的功能，却在高并发场景下暴露出了一系列问题。

📋 接口设计¶

入参结构：

• tableName：目标 DWB 表名
• List<Map<String, Object>> data：清洗后数据，每条数据约 20 个字段

核心逻辑：

1. 按业务主键（如 biz_id）进行分组
2. 每个组串行执行：先删除旧数据 → 插入新数据（在同一个事务中完成）
3. 并发调用时，不同组之间通过分布式锁串行化（以 biz_id 为单位）

🎯 预期 vs 现实¶

理想很丰满，现实很骨感。下面的对比表清晰展示了我们遇到的三大核心问题：

💥 问题一：插入性能瓶颈¶

🔍 问题代码¶

/**
 * JDBC 批量插入实现
 * 性能问题：对于大数据集极其缓慢
 */
public void batchInsert(Connection conn, String tableName, 
                       List<Map<String, Object>> dataList) throws SQLException {
    String sql = buildInsertSql(tableName); // INSERT INTO table (col1, col2, ..., col20) VALUES (?, ?, ..., ?)
    
    try (PreparedStatement ps = conn.prepareStatement(sql)) {
        for (Map<String, Object> row : dataList) {
            // 为每行设置参数
            for (int i = 1; i <= 20; i++) {
                ps.setObject(i, row.get("col" + i));
            }
            ps.addBatch();
        }
        
        // 执行批处理 - 这里是瓶颈所在
        ps.executeBatch();
    }
}

🔬 根因分析¶

通过实际测试，我们发现了两种插入方式的巨大性能差异：

性能对比：

💥 问题二：分布式锁与事务边界不匹配¶

并发安全事故 明明加了分布式锁，理论上同一 bizKey 的处理应该是串行的，但还是频繁出现：

org.postgresql.util.PSQLException: ERROR: duplicate key value violates unique constraint

🔍 问题代码¶

/**
 * 有问题的实现：锁在事务提交前就释放了
 */
@Transactional
public void refreshData(String table, List<Map<String, Object>> data) {
    String bizKey = generateBizKey(data);
    RLock lock = redissonClient.getLock(bizKey);
    
    lock.lock();
    try {
        // 删除旧数据
        deleteOldData(table, data);
        // 插入新数据  
        insertNewData(table, data);
    } finally {
        lock.unlock(); // 🔥 此时事务未提交！另一个线程已进来执行删除 → 主键冲突
    }
    // 🚨 事务在方法返回后才提交 (Spring AOP)
}

🎯 根因分析¶

sequenceDiagram
    participant T1 as Thread 1
    participant T2 as Thread 2
    participant DB as Database
    participant Redis as Redis Lock
    
    T1->>Redis: lock.lock()
    T1->>DB: DELETE old data
    T1->>DB: INSERT new data
    T1->>Redis: lock.unlock() 🔥
    Note over T1: Transaction NOT committed yet!
    
    T2->>Redis: lock.lock() ✅ (gets lock)
    T2->>DB: DELETE old data
    T2->>DB: INSERT new data 💥 (conflicts with T1's uncommitted data)
    
    Note over T1: Spring commits T1's transaction
    Note over T2: T2 fails with duplicate key error

💥 问题三：PostgreSQL 死锁¶

数据库死锁频发 并发高峰期，PostgreSQL 频繁出现：

ERROR: deadlock detected
DETAIL: Process 12345 waits for ShareLock on transaction 67890; blocked by process 54321.
Process 54321 waits for ShareLock on transaction 12346; blocked by process 12345.

🔍 死锁场景还原¶

假设两个线程并发执行：

操作序列：

-- 线程 A
BEGIN;
DELETE FROM dwb_target WHERE biz_id = 101;  -- 获取行锁 A
-- 准备插入...

-- 线程 B  
BEGIN;
DELETE FROM dwb_target WHERE biz_id = 202;  -- 获取行锁 B
-- 准备插入...

🧠 死锁形成机制¶

为什么不同 biz_id 也会冲突？

• 共享相同的索引结构（主键索引、唯一索引）
• 相邻值可能命中同一个索引页
• DELETE + INSERT 操作涉及TOAST 表、WAL 写入等共享资源

🛠️ 解决方案¶

✅ 方案一：使用 PostgreSQL COPY 提升插入性能¶

🚀 实现代码¶

/**
 * 使用 PostgreSQL COPY 的高性能批量插入
 */
public class PostgresCopyInserter {
    
    public void copyInsert(Connection connection, String tableName, 
                          List<Map<String, Object>> dataList) throws SQLException, IOException {
        
        // 构建 COPY 命令
        String copyCommand = buildCopyCommand(tableName);
        
        // 将数据转换为 CSV 格式
        String csvData = convertToCsv(dataList);
        
        // 执行 COPY 操作
        CopyManager copyManager = new CopyManager((BaseConnection) connection);
        try (StringReader reader = new StringReader(csvData)) {
            long rowsInserted = copyManager.copyIn(copyCommand, reader);
            log.info("Successfully inserted {} rows using COPY", rowsInserted);
        }
    }
    
    private String buildCopyCommand(String tableName) {
        return String.format(
            "COPY %s (col1, col2, col3, ..., col20) FROM STDIN WITH (FORMAT csv, HEADER false)",
            tableName
        );
    }
    
    private String convertToCsv(List<Map<String, Object>> dataList) {
        StringBuilder csv = new StringBuilder();
        for (Map<String, Object> row : dataList) {
            csv.append(escapeCsvValue(row.get("col1"))).append(',')
               .append(escapeCsvValue(row.get("col2"))).append(',')
               // ... 其他列
               .append(escapeCsvValue(row.get("col20"))).append('\n');
        }
        return csv.toString();
    }
    
    private String escapeCsvValue(Object value) {
        if (value == null) return "";
        String str = value.toString();
        // 处理 CSV 转义：引号、逗号、换行符
        if (str.contains(",") || str.contains("\"") || str.contains("\n")) {
            return "\"" + str.replace("\"", "\"\"") + "\"";
        }
        return str;
    }
}

⚠️ COPY 使用注意事项¶

使用限制与注意事项：

在选择 PostgreSQL COPY 时，需要权衡其优势与限制：

✅ 方案二：编程式事务精确控制边界¶

🎯 实现代码¶

编程式事务完整实现：

/**
 * 编程式事务管理，精确控制
 */
@Service
public class DataRefreshService {
    
    @Autowired
    private TransactionTemplate transactionTemplate;
    
    @Autowired
    private RedissonClient redissonClient;
    
    public void refreshDataWithLock(String tableName, List<Map<String, Object>> data) {
        String bizKey = generateBizKey(data);
        RLock lock = redissonClient.getLock(bizKey);
        
        // 获取分布式锁
        lock.lock();
        try {
            // 在事务边界内执行
            transactionTemplate.execute(status -> {
                try {
                    // 删除旧数据
                    deleteOldData(tableName, data);
                    
                    // 使用 COPY 插入新数据
                    copyInsert(tableName, data);
                    
                    return null;
                } catch (Exception e) {
                    status.setRollbackOnly();
                    throw new RuntimeException("Data refresh failed", e);
                }
            });
            // ✅ 事务在这里提交，锁释放之前
            
        } finally {
            lock.unlock(); // 🔐 锁在事务提交后释放
        }
    }
    
    private String generateBizKey(List<Map<String, Object>> data) {
        // 从数据中提取业务键
        return data.stream()
                  .map(row -> String.valueOf(row.get("biz_id")))
                  .distinct()
                  .sorted()
                  .collect(Collectors.joining(","));
    }
}

📊 事务管理对比¶

两种事务管理方式的对比：

选择合适的事务管理方式对于解决锁与事务边界问题至关重要：

✅ 方案三：统一访问顺序避免死锁¶

🔄 实现策略¶

有序处理防死锁完整实现：

/**
 * 通过有序处理防止死锁
 */
public void processDataGroups(String tableName, List<Map<String, Object>> allData) {
    // 按业务键分组数据
    Map<String, List<Map<String, Object>>> groupedData = 
        allData.stream().collect(Collectors.groupingBy(this::extractBizKey));
    
    // 按排序顺序处理分组以防止死锁
    groupedData.entrySet().stream()
        .sorted(Map.Entry.comparingByKey()) // 🔑 关键点：一致的排序
        .forEach(entry -> {
            String bizKey = entry.getKey();
            List<Map<String, Object>> groupData = entry.getValue();
            
            // 使用分布式锁处理每个分组
            processGroupWithLock(tableName, bizKey, groupData);
        });
}

private void processGroupWithLock(String tableName, String bizKey, 
                                 List<Map<String, Object>> groupData) {
    RLock lock = redissonClient.getLock("data_refresh:" + bizKey);
    lock.lock();
    try {
        transactionTemplate.execute(status -> {
            deleteOldData(tableName, bizKey);
            copyInsert(tableName, groupData);
            return null;
        });
    } finally {
        lock.unlock();
    }
}

🛡️ 死锁预防策略总结¶

三重保障机制：

我们采用了多层次的死锁预防策略，确保系统在高并发场景下的稳定性：

📈 优化效果¶

🚀 性能提升¶

经过系统性优化后，我们取得了令人瞩目的性能提升效果：

🎯 最佳实践总结¶

💡 性能优化原则¶

1. 选择合适的批量操作方式

   • 大批量数据：优先考虑 PostgreSQL COPY
   • 小批量数据：JDBC Batch 可接受
   • 需要复杂逻辑：考虑存储过程

2. 事务边界设计

   • 明确事务的开始和结束时机
   • 避免长事务持有锁资源
   • 编程式事务提供更精确的控制

🔒 并发安全原则¶

1. 分布式锁使用规范

   • 锁的粒度要合适（不要太粗也不要太细）
   • 确保锁释放在事务提交之后
   • 设置合理的锁超时时间

2. 死锁预防策略

   • 统一资源访问顺序
   • 减少事务持有时间
   • 避免嵌套锁和交叉锁

🛠️ 技术选型建议¶

根据不同的应用场景，我们推荐以下技术选型方案：

📝 写在最后¶

这次踩坑经历让我深刻认识到：