用于芹菜的原始 PyMySQL 多线程

Question

在我目前正在进行的项目中，我不允许使用 ORM，所以我创建了my own

效果很好，但我在使用 Celery 时遇到了问题，而且它是并发的。有一段时间，我将其设置为1（使用--concurrency=1），但我添加了new tasks，这需要比使用celery beat 运行更多的时间来处理，这会导致大量任务积压。

当我将 celery 的并发设置为 > 1 时，会发生以下情况（pastebin，因为它很大）：

https://pastebin.com/M4HZXTDC

关于如何在其他进程上实现某种锁定/等待以使不同的工作人员不会相互交叉的任何想法？

编辑：这里是我设置PyMySQL instance 的地方以及open and close are handled 的设置方式

Answer 1

PyMSQL 不允许线程共享同一个连接（模块可以共享，但线程不能共享连接）。您的 Model 类在任何地方都是 reusing the same connection。

所以，当不同的worker调用模型进行查询时，他们使用的是同一个连接对象，导致冲突。

确保您的连接对象是线程本地的。与其拥有db 类属性，不如考虑一种检索线程本地连接对象的方法，而不是重用可能在不同线程中创建的连接对象。

例如，create your connection in the task。

现在，您对每个模型都使用全局连接。

# Connect to the database
connection = pymysql.connect(**database_config)


class Model(object):
    """
    Base Model class, all other Models will inherit from this
    """

    db = connection

为避免这种情况，您可以改为在 __init__ 方法中创建数据库...

class Model(object):
    """
    Base Model class, all other Models will inherit from this
    """

    def __init__(self, *args, **kwargs):
        self.db = pymysql.connect(**database_config)

但是，这可能不是高效/实用的，因为 db 对象的每个实例都会创建一个会话。

为了改进这一点，您可以使用一种使用threading.local 的方法来保持线程本地连接。

class Model(object):
    """
    Base Model class, all other Models will inherit from this
    """
    _conn = threading.local()
    @property
    def db(self):
        if not hasattr(self._conn, 'db'):
            self._conn.db = pymysql.connect(**database_config)
        return self._conn.db

请注意，假设您使用的是线程并发模型，则线程本地解决方案有效。另请注意，celery 默认使用多个进程（prefork）。这可能是也可能不是问题。如果这是一个问题，你可以通过change the workers to use eventlet 来解决它。