在所有 celery worker 之间共享一个到 mongo db 的连接

Posted 2023-03-16

【中文标题】在所有 celery worker 之间共享一个到 mongo db 的连接

【英文标题】：Sharing a single connection to mongo db between all celery workers

【发布时间】：2021-09-26 12:12:47

【问题描述】：

我有一系列执行任务的芹菜工人，使用 REDIS 存储；特别是我有三类工人，它们分别执行三类任务。属于这些类别中的每一个的任务都需要访问 mongo DB。出于效率的原因，我想使用单个连接到数据库以供所有工作人员使用。到目前为止，我已经尝试通过以下方式将连接作为参数传递给“send_task”：

myclient = MongoClient('localhost:27017')
celeryWorker.send_task('tasks.beampolyline', myclient)

但显然它返回对象不是 JSON 可序列化的错误；两者都试图在所有工作人员之间共享 myclient 对象，但结果不佳。任何想法 ？我觉得我非常接近解决方案，但我被困在我尝试使用的两种方法上。

【参考方案1】：

一种解决方案是定义模块级连接。假设 worker.py 作为你的工作模块：

from mongodb import MongoClient

shared_connection = MongoClient()

@app.task(...)
def task(self, a, b):
    ...  # use shared_connection inside task
    return 

在此解决方案中，此工作线程/进程之间共享连接。

【讨论】：

也许我错了，但在那种情况下，我不会为每种类型的任务建立连接吗？因此，在实践中拥有 3 种类型的任务，我将拥有 3 个与数据库的连接。我想要一个可以在所有任务/工作人员之间共享的独特的，基本上是单例方法

你可以在一个模块中定义所有的worker。在模块导入中建立连接。因此，一个模块中所有工作人员的实例之间只建立和共享一个连接。

不幸的是，这对我来说是不可能的，我必须将模块中的各种工作人员保存在单独的文件中。这样的组织真的没有办法吗？

【参考方案2】：

你不能做类似的事情吗？：

import pymongo
from pymongo import MongoClient
from flask import *

client = pymongo.MongoClient("yourcredentials")
db = client['ALLWORKER']
worker_1 = db['worker1']
worker_2 = db['worker_2']