Python3 pika channel.basic_consume() 导致 MySQL 连接过多

Question

我曾使用 pika 与 RabbitMQ 建立连接并使用消息，一旦我在 ubuntu prod 环境中启动脚本，它就会按预期工作，但正在打开 mysql 连接并且从不关闭它们并最终导致 mysql 服务器上的连接过多.

将感谢对以下代码的任何建议，以及无法理解出了什么问题。提前谢谢你。

流程如下

1. 在 Python3 上启动 pika
2. 订阅频道并等待消息
3. 在回调中，我进行各种验证并在 MySql 中保存或更新数据
4. 显示问题的结果是问题结尾处来自 ubuntu htop 的屏幕截图，显示 MySql 上的新连接并继续在顶部添加它们

鼠兔版本 = 0.13.0

对于 MySql，我使用 pymysql。

鼠兔脚本

def main():
    credentials = pika.PlainCredentials(tunnel['queue']['name'], tunnel['queue']['password'])

    while True:
        try:
            cp = pika.ConnectionParameters(
                host=tunnel['queue']['host'],
                port=tunnel['queue']['port'],
                credentials=credentials,
                ssl=tunnel['queue']['ssl'],
                heartbeat=600,
                blocked_connection_timeout=300
            )

            connection = pika.BlockingConnection(cp)
            channel = connection.channel()

            def callback(ch, method, properties, body):
                if 'messageType' in properties.headers:
                    message_type = properties.headers['messageType']

                    if events.get(message_type):
                        result = Descriptors._reflection.ParseMessage(events[message_type]['decode'], body)
                        if result:
                            result = protobuf_to_dict(result)
                            model.write_response(external_response=result, message_type=message_type)
                    else:
                        app_log.warning('Message type not in allowed list = ' + str(message_type))
                        app_log.warning('continue listening...')

            channel.basic_consume(callback, queue=tunnel['queue']['name'], no_ack=True)
            try:
                channel.start_consuming()
            except KeyboardInterrupt:
                channel.stop_consuming()
                connection.close()
                break
        except pika.connection.exceptions.ConnectionClosed as e:
            app_log.error('ConnectionClosed :: %s' % str(e))
            continue
        except pika.connection.exceptions.AMQPChannelError as e:
            app_log.error('AMQPChannelError :: %s' % str(e))
            continue
        except Exception as e:
            app_log.error('Connection was closed, retrying... %s' % str(e))
            continue


if __name__ == '__main__':
    main()

在脚本中，我有一个在数据库中进行插入或更新的模型，代码如下

def write_response(self, external_response, message_type):
    table_name = events[message_type]['table_name']
    original_response = external_response[events[message_type]['response']]
    if isinstance(original_response, list):
        external_response = []
        for o in original_response:
            record = self.map_keys(o, message_type, events[message_type].get('values_fix', {}))
            external_response.append(self.validate_fields(record))
    else:
        external_response = self.map_keys(original_response, message_type, events[message_type].get('values_fix', {}))
        external_response = self.validate_fields(external_response)

    if not self.mysql.open:
        self.mysql.ping(reconnect=True)

    with self.mysql.cursor() as cursor:
        if isinstance(original_response, list):
            for e in external_response:
                id_name = events[message_type]['id_name']
                filters = {id_name: e[id_name]}
                self.event(
                    cursor=cursor,
                    table_name=table_name,
                    filters=filters,
                    external_response=e,
                    message_type=message_type,
                    event_id=e[id_name],
                    original_response=e  # not required here
                )
        else:
            id_name = events[message_type]['id_name']
            filters = {id_name: external_response[id_name]}
            self.event(
                cursor=cursor,
                table_name=table_name,
                filters=filters,
                external_response=external_response,
                message_type=message_type,
                event_id=external_response[id_name],
                original_response=original_response
            )
    cursor.close()
    self.mysql.close()

    return

在 ubuntu 上我使用 systemd 运行脚本并在出现问题时重新启动，下面是 systemd 文件

[Unit]
Description=Pika Script
Requires=stunnel4.service
Requires=mysql.service
Requires=mongod.service

[Service]
User=user
Group=group
WorkingDirectory=/home/pika_script
ExecStart=/home/user/venv/bin/python pika_script.py
Restart=always

[Install]
WantedBy=multi-user.target

来自 ubuntu htop 的图片，MySql 如何不断在列表中添加并且从不关闭它

错误

tornado_mysql.err.OperationalError: (1040, 'Too many connections')

Answer 1

我找到了问题，发布如果对其他人有帮助。

问题是 mysqld 进入无限循环，试图为特定数据库创建索引，在发现哪个数据库试图创建索引并且从未成功并且一次又一次地尝试之后。

解决方案是删除数据库并重新创建它，mysqld 进程恢复正常。并且创建索引的无限循环也消失了。

Answer 2

我会说增加连接可能会温和地解决您的问题。

首先找出应用程序在任务完成后没有关闭连接的原因。

第二个数据库上的任何慢查询/调用，如果有的话，修复它们。

第三考虑到数据库上没有缓慢的查询/调用，并且应用程序在立即完成任务后关闭连接/线程，然后考虑在 mysql 端使用“wait_timeout”。

Answer 3

根据this的回答，如果你有 MySQL 5.7 和 5.8：

值得知道的是，如果您的可用磁盘空间用完 服务器分区或驱动器，这也会导致 MySQL 返回 这个错误。如果您确定这不是实际的用户数 已连接，那么下一步是检查您是否有可用空间 您的 MySQL 服务器驱动器/分区。

来自同一个线程。您可以检查和增加 MySQL 连接数。