PostgreSQL锁机制测试及锁统计报告工具咨询

针对你要测试1000次并行请求下各类锁组合的需求，以下几种工具和方法可以生成包含锁等待时长、锁定行信息、锁开销的统计报告：

一、PostgreSQL内置工具（无需额外安装） #

1. 系统视图实时统计 #

通过pg_stat_activity和pg_locks这两个系统视图的组合查询，能实时抓取锁相关数据。可以写个脚本循环执行查询，收集1000次请求期间的信息，最后汇总统计：
核心查询示例：

SELECT
  pid,
  usename,
  query,
  locktype,
  mode,
  relation::regclass AS table_name,
  page,
  tuple,
  now() - query_start AS query_duration,
  wait_event_type,
  wait_event
FROM pg_stat_activity
JOIN pg_locks ON pg_stat_activity.pid = pg_locks.pid
WHERE query LIKE '%for update%'
  AND state IN ('active', 'idle in transaction');

比如用shell脚本每100ms执行一次这个查询，把结果存入文件，最后用awk或Python分析锁等待时长分布、锁定行的频次、各锁类型的平均耗时。

2. 日志分析 #

修改postgresql.conf开启锁等待日志：

log_lock_waits = on
log_statement = 'all'
log_min_duration_statement = 0

重启数据库后，所有锁等待事件和执行的查询都会被记录到日志里。之后直接用grep、awk这类工具提取锁等待时长、对应查询、锁定对象，就能统计1000次请求的锁指标。

二、第三方工具 #

1. pgBadger #

这是专门的PostgreSQL日志分析工具，能自动解析锁等待日志，生成可视化的HTML报告，里面包含锁等待时长分布、高频锁等待查询、涉及的表/行等信息。只要开启了上面的日志参数，把日志文件喂给pgBadger就能得到现成的统计报告。

2. pgHero #

它能实时监控锁状态，也能生成历史统计报告，涵盖锁等待次数、平均等待时长、热点锁对象等数据。UI界面直观，适合快速查看1000次请求后的锁开销情况。

三、自定义测试脚本 #

如果需要精准控制1000次并行请求的逻辑，用Python的psycopg2或Go的pgx写测试脚本就行：
脚本逻辑是启动多线程/协程同时执行带锁的查询，记录每个请求的开始/结束时间、锁定行数，最后汇总统计锁等待时长、总开销等。
Python示例片段：

import psycopg2
import threading
import time

results = []

def run_query():
    conn = psycopg2.connect("dbname=test user=postgres")
    cur = conn.cursor()
    start_time = time.time()
    try:
        cur.execute("select * from mytable where age > 12 and age < 45 for update")
        rows = cur.fetchall()
        duration = time.time() - start_time
        results.append({"duration": duration, "locked_rows": len(rows), "success": True})
    except Exception as e:
        duration = time.time() - start_time
        results.append({"duration": duration, "error": str(e), "success": False})
    finally:
        cur.close()
        conn.close()

# 启动1000次请求
threads = []
for _ in range(1000):
    t = threading.Thread(target=run_query)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

# 统计输出
valid_results = [r for r in results if r["success"]]
avg_wait = sum(r["duration"] for r in valid_results) / len(valid_results)
max_wait = max(r["duration"] for r in valid_results)
total_locked = sum(r["locked_rows"] for r in valid_results)

print(f"平均查询/锁等待时长: {avg_wait:.4f}秒")
print(f"最长等待时长: {max_wait:.4f}秒")
print(f"累计锁定行数: {total_locked}")

内容的提问来源于stack exchange，提问作者mleko