pgsql binlog監聽功能點解析

2023-11-02 06:00:27

引言

監聽mysql binlog 大家都知道canal，但是如果是pglog呢，先百度

也就這個靠點譜，文章沒有我想要的demo

去官網看看debezium.io/

其中這個網址給了demo 但是不能直接用於生產

功能點

首先 pglog binlog監聽需要滿足哪些功能點

機器宕機，能支援斷點續接

進行磁碟持久化

如果監聽的表長時間沒有資料變動，delay_size 會變大

//設定心跳時間，就算沒有資料 也會保持心跳
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);
    }
}

心跳這個是當時上生產的時候，突然發現沒有資料變更的時候，有報警，說delay了。。。這頓害怕

大概意思

資料庫中有許多更新正在被跟蹤，但只有極少數更新與聯結器正在為其捕獲更改的表和模式相關。這種情況可以通過週期性的心跳事件輕鬆解決。設定heartbeat.interval.ms聯結器設定屬性。

由於WAL由所有資料庫共用，因此使用的WAL數量趨於增長，直到Debezium為其捕獲更改的資料庫觸發事件。為了克服這一點，有必要:使用heartbeat.interval.ms聯結器設定屬性啟用週期性心跳記錄生成。定期從Debezium正在捕捉變化的資料庫中發出更改事件。

其中

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

這是因為如果沒有資料來的話，會是ts_ms 開頭的，代表，沒有新資料

每次binlog傳的size 太多，導致伺服器處理不過來

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(",")));

希望磁碟持久化offset，保持資料的正確性

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

小工具

查詢資料庫 offset推遲多少

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;

這個工具在生產已經實踐穩定，可以直接使用，有問題可以評論

程式碼 github.com/a25017012/y…

以上就是pgsql binlog監聽功能點解析的詳細內容，更多關於pgsql binlog監聽的資料請關注it145.com其它相關文章！