如何使用芹菜来使用mongoengine将数据插入mongodb

Question

我正试图使用芹菜在我的mongodb中插入大数据，但问题是mongodb的并发性。如果我一次向芹菜发送多个任务，部分数据将插入mongodb，而其他一些数据则不会。我认为这是因为mongodb在插入操作时锁定了数据库，但我需要一个能够发送多个相同类型的任务来将数据插入数据库的解决方案。比如，如果数据库正在等待解锁，则检查数据库是否已锁定。这是我的部分代码：

@celery.task(name='celery_tasks.add_book_product')
def add_book_product(product_dict, store_id):
    connect(DefaultConfig.MONGODB_DB, host=DefaultConfig.MONGODB_HOST)
    store_obj = Store.objects.get(pk=store_id)
    try:
        book = Books.objects.get(pk=product_dict['RawBook'])
        try:
            product_obj = Product.objects.get(store=store_obj, related_book=book, kind='book')
            print("Product {} found for store {}".format(product_obj.id, store_obj.id))
            product_obj.count = int(product_dict['count'])
            product_obj.buy_price = int(product_dict['buy_book'])
            product_obj.sell_price = int(product_dict['sell_book'])
            product_obj.save()
        except (DoesNotExist, ValidationError):
            product_obj = Product(store=store_obj,
                                  related_book=book,
                                  kind='book',
                                  count=int(product_dict['count']),
                                  buy_price=int(product_dict['buy_book']),
                                  sell_price=int(product_dict['sell_book']),
                                  name=book.name_fa)
            product_obj.save()
            print("Appending books to store obj...")
            store_obj.products.append(product_obj)
            store_obj.save()
            print("Appending books to store obj done")
        return "Product {} saved for store {}".format(product_obj.id, store_obj.id)
    except (DoesNotExist, ValidationError):
        traceback.print_exc()
        return "Product with raw book {} does not exist.".format(product_dict['RawBook'])

Answer 1

默认情况下，多处理用于并行执行芹菜中的任务。但是有两种方法可以确保在任何给定时间只执行一个任务。

解决方案1:

当你用启动芹菜工人时

celery -A your_app worker -l info

默认的并发性等于您的机器拥有的核心数量。所以，如果你开始一个像这样的工人

celery -A your_app worker -l info -c 1

它在任何给定的时间只运行一个任务。如果您有一些其他任务需要执行，您可以启动一个新队列并指派一名工作人员来执行

解决方案2:

这有点复杂。你需要在你的任务中使用锁，类似这样的东西。

if acquire_lock():
    try:
        #do something
    finally:
        release_lock()
    return 

您可以在Celery文档中阅读更多关于此的信息。