一 安装

BottledWater-PG的安装前文已经表述，本文不赘述直接进入集成应用阶段。

二 启动KafKa

#启动zookeeper
[root@bogon kafka_2.11-0.10.2.0]# bin/zookeeper-server-start.sh config/zookeeper.properties
#启动kafka服务端
[root@bogon kafka_2.11-0.10.2.0]# bin/kafka-server-start.sh config/server.properties

三 配置PostgreSQL

3.1 配置读取权限

Bottled Water会连接到postgresql获取相关数据，连接的账户需要有replication权限，pg中数据库的变化存储在WAL中，至少需要replication权限才能读取WAL。
编辑$PGDATA目录中postgresql.conf和pg_hba.conf文件。

vi $PGDATA/postgresql.conf
#编辑内容如下：
listen_addresses = '*'
port = 5432 
wal_level = logical         
max_wal_senders = 8
wal_keep_segments = 4
max_replication_slots = 4

vi $PGDATA/pg_hba.conf
#编辑内容如下：

# IPv4 local connections:
host    all             all             0.0.0.0/0               md5
# replication privilege.
local   replication     freerep                                trust
host    replication     freerep        127.0.0.1/32            trust
host    replication     freerep        ::1/128                 trust

编辑完保存，重启数据库服务：

pg_ctl restart
psql
postgres=# CREATE ROLE freerep  WITH REPLICATION PASSWORD 'password' LOGIN;
CREATE ROLE

配置完毕！

3.2 Bottled Water使用演示

3.2.1 创建测试库表

创建一个测试数据库，建立测试表

postgres=# create database mcsas;
postgres=# \c mcsas;
mcsas=# create extension bottledwater;
mcsas=# create extension postgis;
#赋予public下的表给freerep角色，要创建如下语句，否则建立的表freerep没有读取权限
mcsas=# alter default privileges in schema public grant all on tables to freerep;
mcsas=# create table gps(gid serial primary key,name text,geom text);
mcsas=# create index gps_geom_idx on gps using gist(ST_GeomFromText(geom,4326));

在另一个终端启动bottledwater可执行程序:

source /home/postgres/.bashrc
cd /opt/bottledwater-pg-master/kafka
[root@localhost kafka]# ./bottledwater -d postgres://freerep:password@127.0.0.1/mcsas -b 192.168.43.27:9092 -f json

启动结果如下：

[root@bogon kafka]# ./bottledwater -d postgres://freerep:password@127.0.0.1/mcsas -b 192.168.43.27:9092 -f json
[INFO] Writing messages to Kafka in JSON format
[INFO] Created replication slot "bottledwater", capturing consistent snapshot "0000DA72-1".
INFO:  bottledwater_export: Table public.spatial_ref_sys is keyed by index spatial_ref_sys_pkey
INFO:  bottledwater_export: Table public.mark is keyed by index mark_pkey
[INFO] Registering metadata for table spatial_ref_sys (relid 24263)
[INFO] Opening Kafka topic "spatial_ref_sys" for table "spatial_ref_sys"
[INFO] Storing key schema for table 24263
[INFO] Storing row schema for table 24263
[INFO] Snapshot complete, streaming changes from 0/AB016F30.

代表启动成功了。

3.2.2 监听数据改变消息

插入数据

mcsas=# insert into gps(name,geom) values ('china','Point(118 32)');
INSERT 0 1
mcsas=# insert into gps(name,geom) values ('england','Point(118 12)');
INSERT 0 1

启动监听topic

bin/kafka-console-consumer.sh --bootstrap-server 192.168.43.27:9092 --topic gps --from-beginning
{"gid": {"int": 1}, "name": {"string": "china"}, "geom": {"string": "Point(118 32)"}}
{"gid": {"int": 2}, "name": {"string": "england"}, "geom": {"string": "Point(118 12)"}}

每当插入或者更新，收听的消息会源源不断的输出出来，这样，pg与kafka集成就完毕了。