MySQL与PostgreSQL全面对比与压测方案

Posted on Sun 15 December 2024 in 数据库

一、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()

三、数据库压测方案

压测总体流程

  1. 环境准备：部署数据库和测试客户端
  2. 场景设计：包含OLTP、高并发读写、复杂查询等
  3. 基线测试：默认配置性能测量
  4. 参数调优：调整配置重复测试
  5. 结果分析：整理吞吐、延迟和资源利用数据
  6. 自动化脚本：支持快速复用

主要压测工具与命令示例

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数据库操作示例，以及一整套可执行的数据库压测方案与脚本，便于用户快速搭建测试环境，评估性能，指导优化与选型。

MySQL PostgreSQL 数据库对比 压测 Python 性能优化

Previous Post Next Post

You might enjoy