如何用脚本批量终止PostgreSQL查询？

wen 实用脚本 2026-07-21 1

本文目录导读：

方法一：使用pg_cancel_backend()（单个中断）
方法二：批量终止脚本（Shell）
方法三：Python脚本（更灵活）
方法四：使用pg_terminate_backend()（强制终止）
方法五：根据条件过滤的Shell脚本
方法六：使用pgCancel（libpq提供的函数）
使用建议

使用pg_cancel_backend()（单个中断）

-- 生成终止所有活动的查询的SQL语句
SELECT pg_cancel_backend(pid) 
FROM pg_stat_activity 
WHERE state = 'active' 
  AND pid <> pg_backend_pid();

批量终止脚本（Shell）

方式1：使用psql命令行

#!/bin/bash
# 终止所有非当前的查询
psql -h localhost -U postgres -d your_database -c "
SELECT pg_terminate_backend(pid) 
FROM pg_stat_activity 
WHERE state = 'active' 
  AND pid <> pg_backend_pid()
  AND pid <> pg_backend_pid();
"

方式2：更详细的终止脚本

#!/bin/bash
# 批量终止PostgreSQL查询脚本
DB_NAME="your_database"  # 数据库名称，可选
EXCLUDE_USER="postgres"  # 排除的用户
EXCLUDE_APP="psql"       # 排除的应用
# 生成终止命令
SQL="SELECT pg_terminate_backend(pid) 
FROM pg_stat_activity 
WHERE state = 'active' 
  AND pid <> pg_backend_pid()"
# 如果指定了数据库
if [ -n "$DB_NAME" ]; then
    SQL="$SQL AND datname = '$DB_NAME'"
fi
# 排除特定用户
if [ -n "$EXCLUDE_USER" ]; then
    SQL="$SQL AND usename != '$EXCLUDE_USER'"
fi
# 执行终止
psql -h localhost -U postgres -c "$SQL"

Python脚本（更灵活）

#!/usr/bin/env python3
"""
批量终止PostgreSQL查询脚本
支持多种过滤条件
"""
import psycopg2
from psycopg2 import sql
import sys
def terminate_queries(host='localhost', port=5432, 
                     database=None, user='postgres',
                     password=None, exclude_pids=None):
    """
    批量终止PostgreSQL查询
    Args:
        host: 数据库主机
        port: 端口
        database: 数据库名，None表示所有数据库
        user: 用户名
        password: 密码
        exclude_pids: 要排除的PID列表
    """
    # 连接数据库
    conn = psycopg2.connect(
        host=host,
        port=port,
        database='postgres',
        user=user,
        password=password
    )
    try:
        with conn.cursor() as cur:
            # 查询要终止的进程
            query = """
                SELECT pid, usename, application_name, 
                       state, query, query_start
                FROM pg_stat_activity 
                WHERE state = 'active'
                  AND pid <> pg_backend_pid()
            """
            if database:
                query += f" AND datname = '{database}'"
            if exclude_pids:
                pids_str = ','.join(map(str, exclude_pids))
                query += f" AND pid NOT IN ({pids_str})"
            cur.execute(query)
            processes = cur.fetchall()
            if not processes:
                print("没有找到活跃的查询")
                return
            print(f"找到 {len(processes)} 个活跃查询：")
            for pid, username, app, state, query_text, start_time in processes:
                print(f"  PID: {pid}, 用户: {username}, 应用: {app}, "
                      f"开始时间: {start_time}")
            # 确认终止
            response = input("\n是否终止以上所有查询？(y/N): ")
            if response.lower() != 'y':
                print("操作取消")
                return
            # 执行终止
            terminated = 0
            for pid, _, _, _, _, _ in processes:
                try:
                    cur.execute("SELECT pg_terminate_backend(%s)", (pid,))
                    terminated += 1
                    print(f"已终止 PID: {pid}")
                except Exception as e:
                    print(f"终止 PID {pid} 失败: {e}")
            print(f"\n成功终止 {terminated} 个查询")
    finally:
        conn.close()
# 使用示例
if __name__ == "__main__":
    # 终止特定数据库的所有查询，排除当前连接
    terminate_queries(
        host='localhost',
        database='mydb',
        user='postgres',
        exclude_pids=[12345, 67890]  # 排除特定PID
    )

使用pg_terminate_backend()（强制终止）

-- 强制终止所有长时间运行的查询（超过5分钟）
SELECT pg_terminate_backend(pid)
FROM pg_stat_activity
WHERE state = 'active'
  AND pid <> pg_backend_pid()
  AND query_start < now() - interval '5 minutes';

根据条件过滤的Shell脚本

#!/bin/bash
# 根据各种条件终止查询
DB_HOST="localhost"
DB_USER="postgres"
# 终止特定数据库的查询
terminate_by_database() {
    local db=$1
    psql -h $DB_HOST -U $DB_USER -d postgres -c "
    SELECT pg_terminate_backend(pid) 
    FROM pg_stat_activity 
    WHERE datname = '$db' 
      AND pid <> pg_backend_pid();
    "
}
# 终止特定用户的查询
terminate_by_user() {
    local user=$1
    psql -h $DB_HOST -U $DB_USER -d postgres -c "
    SELECT pg_terminate_backend(pid) 
    FROM pg_stat_activity 
    WHERE usename = '$user' 
      AND pid <> pg_backend_pid();
    "
}
# 终止空闲的事务
terminate_idle_transactions() {
    psql -h $DB_HOST -U $DB_USER -d postgres -c "
    SELECT pg_terminate_backend(pid) 
    FROM pg_stat_activity 
    WHERE state = 'idle in transaction' 
      AND age(now(), state_change) > interval '1 hour';
    "
}
# 使用示例
# terminate_by_database "mydb"
# terminate_by_user "appuser"
# terminate_idle_transactions

使用pgCancel（libpq提供的函数）

#!/bin/bash
# 使用pgCancel批量取消查询
# 首先获取要取消的进程列表
psql -U postgres -At -c "
SELECT pid FROM pg_stat_activity 
WHERE state = 'active' 
  AND pid <> pg_backend_pid();
" | while read pid; do
    echo "Canceling query on PID: $pid"
    psql -U postgres -c "SELECT pg_cancel_backend($pid);"
done

使用建议

优先使用pg_cancel_backend()：这会让查询正常终止，类似取消操作
谨慎使用pg_terminate_backend()：这会强制终止进程，可能导事务回滚
添加必要的过滤条件：
- AND pid <> pg_backend_pid()：排除当前连接
- AND datname = 'dbname'：限制特定数据库
- AND state = 'active'：只处理活跃查询
安全建议：
- 先在测试环境验证
- 提供确认机制（如上面的Python脚本）
- 记录操作日志
- 排除重要系统进程

选择哪种方法取决于你的具体需求和安全要求,对于生产环境，建议使用带确认机制的脚本，或者先使用 pg_cancel_backend() 再使用 pg_terminate_backend()。