python - 从 celery 执行插入时mysql命令不同步

Question

在使用自定义数据库库和 celery 时，我遇到了可怕的 MySQL 命令不同步问题。

库如下:

import pymysql
import pymysql.cursors
from furl import furl

from flask import current_app

class LegacyDB:
    """Db

    Legacy Database connectivity library

    """

    def __init__(self,app):
        with app.app_context():
            self.rc = current_app.config['RAVEN']
            self.logger = current_app.logger
            self.data = {}
            # setup Mysql
            try:
                uri = furl(current_app.config['DBCX'])
                self.dbcx = pymysql.connect(
                    host=uri.host,
                    user=uri.username,
                    passwd=uri.password,
                    db=str(uri.path.segments[0]),
                    port=int(uri.port),
                    cursorclass=pymysql.cursors.DictCursor
                    )
            except:
                self.rc.captureException()

    def query(self, sql, params = None, TTL=36):
        # INPUT 1 : SQL query
        # INPUT 2 : Parameters
        # INPUT 3 : Time To Live
        # OUTPUT  : Array of result

        # check that we're still connected to the
        # database before we fire off the query
        try:
            db_cursor = self.dbcx.cursor()
            if params:
              self.logger.debug("%s : %s" % (sql, params))
              db_cursor.execute(sql,params)
              self.dbcx.commit()
            else:
              self.logger.debug("%s" % sql)
              db_cursor.execute(sql)
            self.data = db_cursor.fetchall()
            if self.data == None:
              self.data = {}
            db_cursor.close()
        except Exception as ex:
            if ex[0] == "2006":
                db_cursor.close()
                self.connect()
                db_cursor = self.dbcx.cursor()
                if params:
                  db_cursor.execute(sql,params)
                  self.dbcx.commit()
                else:
                  db_cursor.execute(sql)
                self.data = db_cursor.fetchall()
                db_cursor.close()
            else:
                self.rc.captureException()

        return self.data

该库的目的是在我将遗留数据库架构从基于 C++ 的系统迁移到基于 Python 的系统时与 SQLAlchemy 一起工作。

所有配置都是通过 Flask 应用程序完成的，app.config['DBCX'] 值读取与 SQLAlchemy 字符串(“mysql://user:pass@host:port/dbname”)相同，让我可以轻松地将来切换。

我有许多任务通过 celery 运行“INSERT”语句，所有这些任务都使用这个库。正如您可以想象的那样，运行 Celery 的主要原因是我可以增加此应用程序的吞吐量，但是我似乎在一段时间后(大约 500 条已处理的消息)我的库或应用程序中的线程遇到了问题我在日志中查看以下内容:

Stacktrace (most recent call last):

  File "legacy/legacydb.py", line 49, in query
    self.dbcx.commit()
  File "pymysql/connections.py", line 662, in commit
    self._read_ok_packet()
  File "pymysql/connections.py", line 643, in _read_ok_packet
    raise OperationalError(2014, "Command Out of Sync")

很明显我做错了什么导致了这个错误，但是 MySQL 是否启用/禁用了自动提交或者我在哪里调用 connection.commit() 似乎并不重要。

如果我遗漏了 connection.commit() 那么我就不会将任何内容插入到数据库中。

我最近从 mysqldb 转移到 pymysql 并且出现次数似乎较低，但是考虑到这些是简单的“插入”命令而不是复杂的选择(这个数据库甚至没有任何外键约束!)我正在努力找出问题所在。

就目前情况而言，我无法使用 executemany，因为我无法提前准备语句(我从“firehose”消息队列中提取数据并将其存储在本地以供以后处理)。

Answer 1

首先，确保 celery thingamajig 使用它自己的连接，因为

>>> pymysql.threadsafety
1

意思是:"threads may share the module but not connections" .