1. 批量操作的现实挑战

当我第一次接手需要处理千万级数据同步的项目时，传统的逐条写入方式让整个流程慢得像是老牛拉破车。服务器CPU居高不下，同事的咖啡杯从热放到凉，一个导入任务还没跑完。这种场景让我深刻意识到：数据库操作必须从"单兵作战"转向"集团军作战"。

PostgreSQL本身没有原生表值参数（TVP）的支持，这常常让熟悉SQL Server的开发者感到困惑。但通过我们的实践发现，只要善用其强大的数组类型和JSON处理能力，反而能开发出更具弹性的批量处理方案。

2. 核心技术实现方法

我们基于PostgreSQL 14版本和PL/pgSQL进行实践，以下是三种主流解决方案：

2.1 数组类型实现

-- 创建复合类型
CREATE TYPE order_item AS (
    product_id INT,
    quantity INT,
    unit_price NUMERIC(10,2)
);

-- 批量插入函数
CREATE OR REPLACE FUNCTION batch_insert_orders(
    order_items order_item[]
) RETURNS VOID AS $$
DECLARE
    single_item order_item;
BEGIN
    FOREACH single_item IN ARRAY order_items
    LOOP
        INSERT INTO order_details 
        VALUES (
            NEXTVAL('order_seq'),
            single_item.product_id,
            single_item.quantity,
            single_item.unit_price,
            NOW()
        );
    END LOOP;
END;
$$ LANGUAGE plpgsql;

/* 调用示例 */
SELECT batch_insert_orders(ARRAY[
    (1001, 5, 99.99),
    (1002, 3, 199.50),
    (1003, 10, 29.95)::order_item
]);

2.2 JSON批量处理

-- 创建JSON处理函数
CREATE OR REPLACE FUNCTION json_batch_insert(
    items_json JSON
) RETURNS VOID AS $$
BEGIN
    INSERT INTO inventory_log
    SELECT 
        (elem->>'sku')::VARCHAR,
        (elem->>'change_qty')::INT,
        (elem->>'operation_type')::OPERATION_ENUM,
        NOW()
    FROM json_array_elements(items_json) AS elem;
END;
$$ LANGUAGE plpgsql;

/* 调用示例 */
SELECT json_batch_insert('[
    {"sku": "A001", "change_qty": 50, "operation_type": "STOCK_IN"},
    {"sku": "B205", "change_qty": -20, "operation_type": "SALES"}
]');

2.3 临时表模式

-- 创建临时表对接
CREATE TEMP TABLE temp_employee_data (
    name VARCHAR(50),
    department_id INT,
    salary NUMERIC(10,2)
) ON COMMIT DROP;

-- 使用COPY命令快速装载数据
\COPY temp_employee_data FROM '/data/employee_batch.csv' WITH CSV HEADER;

-- 执行批量合并操作
INSERT INTO employees
SELECT 
    NEXTVAL('emp_seq'),
    name,
    department_id,
    salary
FROM temp_employee_data
ON CONFLICT (department_id, name) 
DO UPDATE SET salary = EXCLUDED.salary;

3. 技术方案对比分析

3.1 数组方式的优势

在最近的压力测试中，使用数组传输1000条记录时，性能比传统方式提升约60%。但需要注意：

-- 调整work_mem提升数组处理性能
SET work_mem = '64MB';

-- 查看数组维度限制
SHOW max_array_size;

3.2 JSON方案的灵活性

处理嵌套数据结构时展现出独特优势：

-- 复杂JSON结构处理示例
WITH complex_data AS (
    SELECT '[{
        "order_no": "20230815001",
        "items": [
            {"sku": "A100", "qty": 2},
            {"sku": "B200", "qty": 1}
        ]
    }]'::JSON AS input
)
INSERT INTO order_headers
SELECT 
    elem->>'order_no',
    (item->>'sku')::VARCHAR,
    (item->>'qty')::INT
FROM complex_data,
json_array_elements(input) AS elem,
json_array_elements(elem->'items') AS item;

3.3 临时表的实战技巧

大容量数据场景下的最佳实践：

-- 创建unlogged表提升写入速度
CREATE UNLOGGED TABLE batch_stage (
    -- 字段定义
);

-- 并行装载配置
SET max_parallel_workers = 8;
SET max_parallel_workers_per_gather = 4;

4. 进阶优化策略

4.1 预编译语句应用

-- 准备批量插入语句
PREPARE bulk_plan (order_item[]) AS
INSERT INTO orders SELECT * FROM unnest($1);

-- 执行预编译语句
EXECUTE bulk_plan(ARRAY[
    (1001,5,99.99),
    (1002,3,199.50)
]::order_item[]);

4.2 批量更新实践

-- 使用CTE批量更新
WITH updates AS (
    VALUES
    (1001, 109.99),
    (1002, 205.00)
) 
UPDATE products p
SET price = u.new_price
FROM updates u
WHERE p.id = u.product_id;

4.3 事务控制要点

BEGIN;
SAVEPOINT before_bulk;

-- 执行批量操作
PERFORM process_large_batch();

-- 异常处理示例
EXCEPTION WHEN OTHERS THEN
    ROLLBACK TO before_bulk;
    RAISE NOTICE '批量操作失败，已回滚';
COMMIT;

5. 典型应用场景解析

某电商平台在促销活动期间，通过数组方式实现秒杀订单的批量处理：

-- 创建订单和库存的原子操作
CREATE OR REPLACE PROCEDURE create_orders_with_stock(
    orders order_item[]
) LANGUAGE plpgsql AS $$
BEGIN
    -- 扣减库存
    UPDATE products p
    SET stock = stock - o.quantity
    FROM unnest(orders) o
    WHERE p.id = o.product_id;

    -- 生成订单
    INSERT INTO order_details
    SELECT 
        NEXTVAL('order_seq'),
        o.product_id,
        o.quantity,
        o.unit_price,
        NOW()
    FROM unnest(orders) o;
END;
$$;

6. 技术方案综合评估

优势对比表

方法 数据量级 类型支持 事务控制 网络开销
数组类型 < 1万 结构化 简单 低
JSON格式 < 10万 灵活 中等 中等
临时表 > 50万 复杂 复杂 高

常见陷阱及解决方案

  1. 数组越界问题：
-- 设置合理的数组尺寸
SET max_array_size = 100000;
  1. JSON性能瓶颈：
-- 创建表达式索引
CREATE INDEX idx_gin_data ON orders USING GIN (order_data);
  1. 临时表连接泄漏：
-- 使用ON COMMIT子句自动清理
CREATE TEMP TABLE temp_data (...) ON COMMIT DELETE ROWS;

7. 最佳实践指南

根据我们的实战经验，推荐的选择策略：

  1. 初创系统优先采用数组方案，结构清晰易维护
  2. 快速迭代项目推荐JSON格式，方便动态扩展
  3. 数据分析系统首选临时表方式，便于与ETL流程集成
  4. 金融交易系统建议数组+预编译语句的组合方案

监控建议配置：

-- 监控批量操作性能
CREATE EXTENSION pg_stat_statements;

SELECT query, calls, total_time
FROM pg_stat_statements
WHERE query LIKE '%unnest(%';

8. 总结与展望

通过三种替代方案的对比，我们发现没有放之四海皆准的"银弹"。某物流企业采用混合方案后，日处理能力从50万单提升到300万单，验证了灵活组合的价值。随着PostgreSQL 16即将推出的逻辑复制增强，未来批量处理还可能迎来更多创新方案。