PostgreSQL MCP 使用案例
## 概述
PostgreSQL MCP（PostgreSQL Multi-host Cluster Provisioning）是一种用于部署和管理多节点PostgreSQL集群的工具和架构。它提供了高效的数据库集群管理、高可用性保障和负载均衡功能。本文档将介绍PostgreSQL MCP的基本使用方法和常见应用场景。
## 环境准备
### 安装PostgreSQL MCP
```bash
pip install pg-mcp
```
### 基本配置
创建配置文件 `pgmcp_config.json`:
```json
{
"masters": [
{
"host": "主数据库1 IP",
"port": 5432,
"user": "postgres",
"password": "密码",
"database": "数据库名"
},
{
"host": "主数据库2 IP",
"port": 5432,
"user": "postgres",
"password": "密码",
"database": "数据库名"
}
],
"replicas": [
{
"host": "只读副本1 IP",
"port": 5432,
"user": "postgres",
"password": "密码",
"database": "数据库名"
},
{
"host": "只读副本2 IP",
"port": 5432,
"user": "postgres",
"password": "密码",
"database": "数据库名"
}
],
"connection_pool": {
"min_connections": 5,
"max_connections": 20,
"idle_timeout": 300
},
"high_availability": {
"failover_timeout": 30,
"max_retry_attempts": 3,
"enable_auto_failover": true
}
}
```
## 基本使用案例
### 案例1: 连接数据库集群
```python
from pg_mcp import ConnectionPool
# 初始化连接池
pool = ConnectionPool.from_config("pgmcp_config.json")
# 获取连接
connection = pool.get_connection()
try:
# 使用连接
with connection.cursor() as cursor:
cursor.execute("SELECT version()")
version = cursor.fetchone()
print(f"PostgreSQL 版本: {version[0]}")
finally:
# 归还连接到连接池
connection.close()
```
### 案例2: 读写分离
```python
from pg_mcp import ConnectionPool
pool = ConnectionPool.from_config("pgmcp_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_replica_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: 事务处理与MVCC优化
```python
from pg_mcp import ConnectionPool
pool = ConnectionPool.from_config("pgmcp_config.json")
def transfer_money(from_account, to_account, amount):
connection = pool.get_master_connection()
try:
# PostgreSQL默认是事务模式，不需要显式begin
with connection.cursor() as cursor:
# 检查余额 - 使用FOR UPDATE避免并发问题
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: 批量操作与COPY命令
```python
from pg_mcp import ConnectionPool
import io
import csv
pool = ConnectionPool.from_config("pgmcp_config.json")
# 使用executemany进行批量插入
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()
# 使用PostgreSQL的COPY命令进行大批量数据导入（性能更佳）
def bulk_copy(users):
connection = pool.get_master_connection()
try:
# 准备CSV数据
csv_data = io.StringIO()
csv_writer = csv.writer(csv_data)
for user in users:
csv_writer.writerow(user)
csv_data.seek(0)
with connection.cursor() as cursor:
cursor.copy_from(csv_data, 'users', sep=',', columns=('name', 'age', 'email'))
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 pg_mcp import ConnectionPool
pool = ConnectionPool.from_config("pgmcp_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
# 监控复制延迟
def check_replication_lag():
connection = pool.get_master_connection()
try:
with connection.cursor() as cursor:
cursor.execute("""
SELECT client_addr, state, sent_lsn, write_lsn,
pg_wal_lsn_diff(sent_lsn, write_lsn) AS lag_bytes
FROM pg_stat_replication
""")
return cursor.fetchall()
finally:
connection.close()
# 使用示例
get_pool_status()
lag_info = check_replication_lag()
for replica in lag_info:
print(f"复制节点: {replica[0]}, 状态: {replica[1]}, 延迟: {replica[4]} 字节")
```
## 高级用法
### 自定义负载均衡策略
```python
from pg_mcp import ConnectionPool, LoadBalancer
class CustomLoadBalancer(LoadBalancer):
def select_replica(self, replicas):
# 自定义选择副本的逻辑
# 例如: 根据副本的负载情况来选择
return min(replicas, key=lambda replica: replica.current_load)
# 使用自定义负载均衡器
pool = ConnectionPool.from_config("pgmcp_config.json", load_balancer=CustomLoadBalancer())
```
### 故障转移与自动恢复
```python
from pg_mcp import ConnectionPool, FailoverStrategy
# 配置故障转移策略
config = {
"failover": {
"check_interval": 5,
"max_retry_attempts": 3,
"retry_delay": 1,
"promote_replica": True
}
}
pool = ConnectionPool.from_config("pgmcp_config.json", failover_strategy=FailoverStrategy(**config["failover"]))
# 带有故障转移的查询执行
def execute_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)")
```
### 使用PostgreSQL特有功能
```python
from pg_mcp import ConnectionPool
pool = ConnectionPool.from_config("pgmcp_config.json")
# 使用JSON数据类型
def store_json_data(user_id, preferences):
connection = pool.get_master_connection()
try:
with connection.cursor() as cursor:
# PostgreSQL支持直接存储JSON数据
cursor.execute(
"INSERT INTO user_preferences (user_id, preferences) VALUES (%s, %s::jsonb)",
(user_id, json.dumps(preferences))
)
connection.commit()
finally:
connection.close()
# 使用全文搜索
def search_products(query):
connection = pool.get_replica_connection()
try:
with connection.cursor() as cursor:
cursor.execute("""
SELECT id, name, description
FROM products
WHERE to_tsvector('chinese', name || ' ' || description) @@ plainto_tsquery('chinese', %s)
ORDER BY ts_rank(to_tsvector('chinese', name), plainto_tsquery('chinese', %s)) DESC
""", (query, query))
return cursor.fetchall()
finally:
connection.close()
```
## 性能优化建议
1. **合理设置连接池大小**：根据服务器性能和负载情况调整最小和最大连接数。PostgreSQL默认max_connections为100，应避免连接池总大小超过此值。
2. **使用prepared语句**：对于频繁执行的SQL，使用prepared语句可以减少解析开销。
```python
connection = pool.get_connection()
try:
with connection.cursor() as cursor:
cursor.execute("PREPARE get_user AS SELECT * FROM users WHERE id = $1")
cursor.execute("EXECUTE get_user(%s)", (user_id,))
result = cursor.fetchone()
finally:
connection.close()
```
3. **适当配置PostgreSQL参数**：
- `shared_buffers`: 通常设置为系统内存的25%
- `work_mem`: 调整排序和哈希操作的内存使用
- `maintenance_work_mem`: 提高VACUUM等维护操作性能
- `effective_cache_size`: 设置为系统可用缓存的估计值
4. **启用连接池状态监控**：定期检查连接池状态，避免连接泄漏和资源耗尽。
5. **利用PostgreSQL并行查询**：对于大表查询，启用并行查询可提高性能。
```sql
SET max_parallel_workers_per_gather = 4;
```
## 总结
PostgreSQL MCP提供了强大的数据库集群管理、高可用性和读写分离功能。通过合理配置和使用MCP，可以显著提高PostgreSQL数据库的性能、可靠性和可扩展性。特别是利用PostgreSQL的高级特性（如JSONB支持、全文搜索和MVCC并发控制），能够构建功能丰富且高效的应用系统。
在实际应用中，应根据具体业务需求和系统负载情况，调整PostgreSQL MCP的配置参数，以达到最佳的使用效果。定期的性能监控和维护也是保障系统稳定运行的关键因素。