Django批量导入用户优化：新增用户时SELECT与INSERT延迟过高

我需要实现创建新课程并导入用户的功能，要求用户可加入不同课程但不能重复导入同一用户到同一课程。最初用get_or_create()实现，但批量导入大量用户时程序运行极慢。

原代码实现 #

def post(self, request):
    class_data = request.data
    class_data["start_time"] = dateutil.parser.parse(class_data["start_time"])
    class_data["end_time"] = dateutil.parser.parse(class_data["end_time"])
    class_data["created_by"] = request.user

    if class_data["end_time"] <= class_data["start_time"]:
        return self.error("Start time must occur earlier than end time")

    user_data = class_data.pop("users")  # Retrieve user data and remove it from the class data

    try:
        with transaction.atomic():
            if Class.objects.filter(title=class_data["title"]).exists():
                return self.error("Class with the same title already exists")

            class_obj = Class.objects.create(**class_data)  # Create the class object

            # Add the creator to the class members
            if not class_obj.users.filter(id=request.user.id).exists():
                class_obj.users.add(request.user)

            for data in user_data:
                if len(data) != 4 or len(data[1]) > 32:
                    return self.error(f"Error occurred while processing data '{data}'")

                username = data[1]
                user, created = User.objects.get_or_create(username=username, defaults={
                    "password": make_password(data[1]),
                    "college": data[3],
                    "student_number": data[1]
                })
                if created:
                    profile = UserProfile(user=user, real_name=data[2])
                    profile.save()

                # class_obj.users.add(user)
                if not class_obj.users.filter(id=user.id).exists():
                    class_obj.users.add(user)
        return self.success(ClassAdminSerializer(class_obj).data)
    except IntegrityError as e:
        return self.error(str(e).split("\n")[1])

性能瓶颈定位 #

我找到了三个核心性能瓶颈：

• get_or_create()逐个检查用户存在性，单次查询累加后耗时极久
• profile.save()逐条插入用户档案数据，IO开销大
• class_obj.users.add(user)逐条向多对多关联表插入数据，重复发起数据库请求

模型定义 #

User模型 #

class User(AbstractBaseUser):
    id = models.BigAutoField(primary_key=True)

    username = models.TextField(unique=True)
    email = models.TextField(blank=True, null=True)
    college = models.TextField(default=College.COMMUNICATION_COLLEGE)
    student_number = models.TextField(null=True)
    major = models.TextField(null=True)

    create_time = models.DateTimeField(auto_now_add=True, null=True)
    admin_type = models.TextField(default=AdminType.REGULAR_USER)
    problem_permission = models.TextField(default=ProblemPermission.NONE)
    reset_password_token = models.TextField(null=True)
    reset_password_token_expire_time = models.DateTimeField(null=True)
    auth_token = models.TextField(null=True)
    two_factor_auth = models.BooleanField(default=False)
    tfa_token = models.TextField(null=True)
    session_keys = JSONField(default=list)
    open_api = models.BooleanField(default=False)
    open_api_appkey = models.TextField(null=True)
    is_disabled = models.BooleanField(default=False)

    USERNAME_FIELD = "username"
    REQUIRED_FIELDS = []

    objects = UserManager()

    def is_admin(self):
        return self.admin_type == AdminType.ADMIN

    def is_super_admin(self):
        return self.admin_type == AdminType.SUPER_ADMIN

    def is_admin_role(self):
        return self.admin_type in [AdminType.ADMIN, AdminType.SUPER_ADMIN]

    def can_mgmt_all_problem(self):
        return self.problem_permission == ProblemPermission.ALL

    def is_contest_admin(self, contest):
        return self.is_authenticated and (contest.created_by == self or self.admin_type == AdminType.SUPER_ADMIN)

    def is_Class_admin(self, class_obj):
        return self.is_authenticated and (class_obj.created_by == self or self.admin_type == AdminType.SUPER_ADMIN)

    class Meta:
        db_table = "user"

Class模型 #

class Class(models.Model):
    title = models.TextField()
    start_time = models.DateTimeField()

    end_time = models.DateTimeField()
    create_time = models.DateTimeField(auto_now_add=True)
    
    real_time_rank = models.BooleanField(default=True)

    created_by = models.ForeignKey(User, on_delete=models.CASCADE, related_name='created_classes')

    users = models.ManyToManyField(User, related_name='enrolled_classes')

    @property
    def status(self):
        if self.start_time > now():
            return ClassStatus.CLASS_NOT_START
        elif self.end_time < now():
            return ClassStatus.CLASS_ENDED
        else:
            return ClassStatus.CLASS_UNDERWAY

    def problem_details_permission(self, user):
        return self.status == ClassStatus.CLASS_ENDED or \
                user.is_authenticated and user.is_Class_admin(self) or \
                self.real_time_rank

    class Meta:
        db_table = "Class"
        ordering = ("-start_time",)

优化后的批量操作代码 #

def post(self, request):
    class_data = request.data
    class_data["start_time"] = dateutil.parser.parse(class_data["start_time"])
    class_data["end_time"] = dateutil.parser.parse(class_data["end_time"])
    class_data["created_by"] = request.user

    if class_data["end_time"] <= class_data["start_time"]:
        return self.error("Start time must occur earlier than end time")

    user_data = class_data.pop("users")  # Retrieve user data and remove it from the class data

    try:
        with transaction.atomic():
            if Class.objects.filter(title=class_data["title"]).exists():
                return self.error("Class with the same title already exists")

            class_obj = Class.objects.create(**class_data)  # Create the class object
            # To add the creator to the class members
            class_obj.users.add(request.user)

            # Bulk create users
            usernames = [data[1] for data in user_data]

            # To check if a user already exists
            existing_users = User.objects.filter(username__in=usernames)
            existing_usernames = set(existing_users.values_list('username', flat=True))

            users_to_create = []
            users_to_add = []

            for data in user_data:
                if len(data) != 4 or len(data[1]) > 32:
                    return self.error(f"Error occurred while processing data '{data}'")

                username = data[1]

                if username not in existing_usernames:
                    # If the user does not exist, create a new user
                    user = User(username=username, password=make_password(data[1]),
                                college=data[3], student_number=data[1])
                    users_to_create.append(user)
                    users_to_add.append(user)
                else:
                    # If the user already exists, add them to the class 
                    user = existing_users.get(username=username)
                    if not class_obj.users.filter(id=user.id).exists():
                        users_to_add.append(user)

            User.objects.bulk_create(users_to_create)

            # Bulk create user profiles
            profiles_to_create = [UserProfile(user=user, real_name=data[2]) for user, data in zip(users_to_create, user_data)]
            UserProfile.objects.bulk_create(profiles_to_create)

            # Bulk add users to class
            class_obj.users.add(*users_to_add)

        return self.success(ClassAdminSerializer(class_obj).data)

    except IntegrityError as e:
        return self.error(str(e).split("\n")[1])

当前问题 #

修改后发现，若导入的用户均已存在，课程创建速度很快；但需新增大量用户时，SELECT与INSERT操作间出现约一分钟延迟，怀疑可能是硬件问题，寻求解决方案。

解决方案 #

1. 修复批量创建Profile的匹配错误 #

当前zip(users_to_create, user_data)会导致用户档案与数据不匹配（users_to_create仅包含新用户，而user_data是全部导入数据），需先构建用户名映射：

# 提前构建用户名到真实姓名的映射
username_realname_map = {data[1]: data[2] for data in user_data}
# 批量创建Profile时直接通过用户名匹配
profiles_to_create = [
    UserProfile(user=user, real_name=username_realname_map[user.username])
    for user in users_to_create
]

2. 减少重复数据库查询 #

将existing_users转为字典，避免循环中多次调用get()触发查询：

existing_users_dict = {user.username: user for user in existing_users}
# 后续获取用户直接从字典读取
user = existing_users_dict[username]

3. 优化批量插入的批次大小 #

使用bulk_create的batch_size参数拆分插入批次，降低数据库负载：

# 根据数据库性能调整批次大小，MySQL建议1000条/批
User.objects.bulk_create(users_to_create, batch_size=1000)
UserProfile.objects.bulk_create(profiles_to_create, batch_size=1000)

4. 检查并优化数据库索引 #

• 确认User.username的唯一索引正常生效（模型中unique=True已自动创建）
• 检查多对多关联表（默认表名class_users）的联合索引，若缺失则手动创建：

CREATE INDEX idx_class_users_class_user ON class_users(class_id, user_id);

5. 事务内操作瘦身 #

将用户数据格式检查（len(data)!=4等）移到事务外部，避免事务占用数据库连接过长时间：

# 先做数据格式校验，不符合直接返回错误
for data in user_data:
    if len(data) != 4 or len(data[1]) > 32:
        return self.error(f"Error occurred while processing data '{data}'")

# 再进入事务执行数据库操作
with transaction.atomic():
    # ... 后续逻辑

6. 硬件与数据库配置优化 #

• 若磁盘IO是瓶颈，更换为SSD存储，可大幅提升写入速度
• 调整数据库缓存配置，比如MySQL的innodb_buffer_pool_size设置为服务器内存的70%-80%
• 临时关闭非必要的数据库日志（如二进制日志），或调整日志刷盘策略为异步

7. 异步处理超大批量导入 #

如果导入数据量超过上万条，可将导入逻辑放到异步任务中（如Celery），避免HTTP请求超时，同时后台处理不影响用户体验。

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