tunnel同步PG數(shù)據(jù)到kafka

來自哈羅單車開源的組件。支持同步PG數(shù)據(jù)到kafka或者ES。

目前創(chuàng)新互聯(lián)已為成百上千家的企業(yè)提供了網(wǎng)站建設(shè)、域名、虛擬主機、網(wǎng)站托管、服務(wù)器托管、企業(yè)網(wǎng)站設(shè)計、梅州網(wǎng)站維護等服務(wù)，公司將堅持客戶導(dǎo)向、應(yīng)用為本的策略，正道將秉承"和諧、參與、激情"的文化，與客戶和合作伙伴齊心協(xié)力一起成長，共同發(fā)展。

?

https://github.com/hellobike/tunnel

?

tunnel整體的部署比較簡單的

需要事先部署好zk和kafka（我下面演示的是單節(jié)點的zk和kafka）

?

節(jié)點部署關(guān)系：

192.168.2.4?? 部署zk、kafka、pg10運行在1921端口

192.168.2.189 部署tunnel

?

?

確保已開啟PG的邏輯復(fù)制

wal_level = 'logical';

max_replication_slots = 20

注意這個設(shè)置要重啟PG進程的

?

然后，創(chuàng)建測試庫表和同步用的賬號

CREATE DATABASE test_database;

\c test_database

create table test_1 (id int primary key , name char(40));

create table test_2 (id int primary key , name char(40));

?

CREATE ROLE test_rep LOGIN ENCRYPTED PASSWORD 'xxxx' REPLICATION;

GRANT CONNECT ON DATABASE test_database to test_rep;

?

vim pg_hba.conf增加2行配置：

host??? all????? ???? ?????? ?test_rep??????? 192.168.2.0/24??????? ?md5

host??? replication???? test_rep??????? 192.168.2.0/24???????? md5

然后 reload 下PG

?

到192.168.2.189機器上去編譯tunnel：

注意： tunnel的啟動需要事先安裝好oracle jdk 1.8

?

git clone https://github.com/hellobike/tunnel

cd tunnel

mvn clean package -Dmaven.test.skip=true

cd target

unzip AppTunnelService.zip

cd AppTunnelService

?

vim conf/test.yml內(nèi)容如下：

tunnel_subscribe_config:

? pg_dump_path: '/usr/local/pgsql-10.10/bin/pg_dump'

? subscribes:

? - slotName: slot_for_test

??? pgConnConf:

????? host: 192.168.2.4

????? port: 1921

????? database: test_database

????? user: test_rep

????? password: xxxx

??? rules:

??? - {table: test_1, pks: ['id'], topic: test_1_logs}

??? - {table: test_2, pks: ['id'], topic: test_2_logs}

??? kafkaConf:

????? addrs:

????? - 192.168.2.4:9092

tunnel_zookeeper_address:192.168.2.4:2181

?

前臺啟動：

java -server -classpath conf/*:lib/* com.hellobike.base.tunnel.TunnelLauncher -u false -c cfg.properties -p 7788? ? ?#暴露prometheus metric在7788端口(配置監(jiān)控不是這里的重點，也很簡單，暫時先跳過)

然后，我們再在PG10上面的test_database的2張表隨便造些數(shù)據(jù)，然后可以看到kafka里面已經(jīng)有數(shù)據(jù)了（下圖是通過kafkamanager和 kafka-eagle的結(jié)果）。

格式化下，數(shù)據(jù)就是這樣的：

UPDATE的記錄的樣子：

{

???????? "dataList": [{

?????????????????? "dataType": "integer",

?????????????????? "name": "id",

?????????????????? "value": "1111"

???????? }, {

?????????????????? "dataType": "character",

?????????????????? "name": "name",

?????????????????? "value": "大狗蛋 "

???????? }],

???????? "eventType": "UPDATE",

???????? "lsn": 10503246616,

???????? "schema": "public",

???????? "table": "test_1"

}

?

DELETE的記錄的樣子：

{

???????? "dataList": [{

?????????????????? "dataType": "integer",

?????????????????? "name": "id",

?????????????????? "value": "3"

???????? }],

???????? "eventType": "DELETE",

???????? "lsn": 10503247064,

???????? "schema": "public",

???????? "table": "test_1"

}