主键重复错误：原理、解决方案与预防体系

一、错误现象与核心原因分析

（1）典型错误提示形态： MySQL 5.7.34 环境下的完整报错示例：

ERROR 1062 (23000): Duplicate entry '15874' for key 'PRIMARY'

PostgreSQL 14.5 的类似报错：

duplicate key value violates unique constraint "users_pkey"

（2）主键设计规范要求：

• 唯一性原则：主键列的值在表中具有绝对唯一性
• 非空原则：主键字段不允许存储NULL值
• 不可变性：主键值一旦确定不应修改
• 最小化原则：建议使用最简数据类型（如INT而非VARCHAR）

（3）常见触发场景：

1. 显式插入重复值：INSERT INTO users VALUES(101, '张三')
2. 批量导入数据时源文件包含重复主键
3. 从其他表迁移数据时的主键冲突
4. 分布式系统全局ID生成异常
5. 自增主键计数器重置导致的值重复

（4）底层存储原理： 以InnoDB存储引擎为例，主键对应的聚簇索引结构采用B+树组织。当执行插入操作时：

• 通过二分查找法在B+树中定位插入位置
• 发现目标位置已存在相同键值时触发唯一性检查
• 存储引擎层抛出ER_DUP_ENTRY错误
• 事务进入回滚流程（如果启用了事务）

二、深度诊断流程

（1）实时数据核查：

-- MySQL查询重复记录
SELECT *, COUNT(*) AS duplicate_count 
FROM your_table 
GROUP BY primary_key_column 
HAVING duplicate_count > 1;

-- PostgreSQL查询方法
SELECT primary_key_column, COUNT(*) 
FROM your_table 
GROUP BY primary_key_column 
HAVING COUNT(*) > 1;

（2）表结构检查：

SHOW CREATE TABLE your_table;

重点关注：

• 主键字段定义
• AUTO_INCREMENT当前值
• 唯一索引情况

（3）数据溯源方法：

1. 检查最近变更的SQL日志
2. 分析应用程序中的DAO层代码
3. 验证ETL流程的转换逻辑
4. 审计数据库触发器逻辑
5. 检查定时任务的执行日志

（4）并发写入检测：

-- 查看当前锁状态（MySQL）
SHOW ENGINE INNODB STATUS;

-- 检查未提交事务（PostgreSQL）
SELECT * FROM pg_stat_activity 
WHERE state = 'idle in transaction';

三、七种核心解决方案

（1）智能忽略重复插入

-- MySQL语法
INSERT IGNORE INTO users (id, name) 
VALUES (101, '张三'), (102, '李四');

-- PostgreSQL语法
INSERT INTO users (id, name)
VALUES (101, '张三'), (102, '李四')
ON CONFLICT DO NOTHING;

（2）条件更新策略

-- MySQL替换更新
REPLACE INTO users (id, name) VALUES (101, '王五');

-- 更新指定字段
INSERT INTO users (id, name) 
VALUES (101, '赵六') 
ON DUPLICATE KEY UPDATE name=VALUES(name);

-- PostgreSQL合并更新
INSERT INTO users (id, name) 
VALUES (101, '陈七') 
ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name;

（3）安全删除重复项

-- 创建临时表存储重复ID
CREATE TABLE temp_duplicates AS
SELECT id 
FROM users 
GROUP BY id 
HAVING COUNT(*) > 1;

-- 分级删除重复记录
DELETE t1 
FROM users t1 
JOIN users t2 
WHERE t1.id = t2.id 
AND t1.created_at < t2.created_at;

（4）自增主键修复

-- 查询当前自增值
SELECT AUTO_INCREMENT 
FROM information_schema.TABLES 
WHERE TABLE_NAME = 'users';

-- 重置自增计数器
ALTER TABLE users AUTO_INCREMENT = 20000;

（5）主键结构调整

-- 变更主键字段类型
ALTER TABLE users 
MODIFY id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT;

-- 添加复合主键
ALTER TABLE users 
DROP PRIMARY KEY, 
ADD PRIMARY KEY (id, region_code);

（6）事务隔离优化

-- 设置事务隔离级别
SET SESSION TRANSACTION ISOLATION LEVEL SERIALIZABLE;

BEGIN;
SELECT * FROM users WHERE id = 101 FOR UPDATE;
INSERT INTO users (id, name) VALUES (101, '测试');
COMMIT;

（7）全局ID生成方案

-- Snowflake算法实现示例
CREATE FUNCTION generate_snowflake_id()
RETURNS BIGINT
BEGIN
  DECLARE epoch BIGINT DEFAULT 1609459200000; -- 2021-01-01
  DECLARE machine_id INT DEFAULT 5; -- 机器ID
  DECLARE sequence INT DEFAULT 0;
  
  SET @current_time = (UNIX_TIMESTAMP() * 1000) - epoch;
  RETURN (@current_time << 22) | (machine_id << 12) | sequence;
END;

四、预防体系构建

（1）数据校验层设计：

# Python预检查示例
def safe_insert(cursor, id, name):
    cursor.execute("SELECT 1 FROM users WHERE id = %s", (id,))
    if cursor.fetchone():
        print(f"ID {id} 已存在，执行更新操作")
        cursor.execute("UPDATE users SET name = %s WHERE id = %s", (name, id))
    else:
        print(f"插入新记录 {id}")
        cursor.execute("INSERT INTO users (id, name) VALUES (%s, %s)", (id, name))

（2）数据库约束强化：

-- 添加唯一索引
CREATE UNIQUE INDEX idx_unique_email ON users(email);

-- 外键约束示例
ALTER TABLE orders 
ADD CONSTRAINT fk_user_id 
FOREIGN KEY (user_id) 
REFERENCES users(id) 
ON DELETE CASCADE;

（3）数据导入规范：

# 使用mysqlimport预处理
mysqlimport --ignore-lines=1 \
            --fields-terminated-by=, \
            --local --replace \
            -u root -p database users.csv

# 带预处理脚本的导入流程
1. 创建临时表 staging_table
2. 清洗数据：去重、格式转换
3. 执行合并操作：
   INSERT INTO main_table
   SELECT * FROM staging_table
   ON DUPLICATE KEY UPDATE ...
4. 删除临时表

（4）监控预警系统：

-- 创建错误日志表
CREATE TABLE error_log (
    id INT AUTO_INCREMENT PRIMARY KEY,
    error_time DATETIME DEFAULT CURRENT_TIMESTAMP,
    error_code INT,
    error_message TEXT,
    sql_statement TEXT
);

-- 设置触发器记录错误
DELIMITER //
CREATE TRIGGER log_dup_entry 
AFTER ERROR ON your_table 
FOR EACH ROW 
BEGIN
    IF ERRNO = 1062 THEN
        INSERT INTO error_log (error_code, error_message, sql_statement)
        VALUES (1062, OLD.error_message, CURRENT_STATEMENT());
    END IF;
END//
DELIMITER ;

五、高级处理方案

（1）分布式锁实现：

// Redisson分布式锁示例
public void safeInsert(Long id, String name) {
    RLock lock = redisson.getLock("user_lock:" + id);
    try {
        lock.lock();
        User existing = userDao.getById(id);
        if (existing != null) {
            userDao.update(name, id);
        } else {
            userDao.insert(id, name);
        }
    } finally {
        lock.unlock();
    }
}

（2）数据库分片策略：

-- 按地域分片示例
CREATE TABLE users_shard_1 (
    id BIGINT PRIMARY KEY,
    region_code CHAR(2),
    name VARCHAR(50),
    UNIQUE KEY (region_code, id)
) PARTITION BY KEY(region_code);

CREATE TABLE users_shard_2 (
    -- 相同结构
);

（3）时序数据库方案：

-- TimescaleDB处理时序数据
CREATE TABLE sensor_data (
    time TIMESTAMPTZ NOT NULL,
    sensor_id INT,
    value FLOAT,
    PRIMARY KEY (time, sensor_id)
);

SELECT create_hypertable('sensor_data', 'time');

（4）区块链式防篡改设计：

-- 使用触发器实现数据历史记录
CREATE TABLE users_audit (
    id INT,
    name VARCHAR(50),
    modified_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    operation CHAR(1)
);

DELIMITER //
CREATE TRIGGER users_history 
AFTER UPDATE ON users 
FOR EACH ROW 
BEGIN
    INSERT INTO users_audit (id, name, operation)
    VALUES (OLD.id, OLD.name, 'U');
END//
DELIMITER ;