MySQL MCP 使用案例
## 概述
MySQL MCP（MySQL Multi-Channel Protocol）是MySQL的多通道协议实现，提供了高效的数据库连接池和负载均衡功能。本文档将介绍MySQL MCP的基本使用方法和常见案例。
## 环境准备
### 安装MySQL MCP
```bash
pip install mysql-mcp
```
### 基本配置
创建配置文件 `mcp_config.json`:
```json
{
"master": {
"host": "主数据库IP",
"port": 3306,
"user": "用户名",
"password": "密码",
"database": "数据库名"
},
"slaves": [
{
"host": "从数据库1IP",
"port": 3306,
"user": "用户名",
"password": "密码",
"database": "数据库名"
},
{
"host": "从数据库2IP",
"port": 3306,
"user": "用户名",
"password": "密码",
"database": "数据库名"
}
],
"connection_pool": {
"min_connections": 5,
"max_connections": 20,
"idle_timeout": 300
}
}
```
## 基本使用案例
### 案例1: 连接数据库
```python
from mysql_mcp import ConnectionPool
# 初始化连接池
pool = ConnectionPool.from_config("mcp_config.json")
# 获取连接
connection = pool.get_connection()
try:
# 使用连接
with connection.cursor() as cursor:
cursor.execute("SELECT VERSION()")
version = cursor.fetchone()
print(f"数据库版本: {version[0]}")
finally:
# 归还连接到连接池
connection.close()
```
### 案例2: 读写分离
```python
from mysql_mcp import ConnectionPool
pool = ConnectionPool.from_config("mcp_config.json")
# 写操作 - 使用主库
def insert_data(name, age):
connection = pool.get_master_connection()
try:
with connection.cursor() as cursor:
sql = "INSERT INTO users (name, age) VALUES (%s, %s)"
cursor.execute(sql, (name, age))
connection.commit()
finally:
connection.close()
# 读操作 - 使用从库
def get_user(user_id):
connection = pool.get_slave_connection()
try:
with connection.cursor() as cursor:
sql = "SELECT * FROM users WHERE id = %s"
cursor.execute(sql, (user_id,))
return cursor.fetchone()
finally:
connection.close()
# 使用示例
insert_data("张三", 25)
user = get_user(1)
print(user)
```
### 案例3: 事务处理
```python
from mysql_mcp import ConnectionPool
pool = ConnectionPool.from_config("mcp_config.json")
def transfer_money(from_account, to_account, amount):
connection = pool.get_master_connection()
try:
connection.begin()
with connection.cursor() as cursor:
# 检查余额
cursor.execute("SELECT balance FROM accounts WHERE id = %s FOR UPDATE", (from_account,))
from_balance = cursor.fetchone()[0]
if from_balance < amount:
raise Exception("余额不足")
# 更新转出账户
cursor.execute("UPDATE accounts SET balance = balance - %s WHERE id = %s",
(amount, from_account))
# 更新转入账户
cursor.execute("UPDATE accounts SET balance = balance + %s WHERE id = %s",
(amount, to_account))
connection.commit()
return True
except Exception as e:
connection.rollback()
print(f"转账失败: {e}")
return False
finally:
connection.close()
```
### 案例4: 批量操作
```python
from mysql_mcp import ConnectionPool
pool = ConnectionPool.from_config("mcp_config.json")
def batch_insert(users):
connection = pool.get_master_connection()
try:
with connection.cursor() as cursor:
sql = "INSERT INTO users (name, age, email) VALUES (%s, %s, %s)"
cursor.executemany(sql, users)
connection.commit()
print(f"成功插入 {len(users)} 条记录")
finally:
connection.close()
# 批量插入示例
users_data = [
("李四", 30, "lisi@example.com"),
("王五", 25, "wangwu@example.com"),
("赵六", 35, "zhaoliu@example.com")
]
batch_insert(users_data)
```
### 案例5: 连接池监控
```python
from mysql_mcp import ConnectionPool
pool = ConnectionPool.from_config("mcp_config.json")
# 获取连接池状态
def get_pool_status():
status = pool.get_status()
print(f"总连接数: {status['total_connections']}")
print(f"活跃连接数: {status['active_connections']}")
print(f"空闲连接数: {status['idle_connections']}")
print(f"等待连接数: {status['waiting_connections']}")
return status
# 使用示例
get_pool_status()
```
## 高级用法
### 自定义负载均衡策略
```python
from mysql_mcp import ConnectionPool, LoadBalancer
class CustomLoadBalancer(LoadBalancer):
def select_slave(self, slaves):
# 自定义选择从库的逻辑
# 例如: 根据从库的响应时间来选择
return min(slaves, key=lambda slave: slave.response_time)
# 使用自定义负载均衡器
pool = ConnectionPool.from_config("mcp_config.json", load_balancer=CustomLoadBalancer())
```
### 故障转移处理
```python
from mysql_mcp import ConnectionPool, FailoverStrategy
# 配置故障转移策略
config = {
"failover": {
"retry_attempts": 3,
"retry_delay": 1,
"auto_reconnect": True
}
}
pool = ConnectionPool.from_config("mcp_config.json", failover_strategy=FailoverStrategy(**config["failover"]))
# 带有故障转移的查询
def query_with_failover(sql, params=None):
retries = 0
while retries < 3:
try:
connection = pool.get_connection()
try:
with connection.cursor() as cursor:
cursor.execute(sql, params)
return cursor.fetchall()
finally:
connection.close()
except Exception as e:
retries += 1
if retries >= 3:
raise Exception(f"查询失败，已重试3次: {e}")
print(f"查询失败，正在重试 ({retries}/3)")
```
## 性能优化建议
1. **合理设置连接池大小**：根据服务器性能和负载情况调整最小和最大连接数。
2. **监控连接使用情况**：定期检查连接池状态，避免连接泄漏。
3. **设置合理的超时时间**：防止长时间未使用的连接占用资源。
4. **使用预编译语句**：对于频繁执行的SQL语句，使用预编译语句可以提高性能。
## 总结
MySQL MCP提供了高效的数据库连接池管理和读写分离功能，通过以上案例可以看出，使用MySQL MCP可以显著提高数据库操作的性能和稳定性。在实际应用中，可以根据具体需求进行配置和优化，以达到最佳的使用效果。