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

当使用自定义数据库库和 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”消息队列中提取数据并将其存储在本地以供以后处理）。

首先，请确保celerythingamajig 使用自己的连接，因为

>>> pymysql.threadsafety
1

意思是：“线程可以共享模块，但不能共享连接” https://www.python.org/dev/peps/pep-0249/#threadsafety.