能否定义随外键值自增的字段？求Django/MySQL安全高效实现方案

这个需求其实很常见——要让每个Company下的Department的department_number从1开始独立自增，还得保证多进程并发创建时不出现重复编号。下面我从数据库层面和Django层面分别给出安全高效的实现方案，也会对比其他数据库的思路。

一、数据库层面实现（最推荐，并发安全+性能最优） #

数据库原生的原子操作是解决并发问题最靠谱的方式，尤其是MySQL，我们可以通过序列表+触发器或者序列表+Django事务来实现。

1. MySQL 核心思路：用序列表维护每个公司的当前编号 #

首先创建一个单独的序列表，用来记录每个公司下一个要使用的部门编号：

CREATE TABLE company_dept_sequence (
    company_id INT PRIMARY KEY,
    next_num INT DEFAULT 1
);

然后给department表添加复合唯一约束，作为最后一道防线防止重复：

ALTER TABLE department ADD UNIQUE KEY idx_company_dept_num (company_id, department_number);

方式A：数据库触发器自动处理（无需修改Django代码）

写一个触发器，在插入部门时自动从序列表获取并更新编号，利用MySQL的事务和行锁保证并发安全：

DELIMITER //
CREATE TRIGGER before_insert_department
BEFORE INSERT ON department
FOR EACH ROW
BEGIN
    DECLARE next_num INT;
    START TRANSACTION;
    -- 锁定当前公司的序列行，防止并发更新
    SELECT next_num INTO next_num FROM company_dept_sequence WHERE company_id = NEW.company_id FOR UPDATE;
    
    -- 如果是该公司第一个部门，初始化序列
    IF next_num IS NULL THEN
        SET next_num = 1;
        INSERT INTO company_dept_sequence (company_id, next_num) VALUES (NEW.company_id, next_num + 1);
    ELSE
        -- 递增序列
        UPDATE company_dept_sequence SET next_num = next_num + 1 WHERE company_id = NEW.company_id;
    END IF;
    
    SET NEW.department_number = next_num;
    COMMIT;
END //
DELIMITER ;

这样你在Django里创建Department时，完全不用管department_number字段，数据库会自动帮你填充。

方式B：Django事务封装逻辑（更灵活）

如果不想写触发器，可以在Django里用自定义管理器+事务来实现，同样依赖序列表：

from django.db import transaction, models

# 先定义序列模型
class CompanyDeptSequence(models.Model):
    company_id = models.IntegerField(primary_key=True)
    next_num = models.IntegerField(default=1)

    class Meta:
        db_table = 'company_dept_sequence'

# 自定义Department管理器
class DepartmentManager(models.Manager):
    def create_for_company(self, company, **kwargs):
        with transaction.atomic():
            # 初始化或获取该公司的序列
            seq, created = CompanyDeptSequence.objects.get_or_create(
                company_id=company.id,
                defaults={'next_num': 1}
            )
            # 原子更新序列，只有一个进程能成功
            updated = CompanyDeptSequence.objects.filter(
                company_id=company.id,
                next_num=seq.next_num
            ).update(next_num=seq.next_num + 1)
            
            # 处理并发冲突，重试一次
            if not updated:
                seq = CompanyDeptSequence.objects.get(company_id=company.id)
                updated = CompanyDeptSequence.objects.filter(
                    company_id=company.id,
                    next_num=seq.next_num
                ).update(next_num=seq.next_num + 1)
                if not updated:
                    raise RuntimeError("并发创建部门失败，请重试")
            
            # 创建部门
            return self.create(
                company=company,
                department_number=seq.next_num,
                **kwargs
            )

# 关联到Department模型
class Department(models.Model):
    company = models.ForeignKey(Company, on_delete=models.CASCADE)
    department_number = models.IntegerField()
    objects = DepartmentManager()

使用时直接调用Department.objects.create_for_company(company, name="技术部")即可，事务和原子更新保证了并发安全。

2. PostgreSQL 对比方案 #

PostgreSQL的锁机制更灵活，我们可以用函数+触发器实现，不需要单独的序列表：

CREATE OR REPLACE FUNCTION get_next_dept_num(p_company_id INT) RETURNS INT AS $$
DECLARE
    next_num INT;
BEGIN
    -- 锁定该公司的所有部门行，防止并发插入
    SELECT COALESCE(MAX(department_number), 0) + 1 INTO next_num
    FROM department
    WHERE company_id = p_company_id
    FOR UPDATE;
    RETURN next_num;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER before_insert_department
BEFORE INSERT ON department
FOR EACH ROW
EXECUTE FUNCTION get_next_dept_num(NEW.company_id);

同样要添加复合唯一约束，这个方案在PostgreSQL里更简洁，但不适用于MySQL。

二、Django层面纯ORM实现（无需修改数据库） #

如果不想改动数据库结构，可以用Django的事务+行锁来实现，但性能上会稍差一些：

from django.db import transaction, models

class Department(models.Model):
    company = models.ForeignKey(Company, on_delete=models.CASCADE)
    department_number = models.IntegerField()

    def save(self, *args, **kwargs):
        # 只在创建新部门时处理编号
        if not self.pk:
            with transaction.atomic():
                # 锁定当前公司的所有部门行，防止并发读取相同的max值
                Department.objects.filter(company=self.company).select_for_update()
                # 获取当前最大编号，没有则从1开始
                max_num = Department.objects.filter(company=self.company).aggregate(
                    models.Max('department_number')
                )['department_number__max'] or 0
                self.department_number = max_num + 1
        super().save(*args, **kwargs)

这个方案的缺点是：如果一个公司有大量部门，select_for_update()会锁住所有相关行，并发性能不如数据库层面的序列表方案，适合部门数量不多的场景。

三、方案对比总结 #

关键注意事项 #

1. 必须加复合唯一约束：无论用哪种方案，(company_id, department_number)的唯一约束是最后一道防线，能避免代码bug或极端并发导致的重复编号。
2. 并发测试：一定要用多进程工具（比如Python的multiprocessing）模拟并发创建场景，验证编号不会重复。
3. 序列初始化：如果用序列表方案，建议在Company的post_save信号里自动创建对应的序列行，避免第一次插入部门时的初始化问题。

内容的提问来源于stack exchange，提问作者thebjorn