面试题：Python优化MySQL数据库高并发读写操作

优化数据库读写性能的方法

连接池的使用：
- 作用：减少每次建立和关闭数据库连接的开销，提高并发性能。在高并发场景下，频繁创建和销毁数据库连接会消耗大量资源，连接池可以复用已有的连接，降低资源消耗。
- 选择合适的连接池：在Python中，对于MySQL，SQLAlchemy自带连接池，aiomysql也有异步连接池。SQLAlchemy的连接池适用于同步和异步混合编程场景，而aiomysql的连接池专门用于异步编程场景。
SQL语句的优化：
- 索引优化：为经常用于查询条件、排序、连接的字段添加合适的索引。例如，如果经常按照user_id查询用户信息，就在user_id字段上添加索引。索引可以加快查询速度，但过多的索引也会增加插入、更新和删除操作的开销，因为每次数据变更都需要更新索引。
- 避免全表扫描：确保SQL语句能够利用索引，避免全表扫描。比如查询时尽量使用WHERE子句限定条件，并且条件字段要有索引。例如SELECT * FROM users WHERE age > 18;，如果age字段有索引，查询效率会大大提高。
- 优化复杂查询：对于复杂的多表连接查询，合理调整连接顺序，确保数据量小的表先连接。例如在SELECT * FROM orders JOIN users ON orders.user_id = users.id;中，如果users表数据量较小，应优先连接users表。同时，避免子查询嵌套过深，可以使用连接代替子查询来提高查询效率。
数据库架构设计：
- 读写分离：将读操作和写操作分离到不同的数据库服务器上。主库负责写操作，从库负责读操作。这样可以减轻主库的压力，提高读性能。在高并发读场景下，多个从库可以分担读请求，提高系统的整体吞吐量。例如，使用MySQL的主从复制功能实现读写分离。
- 分库分表：当数据量过大时，将数据库按照业务模块、数据范围等进行拆分。水平分表可以按照某个字段（如时间、用户ID等）将数据分散到多个表中，垂直分库则是按照业务功能将不同模块的数据存储到不同的数据库中。比如按照时间将订单表按月分表，或者将用户信息、订单信息分别存储在不同的数据库中。

使用SQLAlchemy实现高并发读写的示例代码框架

import asyncio
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import QueuePool

# 创建数据库连接池
engine = create_engine(
    "mysql+asyncmy://user:password@host:port/database",
    poolclass=QueuePool,
    pool_size=10,
    max_overflow=20
)

# 创建异步会话工厂
AsyncSession = sessionmaker(engine, class_=asyncio.AsyncSession)


async def async_write():
    async with AsyncSession() as session:
        try:
            # 示例插入语句
            query = text("INSERT INTO users (name, age) VALUES (:name, :age)")
            await session.execute(query, {"name": "test_user", "age": 25})
            await session.commit()
        except Exception as e:
            await session.rollback()
            print(f"写入错误: {e}")


async def async_read():
    async with AsyncSession() as session:
        try:
            # 示例查询语句
            query = text("SELECT * FROM users WHERE age > :age")
            result = await session.execute(query, {"age": 18})
            rows = result.fetchall()
            for row in rows:
                print(row)
        except Exception as e:
            print(f"读取错误: {e}")


async def main():
    tasks = []
    for _ in range(10):
        tasks.append(async_write())
        tasks.append(async_read())
    await asyncio.gather(*tasks)


if __name__ == "__main__":
    asyncio.run(main())

使用aiomysql实现高并发读写的示例代码框架

import asyncio
import aiomysql


async def create_pool():
    pool = await aiomysql.create_pool(
        host='host',
        port=3306,
        user='user',
        password='password',
        db='database',
        autocommit=True,
        maxsize=10,
        minsize=1
    )
    return pool


async def async_write(pool):
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            try:
                # 示例插入语句
                await cur.execute("INSERT INTO users (name, age) VALUES (%s, %s)", ("test_user", 25))
            except Exception as e:
                print(f"写入错误: {e}")


async def async_read(pool):
    async with pool.acquire() as conn:
        async with conn.cursor(aiomysql.DictCursor) as cur:
            try:
                # 示例查询语句
                await cur.execute("SELECT * FROM users WHERE age > %s", (18,))
                rows = await cur.fetchall()
                for row in rows:
                    print(row)
            except Exception as e:
                print(f"读取错误: {e}")


async def main():
    pool = await create_pool()
    tasks = []
    for _ in range(10):
        tasks.append(async_write(pool))
        tasks.append(async_read(pool))
    await asyncio.gather(*tasks)
    pool.close()
    await pool.wait_closed()


if __name__ == "__main__":
    asyncio.run(main())

面试题：Python优化MySQL数据库高并发读写操作

知识考点

面试题答案

优化数据库读写性能的方法

使用SQLAlchemy实现高并发读写的示例代码框架

使用aiomysql实现高并发读写的示例代码框架