django并发bulk_create没有将所有值插入数据库答案

【问题标题】：django concurrent bulk_create not inserting all values to databasedjango并发bulk_create没有将所有值插入数据库
【发布时间】：2020-01-12 21:24:15
【问题描述】：

我正在使用 django 的 bulk_create 将大量数据存储到 postgresql 数据库。这是在芹菜任务中发生的。基于不同的参数值一次调用此任务的许多实例。有 4 个不同的工作人员并行运行每个任务实例。当只运行一个任务实例时，所有值都被添加到数据库中，但是当多个实例通过不同的工作人员一起运行时，只有一些值被插入

在为 bulk_create 创建模型对象列表时，我添加了 print 以交叉检查值是否已插入列表中。我发现该值始终添加到列表中，但在 bulk_create 之后，相同的值不会反映在 db 中。我通过在 bulk_create 中添加 time.sleep 添加延迟和 batch_size 参数部分解决了这个问题。但是这个解决方案并不理想，由于数据增加了，不再解决问题。

我无法发布确切的代码，但它是这样的： * 芹菜任务：

def some_function():
   param_lst=[1,3,..] #many values
   all_task=[]
   for i in param_lst:
      all_task.append(some_task1.si(i)) #some_task1 and some_task2 are using majorly similar code, only some data processing changes. both use bulk_create to store values
      all_task.append(some_task2.si(i))
   ch = chord( group( all_task )).set(queue="someQueue")(some_funct) #this is how the tasks are invoked

   if ch.ready():
       ch.get()

任务功能定义

@someapp.task
def some_task1(self,i):
  #process/modify data based on param i
  #we end up with json and iterate over it
  batch = []
  for k in someDict:
    result=MyModel(val1=something,...) #all values inside model
    batch.append(result)
  MyModel.objects.bulk_create(batch) #also tried with batch_size parameter

除了数据修改的一些变化外，其他任务类似。它使用与bulk_create 相同的方式来存储值

没有抛出错误消息。当我只为一个参数运行时，bulk_create 存储所有值，但是当为并行参数列表运行任务时，它会错过向数据库插入一些值。我在网上搜索时找不到这样的东西。我对数据库概念不太了解，对芹菜也很陌生。如果我遗漏了什么或做错了什么，请告诉我

【问题讨论】：

标签： django python-3.x concurrency celery bulkinsert

【解决方案1】：

尝试使用围绕您的 bulk_create 的事务。根据后端，它可能不是事务操作。

from django.db import transaction

@someapp.task
@transaction.atomic
def some_task1(self,i):
  #process/modify data based on param i
  #we end up with json and iterate over it
  batch = []
  for k in someDict:
    result=MyModel(val1=something,...) #all values inside model
    batch.append(result)
  MyModel.objects.bulk_create(batch) #also tried with batch_size parameter

【讨论】：