python - mysql 连接与 python 中的并发.futures 不能很好地配合

Question

我正在尝试来自 concurrent.futures 的 ThreadPoolExecutor

import os
import mysql.connector
import concurrent.futures

db = mysql.connector.connect(user=mysql_user, database='wallet', password=mysql_password, port='5009', host='localhost')

def work(i):
    cursor = db.cursor()
    cursor.execute('show tables;')
    return f'{i} done'

with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
    future_to_uid = {executor.submit(work, i): i for i in range(5)}
    for future in concurrent.futures.as_completed(future_to_uid):
        uid = future_to_uid[future]
        try:
            data = future.result()
        except Exception as exc:
            print('%r generated an exception: %s' % (uid, exc))
        else:
            print('status', data)

当我运行这个时，每次运行都会出现不同的错误。

alok@alok-HP-Laptop-14s-cr1xxx:~/tmp$ python3 threadmysql.py
0 generated an exception: bytearray index out of range
1 generated an exception: MySQL Connection not available.
3 generated an exception: MySQL Connection not available.
4 generated an exception: MySQL Connection not available.
2 generated an exception: MySQL Connection not available.

alok@alok-HP-Laptop-14s-cr1xxx:~/tmp$ python3 threadmysql.py
status 0 done
malloc(): corrupted top size
Aborted (core dumped)

alok@alok-HP-Laptop-14s-cr1xxx:~/tmp$ python3 threadmysql.py
0 generated an exception: bytearray index out of range
python3: malloc.c:2392: sysmalloc: Assertion `(old_top == initial_top (av) && old_size == 0) || ((unsigned long) (old_size) >= MINSIZE && prev_inuse (old_top) && ((unsigned long) old_end & (pagesize - 1)) == 0)' failed.
Aborted (core dumped)

为什么我会收到错误并且每次都不同？

Answer 1

使用mysql连接池正在解决问题。如何创建池在
中给出 https://dev.mysql.com/doc/connector-python/en/connector-python-connection-pooling.html

import os
import mysql.connector
import concurrent.futures

POOL_SIZE = 5

db = mysql.connector.connect(user=mysql_user, database='wallet', password=mysql_password, port='5009', host='localhost', pool_name = "mypool", pool_size = POOL_SIZE+1)

def work(i):
    db = mysql.connector.connect(pool_name = "mypool")
    cursor = db.cursor()
    cursor.execute('show tables;')
    db.close()
    return f'{i} done'

with concurrent.futures.ThreadPoolExecutor(max_workers=POOL_SIZE) as executor:
    future_to_uid = {executor.submit(work, i): i for i in range(5)}
    for future in concurrent.futures.as_completed(future_to_uid):
        uid = future_to_uid[future]
        try:
            data = future.result()
        except Exception as exc:
            print('%r generated an exception: %s' % (uid, exc))
        else:
            print('status', data)

或者此代码的简单版本是

import mysql.connector
import concurrent.futures

POOL_SIZE = 5
db = mysql.connector.connect(user=mysql_user, database='wallet', password=mysql_password, port='5009', host='localhost', pool_name = "mypool", pool_size = POOL_SIZE+1)

def work(i):
    db = mysql.connector.connect(pool_name = "mypool")
    cursor = db.cursor()
    cursor.execute('show tables;')
    db.close()
    return f'{i} done'

executor = concurrent.futures.ThreadPoolExecutor(max_workers=POOL_SIZE)
for i in range(1, 100):
    executor.submit(work, i)

print('done')