一、错误现象与核心原因分析
（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）常见触发场景：
显式插入重复值：INSERT INTO users VALUES(101, '张三')
批量导入数据时源文件包含重复主键
从其他表迁移数据时的主键冲突
分布式系统全局ID生成异常
自增主键计数器重置导致的值重复
（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）数据溯源方法：
检查最近变更的SQL日志
分析应用程序中的DAO层代码
验证ETL流程的转换逻辑
审计数据库触发器逻辑
检查定时任务的执行日志
（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 ;