pgsql binlog监听功能点解析

//设置心跳时间，就算没有数据 也会保持心跳
props.setProperty("heartbeat.interval.ms", "20000");

for (ChangeEvent<String, String> r : records) {
    try {
        if (log.isDebugEnabled()) {
            log.debug("{}\n{}", r.key(), r.value());
        }
        if (r.value() != null && r.value().startsWith("{"ts_ms")) {
            continue;
        }
        xxx 具体数据处理
    } catch (Exception e) {
        log.error("PGLog-binlog param:[{}]", r, e);
    }
}

if (r.value() != null && r.value().startsWith("{"ts_ms")) {
    continue;
}

props.setProperty("max.BATch.size", "200");

props.setProperty("table.include.list", schs.stream().map(BinloGConfig::getSch).map(a -> tables.stream().map(b -> a + "." + b).map(String::valueOf).collect(Collectors.joining(","))).map(String::valueOf).collect(Collectors.joining(",")));

props.setProperty("snapshot.mode", "never");

select pg_replication_slots.*, 
pg_current_wal_lsn(), 
pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_insert_lsn(), 
restart_lsn)) as delay_size 
from pg_replication_slots;