多线程RabbitMQ消费者中Django model.save()更新不一致问题

背景 #

在RabbitMQ消费者的回调函数中使用Django ORM操作MySQL数据库，消费者运行在独立线程中，每个消费者都已建立独立队列连接。以下是两个消费者的回调代码：

TasksExecutorService #

# imports
from pika.spec import Basic
from pika.channel import Channel
from pika import BasicProperties

import uuid

from jobs.models import Task

from exceptions import MasterConsumerServiceError as ServiceError

from .master_service import MasterConsumerSerivce


class TaskExecutorService(MasterConsumerSerivce):
  queue = 'master_tasks'

  @classmethod
  def callback(cls, ch: Channel, method: Basic.Deliver, properties: BasicProperties, message: dict):
    # get task
    task_id_str = message.get('task_id')
    task_id = uuid.UUID(task_id_str)
    task_qs = Task.objects.filter(pk=task_id)
    if not task_qs.exists():
      raise ServiceError(message=f'Task {task_id_str} does not exist')
    task = task_qs.first()

    # check if task is stopped
    if task.status == cls.Status.TASK_STOPPED:
      raise ServiceError(message=f'Task {task_id_str} is stopped')

    # send task to results queue
    publisher = cls.get_publisher(queue=cls.Queues.results_queue)
    published, error = publisher.publish(message=message | {'status': True, 'error': None})
    if not published:
      raise ServiceError(message=str(error))

    # update task status
    task.status = cls.Status.TASK_PROCESSING
    task.save()

    return

ResultsHandlerService #

# imports
from pika.spec import Basic
from pika.channel import Channel
from pika import BasicProperties

import uuid

from jobs.models import Task
from exceptions import MasterConsumerServiceError as ServiceError

from .master_service import MasterConsumerSerivce


class ResultHandlerService(MasterConsumerSerivce):
  queue = 'master_results'

  @classmethod
  def callback(cls, ch: Channel, method: Basic.Deliver, properties: BasicProperties, message: dict):
    # get task
    task_id_str = message.get('task_id')
    task_id = uuid.UUID(task_id_str)
    task_qs = Task.objects.filter(pk=task_id)
    if not task_qs.exists():
      raise ServiceError(message=f'Task {task_id_str} does not exist')
    task = task_qs.first()

    # get result data and status
    data = message.get('data')
    status = message.get('status')

    # if task is not successful
    if not status:
      # fail task
      task.status = cls.Status.TASK_FAILED
      task.save()

      # fail job
      task.job.status = cls.Status.JOB_FAILED
      task.job.save()

      return

    # update task status
    task.status = cls.Status.TASK_DONE
    task.save()

    # check if job is complete
    task_execution_order = task.process.execution_order
    next_task_qs = Task.objects.select_related('process').filter(job=task.job, process__execution_order=task_execution_order + 1)
    is_job_complete = not next_task_qs.exists()

    # check job is complete
    if is_job_complete:
      # publish reults
      publisher = cls.get_publisher(queue=cls.Queues.output_queue)
      published, error = publisher.publish(message={'job_id': str(task.job.id), 'data': data})
      if not published:
        raise ServiceError(message=str(error))

      # update job status
      task.job.status = cls.Status.JOB_DONE
      task.job.save()

    # otherwise
    else:
      # publish next task
      next_task = next_task_qs.first()
      publisher = cls.get_publisher(queue=cls.Queues.tasks_queue)
      published, error = publisher.publish(message={'task_id': str(next_task.id), 'data': data})
      if not published:
        raise ServiceError(message=str(error))

      # update next task status
      next_task.status = cls.Status.TASK_QUEUED
      next_task.save()

    return

问题描述 #

执行task.status = cls.Status.TASK_ABC后调用task.save()时行为不稳定：

• 偶尔状态能正常更新到数据库
• 多数情况下，即便流程正常完成、输出队列已填充结果，数据库中的状态仍未变更
• task.save()后打印日志，日志显示状态已更新，但数据库对应值无变化

解决方案 #

1. 确保线程内数据库连接正确初始化/回收 #

Django的数据库连接是线程隔离的，独立线程中的消费者可能未正确初始化连接。在回调函数开头手动初始化连接，结束后关闭：

from django.db import connection

@classmethod
def callback(cls, ch: Channel, method: Basic.Deliver, properties: BasicProperties, message: dict):
    # 初始化数据库连接
    connection.connect()
    try:
        # 原有的ORM操作代码
        ...
    finally:
        # 关闭连接
        connection.close()

或使用上下文管理器自动管理连接：

from django.db import connection

@classmethod
def callback(cls, ch: Channel, method: Basic.Deliver, properties: BasicProperties, message: dict):
    with connection.cursor():
        # 原有的ORM操作代码
        ...

2. 使用行级锁避免并发更新冲突 #

多线程同时操作同一条Task记录时，可能出现丢失更新的情况。获取Task时使用select_for_update()加行级锁：

# 替换原有的task获取逻辑
task = Task.objects.select_for_update().get(pk=task_id)

这样能确保当前线程更新时，其他线程无法修改该记录，避免并发冲突导致的更新丢失。

3. 明确指定更新字段 #

有时Django ORM可能无法检测到状态字段的变化（比如枚举值类型的隐式转换问题），调用save()时明确指定更新字段强制写入：

task.status = cls.Status.TASK_PROCESSING
task.save(update_fields=['status'])

4. 检查MasterConsumerService的连接管理逻辑 #

确保MasterConsumerSerivce中没有共享数据库连接对象的逻辑，每个线程的消费者应使用独立的数据库连接，避免连接复用导致的异常。

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