MySQL 在 celery 任务期间不断失去连接答案

【问题标题】：MySQL keeps losing connection during celery tasksMySQL 在 celery 任务期间不断失去连接
【发布时间】：2015-06-03 20:33:31
【问题描述】：

我正在尝试尽可能快地处理整个 csv 文件，因此我希望将每一行作为 celery 任务并行处理。清理工作也是一项 celery 任务，必须等到每一行都处理完。请参阅下面的示例。

问题是，我似乎无法通过文件，因为我一直遇到 MySQL 连接错误。到目前为止，我已经看到了这两个错误：2013, 'Lost connection to MySQL server during query' 和 2006, 'MySQL server has gone away'

from app.db.meta import Session
from celery import chord, Celery
from celery.signals import task_postrun

celery = Celery()
celery.config_from_object('config')

@task_postrun.connect
def close_session(*args, **kwargs):
    Session.remove()

def main():
    # process each line in parallel
    header = [process_line.s(line) for line in csv_file]
    # pass stats to cleanup after all lines are processed
    callback = cleanup.s()
    chord(header)(callback)

@celery.task
def process_line(line):
    session = Session()
    ...
    # process line
    ...
    return stats

@celery.task
def cleanup(stats):
    session = Session()
    ...
    # do cleanup and log stats
    ...

我正在使用 celery 3.1.18 和 SQLAlchemy 0.9.9。我也在使用连接池。

mysql> SHOW FULL PROCESSLIST;                                                                  
+----+------+-----------+-----------------+---------+------+-------+-----------------------+ 
| Id | User | Host      | db              | Command | Time | State          | Info             | 
+----+------+-----------+-----------------+---------+------+-------+-----------------------+                           
|  1 | root | localhost | ab__development | Sleep   | 4987 |       | NULL                  |                           
| 11 | root | localhost | ab__development | Sleep   | 1936 |       | NULL                  |                           
| 16 | root | localhost | ab__development | Sleep   |  143 |       | NULL                  |                           
| 17 | root | localhost | ab__development | Sleep   | 1045 |       | NULL                  |                           
| 18 | root | localhost | NULL            | Query   |    0 | init  | SHOW FULL PROCESSLIST |                                            
| 21 | root | localhost | ab__development | Sleep   |    7 |       | NULL                  |                           
+----+------+-----------+-----------------+---------+------+-------+-----------------------+                           
6 rows in set (0.01 sec)

【问题讨论】：

没有为 max_connection 设置值，所以我假设默认值为 100。
max_connections=151
我无法复制show processlist 的整个输出。但是，我看到 6 行 - 都是相同的用户和主机。其中 5 个具有相同的 db（应用程序数据库），而另一个则为 NULL。其中 5 个说 Sleep 作为命令，而另一个说 query。其中 5 个 Time 值较大，另一个为 0。
我看到 6 行，MAX(id) == 21 如果这回答了问题
这是 celery 任务运行时的输出。

标签： python mysql sqlalchemy celery

【解决方案1】：

Read the answer。简而言之，您必须禁用 SQLAlchemy's Pool engine 或尝试 ping mysql 服务器：

from flask.ext.sqlalchemy import SQLAlchemy
from sqlalchemy import event, exc


def instance(app):
    """:rtype: SQLAlchemy"""
    db = SQLAlchemy(app)

    if app.testing:
        return db

    @event.listens_for(db.engine, 'checkout')
    def checkout(dbapi_con, con_record, con_proxy):
        try:
            try:
                dbapi_con.ping(False)
            except TypeError:
                app.logger.debug('MySQL connection died. Restoring...')
                dbapi_con.ping()
        except dbapi_con.OperationalError as e:
            app.logger.warning(e)
            if e.args[0] in (2006, 2013, 2014, 2045, 2055):
                raise exc.DisconnectionError()
            else:
                raise

    return db

【讨论】：