PostgreSQL主从复制的监控与故障切换指南

-- 查看所有从库的连接和复制进度
select 
    pid,
    usename,
    application_name,
    client_addr,
    state,
    sync_state,
    sent_lsn,
    write_lsn,
    flush_lsn,
    replay_lsn,
    pg_wal_lsn_diff(sent_lsn, replay_lsn) as replay_lag_bytes
from pg_stat_replication;

-- 判断是否为从库
select pg_is_in_recovery(); -- true 表示是从库

-- 获取最后接收到的 wal 位置
select pg_last_wal_receive_lsn();

-- 获取最后重放的 wal 位置
select pg_last_wal_replay_lsn();

-- 计算时间延迟（需主库支持 track_commit_timestamp）
select 
    extract(epoch from (now() - pg_last_xact_replay_timestamp())) as replay_lag_seconds;

<dependency>
    <groupid>org.postgresql</groupid>
    <artifactid>postgresql</artifactid>
    <version>42.7.3</version>
</dependency>

public class replicationstatus {
    private string host;
    private boolean isstandby;
    private long replaylagbytes;
    private double replaylagseconds;
    private boolean isconnected;
    private string errormessage;

    // getters and setters
}

import java.sql.*;
import java.time.duration;
import java.time.instant;

public class pgreplicationmonitor {

    public static replicationstatus checkreplication(string jdbcurl, string username, string password) {
        replicationstatus status = new replicationstatus();
        status.sethost(jdbcurl);
        status.setconnected(false);

        try (connection conn = drivermanager.getconnection(jdbcurl, username, password)) {
            status.setconnected(true);

            // 检查是否为从库
            try (preparedstatement ps = conn.preparestatement("select pg_is_in_recovery()")) {
                resultset rs = ps.executequery();
                if (rs.next()) {
                    status.setisstandby(rs.getboolean(1));
                }
            }

            if (status.isisstandby()) {
                // 从库：获取延迟
                try (preparedstatement ps = conn.preparestatement(
                        "select " +
                        "pg_last_wal_receive_lsn(), " +
                        "pg_last_wal_replay_lsn(), " +
                        "extract(epoch from (now() - pg_last_xact_replay_timestamp()))")) {
                    resultset rs = ps.executequery();
                    if (rs.next()) {
                        string receivelsn = rs.getstring(1);
                        string replaylsn = rs.getstring(2);
                        double lagseconds = rs.getdouble(3);

                        // 计算字节延迟（需转换 lsn）
                        long bytelag = calculatelsndiff(receivelsn, replaylsn);
                        status.setreplaylagbytes(bytelag);
                        status.setreplaylagseconds(lagseconds);
                    }
                }
            } else {
                // 主库：可选，检查从库连接数等
                // 此处略
            }

        } catch (sqlexception e) {
            status.seterrormessage(e.getmessage());
        }

        return status;
    }

    // 简化版 lsn 差值计算（实际应解析 lsn 格式）
    private static long calculatelsndiff(string lsn1, string lsn2) {
        if (lsn1 == null || lsn2 == null) return 0;
        // 实际项目中建议使用 postgresql 的 pg_wal_lsn_diff 函数在 sql 中计算
        // 此处仅为示意
        return math.abs(lsn1.hashcode() - lsn2.hashcode());
    }
}

import java.util.concurrent.executors;
import java.util.concurrent.scheduledexecutorservice;
import java.util.concurrent.timeunit;

public class replicationwatcher {
    private static final string primary_url = "jdbc:postgresql://primary-db:5432/mydb";
    private static final string standby_url = "jdbc:postgresql://standby-db:5432/mydb";
    private static final string username = "repuser";
    private static final string password = "secret";

    public static void main(string[] args) {
        scheduledexecutorservice scheduler = executors.newscheduledthreadpool(2);

        scheduler.scheduleatfixedrate(() -> {
            replicationstatus standby = pgreplicationmonitor.checkreplication(standby_url, username, password);
            if (!standby.isconnected()) {
                alert("standby db connection failed: " + standby.geterrormessage());
            } else if (standby.getreplaylagseconds() > 30) {
                alert("high replication lag: " + standby.getreplaylagseconds() + " seconds");
            }
        }, 0, 10, timeunit.seconds); // 每10秒检查一次
    }

    private static void alert(string message) {
        system.err.println("[alert] " + instant.now() + ": " + message);
        // 可集成邮件、钉钉、企业微信等通知
    }
}

scope: mycluster
namespace: /service/
name: pg-node1

restapi:
  listen: 0.0.0.0:8008
  connect_address: 192.168.1.10:8008

etcd:
  hosts: ["etcd1:2379", "etcd2:2379", "etcd3:2379"]

bootstrap:
  dcs:
    ttl: 30
    loop_wait: 10
    retry_timeout: 10
    maximum_lag_on_failover: 1048576  # 1mb
    postgresql:
      use_pg_rewind: true
      parameters:
        wal_level: replica
        hot_standby: on
        max_wal_senders: 10
        wal_keep_segments: 8

postgresql:
  listen: 0.0.0.0:5432
  connect_address: 192.168.1.10:5432
  data_dir: /var/lib/postgresql/14/main
  bin_dir: /usr/lib/postgresql/14/bin
  authentication:
    replication:
      username: replicator
      password: rep-pass
    superuser:
      username: postgres
      password: admin-pass

// hikaricp 配置示例
hikariconfig config = new hikariconfig();
config.setjdbcurl("jdbc:postgresql://new-primary:5432/mydb");
config.setusername("appuser");
config.setpassword("pass");
config.setconnectiontimeout(3000);
config.setidletimeout(60000);
config.setmaxlifetime(1800000);
config.setmaximumpoolsize(20);

// 关键：启用自动重连
config.adddatasourceproperty("rewritebatchedinserts", "true");
config.adddatasourceproperty("tcpkeepalive", "true");

hikaridatasource ds = new hikaridatasource(config);

@refreshscope
@restcontroller
public class databasecontroller {

    @value("${db.primary.host}")
    private string primaryhost;

    @getmapping("/db/host")
    public string getdbhost() {
        return primaryhost; // 由 consul 动态注入
    }
}

public class masterdetector {
    private volatile string currentmaster = "primary-db";

    public void startdetection() {
        executors.newsinglethreadscheduledexecutor().schedulewithfixeddelay(() -> {
            try {
                // 尝试连接候选主库列表
                for (string candidate : arrays.aslist("node1", "node2", "node3")) {
                    if (ismaster(candidate)) {
                        currentmaster = candidate;
                        break;
                    }
                }
            } catch (exception e) {
                // log error
            }
        }, 0, 5, timeunit.seconds);
    }

    private boolean ismaster(string host) {
        try (connection conn = drivermanager.getconnection(
                "jdbc:postgresql://" + host + ":5432/mydb", "user", "pass")) {
            try (statement stmt = conn.createstatement();
                 resultset rs = stmt.executequery("select pg_is_in_recovery()")) {
                return rs.next() && !rs.getboolean(1); // 不在恢复模式即为主库
            }
        } catch (sqlexception e) {
            return false;
        }
    }

    public string getcurrentmaster() {
        return currentmaster;
    }
}

-- 创建物理复制槽
select pg_create_physical_replication_slot('standby1_slot');

-- 查看槽状态
select * from pg_replication_slots;

postgresql:
  parameters:
    max_replication_slots: 5
  use_slots: true  # 启用自动槽管理