37、aiomysql实操习题

练习题1：慢查询优化

题目描述

将以下低效查询优化为索引查询：

# 原始低效查询
await cursor.execute("SELECT * FROM orders WHERE YEAR(created_at)=2023")

参考答案

# 优化后查询（使用索引范围扫描）
await cursor.execute("SELECT * FROM orders WHERE created_at >= %s AND created_at < %s",('2023-01-01 00:00:00', '2024-01-01 00:00:00')
)

优化原理：

1. 避免在列上使用函数（YEAR()），防止索引失效
2. 使用created_at字段的索引进行范围扫描
3. 提前创建索引：

ALTER TABLE orders ADD INDEX idx_created_at (created_at);

---

练习题2：死锁复现与解决

题目描述

编写两个并发事务，模拟资金转账场景下的死锁，并实现自动重试机制

参考答案

async def transfer_with_retry(pool, from_id, to_id, amount):async def _execute_transfer():async with pool.acquire() as conn:await conn.begin()try:async with conn.cursor() as cursor:# 故意制造死锁顺序if from_id < to_id:await cursor.execute("SELECT * FROM accounts WHERE user_id=%s FOR UPDATE", (from_id))await cursor.execute("SELECT * FROM accounts WHERE user_id=%s FOR UPDATE", (to_id))else:await cursor.execute("SELECT * FROM accounts WHERE user_id=%s FOR UPDATE", (to_id))await cursor.execute("SELECT * FROM accounts WHERE user_id=%s FOR UPDATE", (from_id))# 实际转账操作await cursor.execute("UPDATE accounts SET balance=balance-%s WHERE user_id=%s", (amount, from_id))await cursor.execute("UPDATE accounts SET balance=balance+%s WHERE user_id=%s", (amount, to_id))await conn.commit()except Exception as e:await conn.rollback()raise# 死锁自动重试            return await retry_on_deadlock(_execute_transfer, max_retries=3)# 测试死锁（并发执行相反顺序转账）
async def test_deadlock():pool = await create_pool()task1 = transfer_with_retry(pool, 1, 2, 100)task2 = transfer_with_retry(pool, 2, 1, 50)await asyncio.gather(task1, task2)

关键机制：

1. 通过FOR UPDATE显式加锁
2. 使用不同的锁顺序触发死锁
3. 重试装饰器捕获错误码1213（死锁）

---

练习题3：连接泄漏检测

题目描述

找出以下代码中的连接泄漏问题并修复：

async def leaky_function(pool):conn = await pool.acquire()cursor = await conn.cursor()await cursor.execute("SELECT * FROM users")result = await cursor.fetchall()await cursor.close()return result

参考答案

async def fixed_function(pool):async with pool.acquire() as conn:  # 自动释放连接async with conn.cursor() as cursor:  # 自动关闭游标await cursor.execute("SELECT * FROM users")return await cursor.fetchall()

修复要点：

1. 使用async with上下文管理器确保连接释放

2. 完整连接生命周期管理：

3. 监控代码示例：

async def check_leaks(pool):print(f"使用中的连接: {pool.size - pool.freesize}")if pool.size > pool.maxsize * 0.8:print("警告: 连接池使用率超过80%!")

---

练习题4：分页查询优化

题目描述

将传统分页改为游标分页：

# 原始分页
await cursor.execute("SELECT * FROM orders ORDER BY id LIMIT 10 OFFSET 20"
)

参考答案

# 优化分页（基于游标）
last_id = 100  # 上一页最后记录的ID
await cursor.execute("SELECT * FROM orders WHERE id > %s ORDER BY id LIMIT 10",(last_id,)
)

优势分析：

1. 避免OFFSET带来的全表扫描
2. 分页效率从O(N)降至O(1)
3. 适合无限滚动加载场景

---

练习题5：批量更新优化

题目描述

将循环单条更新改为批量操作：

# 低效写法
for order in orders:await cursor.execute("UPDATE orders SET status=%s WHERE id=%s",(new_status, order['id']))

参考答案

# 高效批量更新
update_data = [(o['status'], o['id']) for o in orders]
await cursor.executemany("UPDATE orders SET status=%s WHERE id=%s",update_data
)
await conn.commit()

性能对比：

数据量	单条提交耗时	批量提交耗时
1000条	4.2s	0.8s
5000条	21.5s	3.1s

---

综合挑战：订单系统完整实现

class AsyncOrderSystem:def __init__(self, pool):self.pool = poolasync def create_order(self, user_id, items):async with self.pool.acquire() as conn:await conn.begin()try:async with conn.cursor() as cursor:# 创建订单await cursor.execute("INSERT INTO orders (user_id, status) VALUES (%s, 'pending')",(user_id,))order_id = cursor.lastrowid# 插入订单明细await cursor.executemany("INSERT INTO order_items (order_id, product_id, quantity) VALUES (%s, %s, %s)",[(order_id, item['id'], item['qty']) for item in items])# 扣减库存await cursor.executemany("UPDATE products SET stock=stock-%s WHERE id=%s AND stock >= %s",[(item['qty'], item['id'], item['qty']) for item in items])await conn.commit()return order_idexcept Exception as e:await conn.rollback()raise

实现要点：

1. 使用事务保证订单创建的原子性
2. 批量操作提升性能
3. 库存检查前置避免超卖

---

以上练习题覆盖了异步数据库操作的核心难点，建议在开发环境中实际运行并观察效果。可通过以下方式验证结果：

# 性能测试
async def test_performance():pool = await create_pool()# 执行10次查询取平均值start = time.perf_counter()for _ in range(10):await fixed_function(pool)print(f"平均耗时: {(time.perf_counter()-start)/10:.4f}s")# 死锁检测
async def test_deadlock_detection():with pytest.raises(ValueError) as excinfo:await test_deadlock()assert "deadlock" in str(excinfo.value).lower()