一、MySQL与PostgreSQL对比分析 背景与概述 MySQL长期因轻量和高性能占市场主导，PostgreSQL凭借先进特性和稳定性近年快速崛起，尤其在云原生和复杂业务需求场景中表现出色。
核心技术对比 特性 MySQL PostgreSQL 数据一致性 MVCC，读已提交隔离，异步复制 完备MVCC，默认可重复读，逻辑/流复制 SQL标准兼容与扩展 支持有限，扩展围绕存储引擎 几乎完整支持SQL标准，支持丰富扩展 性能优化 读写分离，高并发读优势 并行查询，分区表，分布式支持日益完善 PostgreSQL使用度超MySQL原因 业务需求提升，复杂事务、分析需求增多 社区活跃，插件和扩展丰富 云服务快速支持，官方生态发展强劲 大型企业和专业领域采用增多 未来前景 数据库 优势场景 未来发展方向 MySQL 简单高并发读写，Web应用 高可用分布式，云原生集成 PostgreSQL 复杂事务，BI报表，地理空间 原生分布式，多模扩展，SQL标准领先 二、Python数据库操作用例 原生驱动 MySQL (PyMySQL) import pymysql conn = pymysql.connect(host='localhost', user='user', password='password', database='testdb') cursor = conn.cursor() cursor.execute("CREATE TABLE IF NOT EXISTS users (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(100), email VARCHAR(100) UNIQUE)") cursor.execute("INSERT INTO users (name, email) VALUES (%s, %s)", ('Alice', '[email protected]')) conn.commit() cursor.execute("SELECT * FROM users") print(cursor.fetchall()) cursor.close() conn.close() PostgreSQL (psycopg2) import psycopg2 conn = psycopg2.connect(host='localhost', user='user', password='password', dbname='testdb') cursor = conn.cursor() cursor.execute("CREATE TABLE IF NOT EXISTS users (id SERIAL PRIMARY KEY, name TEXT, email TEXT UNIQUE)") cursor.execute("INSERT INTO users (name, email) VALUES (%s, %s)", ('Bob', '[email protected]')) conn.commit() cursor.execute("SELECT * FROM users") print(cursor.fetchall()) cursor.close() conn.close() ORM 示例：SQLAlchemy from sqlalchemy import create_engine, Column, Integer, String from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker Base = declarative_base() class User(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True) name = Column(String(100)) email = Column(String(100), unique=True) DATABASE_URL = 'postgresql+psycopg2://user:password@localhost:5432/testdb' engine = create_engine(DATABASE_URL) Session = sessionmaker(bind=engine) session = Session() Base.metadata.create_all(engine) new_user = User(name='Carol', email='[email protected]') session.add(new_user) session.commit() users = session.query(User).all() for u in users: print(u.id, u.name, u.email) session.close() 三、数据库压测方案 压测总体流程 环境准备：部署数据库和测试客户端 场景设计：包含OLTP、高并发读写、复杂查询等 基线测试：默认配置性能测量 参数调优：调整配置重复测试 结果分析：整理吞吐、延迟和资源利用数据 自动化脚本：支持快速复用 主要压测工具与命令示例 Sysbench（MySQL） sysbench oltp_read_write \ --db-driver=mysql \ --mysql-host=DB_HOST \ --mysql-user=testuser \ --mysql-password=secret \ --mysql-db=testdb \ --tables=10 \ --table-size=1000000 \ prepare sysbench oltp_read_write --threads=100 --time=300 run sysbench oltp_read_write cleanup pgbench（PostgreSQL） pgbench -h DB_HOST -U testuser -d testdb -i -s 1000 pgbench -h DB_HOST -U testuser -d testdb -c 100 -T 300 -j 4 自定义Python异步压测示例 import asyncio import aiomysql DB_CONFIG = dict(host='DB_HOST', user='testuser', password='secret', db='testdb', minsize=10, maxsize=100) async def task(pool): async with pool.acquire() as conn: async with conn.cursor() as cur: await cur.execute("SELECT COUNT(*) FROM orders WHERE status='pending'") await cur.fetchone() await cur.execute("UPDATE users SET last_login=NOW() WHERE id=%s", (1,)) async def run_load(concurrency, duration): pool = await aiomysql.create_pool(**DB_CONFIG) end_time = asyncio.get_event_loop().time() + duration sem = asyncio.Semaphore(concurrency) async def worker(): async with sem: while asyncio.get_event_loop().time() < end_time: await task(pool) await asyncio.gather(*[worker() for _ in range(concurrency)]) pool.close() await pool.wait_closed() asyncio.run(run_load(concurrency=50, duration=300)) 四、一体化压测脚本模板 #!/bin/bash DB_TYPE=${1:-mysql} DB_HOST="127.0.0.1" DB_PORT_MYSQL=3306 DB_PORT_PG=5432 DB_USER="testuser" DB_PASS="secret" DB_NAME="testdb" CONCURRENCY=(10 50 100) DURATION=120 SCALE=100 function bench_mysql() { sysbench oltp_read_write \ --threads=$1 \ --time=$DURATION \ --db-driver=mysql \ --mysql-host=$DB_HOST \ --mysql-port=$DB_PORT_MYSQL \ --mysql-user=$DB_USER \ --mysql-password=$DB_PASS \ --mysql-db=$DB_NAME run | tee mysql_${1}c.log } function bench_pgsql() { pgbench -h $DB_HOST -p $DB_PORT_PG -U $DB_USER -d $DB_NAME -c $1 -T $DURATION -j $(nproc) | tee pg_${1}c.log } if [[ $DB_TYPE == "pgsql" ]]; then pgbench -h $DB_HOST -p $DB_PORT_PG -U $DB_USER -d $DB_NAME -i -s $SCALE fi for c in "${CONCURRENCY[@]}"; do if [[ $DB_TYPE == "mysql" ]]; then bench_mysql $c else bench_pgsql $c fi done echo "压测完成，查看 *_c.log 文件" 五、压测结果分析与扩展建议 提取日志中的TPS、延迟等数据，生成CSV 使用图表工具（Excel，Grafana，Matplotlib）绘制性能曲线 根据业务需求调整读写比、复杂查询和分布式架构测试 可集成监控系统，实现实时资源指标采样 以上内容系统梳理了MySQL与PostgreSQL的技术对比，Python数据库操作示例，以及一整套可执行的数据库压测方案与脚本，便于用户快速搭建测试环境，评估性能，指导优化与选型。