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  # 启用自动槽管理