实时数据同步利器debezium教程

1 debezium概述

Debezium是一个开源项目，为捕获数据更改(change data capture,CDC)提供了一个低延迟的流式处理平台。你可以安装并且配置Debezium去监控你的数据库，然后你的应用就可以消费对数据库的每一个行级别(row-level)的更改。

只有已提交的更改才是可见的，所以你的应用不用担心事务(transaction)或者更改被回滚(roll back)。Debezium为所有的数据库更改事件提供了一个统一的模型，所以你的应用不用担心每一种数据库管理系统的错综复杂性。另外，由于Debezium用持久化的、有副本备份的日志来记录数据库数据变化的历史，因此，你的应用可以随时停止再重启，而不会错过它停止运行时发生的事件，保证了所有的事件都能被正确地、完全地处理掉。

监控数据库，并且在数据变动的时候获得通知一直是很复杂的事情。关系型数据库的触发器可以做到，但是只对特定的数据库有效，而且通常只能更新数据库内的状态(无法和外部的进程通信)。一些数据库提供了监控数据变动的API或者框架，但是没有一个标准，每种数据库的实现方式都是不同的，并且需要大量特定的知识和理解特定的代码才能运用。确保以相同的顺序查看和处理所有更改，同时最小化影响数据库仍然非常具有挑战性。

Debezium提供了模块为你做这些复杂的工作。一些模块是通用的，并且能够适用多种数据库管理系统，但在功能和性能方面仍有一些限制。另一些模块是为特定的数据库管理系统定制的，所以他们通常可以更多地利用数据库系统本身的特性来提供更多功能。

2 debezium常见应用场景

Debezium有很多非常有价值的使用场景，我们在这儿仅仅列出几个更常见的使用场景。

一、缓存失效(Cache invalidation)
在缓存中缓存的条目(entry)在源头被更改或者被删除的时候立即让缓存中的条目失效。如果缓存在一个独立的进程中运行(例如Redis，Memcache，Infinispan或者其他的)，那么简单的缓存失效逻辑可以放在独立的进程或服务中，从而简化主应用的逻辑。在一些场景中，缓存失效逻辑可以更复杂一点，让它利用更改事件中的更新数据去更新缓存中受影响的条目。

二、简化单体应用(Simplifying monolithic applications)
许多应用更新数据库，然后在数据库中的更改被提交后，做一些额外的工作：更新搜索索引，更新缓存，发送通知，运行业务逻辑，等等。这种情况通常称为双写(dual-writes)，因为应用没有在一个事务内写多个系统。这样不仅应用逻辑复杂难以维护，而且双写容易丢失数据或者在一些系统更新成功而另一些系统没有更新成功的时候造成不同系统之间的状态不一致。使用捕获更改数据技术(change data capture,CDC)，在源数据库的数据更改提交后，这些额外的工作可以被放在独立的线程或者进程(服务)中完成。这种实现方式的容错性更好，不会丢失事件，容易扩展，并且更容易支持升级。

三、共享数据库(Sharing databases)
当多个应用共用同一个数据库的时候，一个应用提交的更改通常要被另一个应用感知到。一种实现方式是使用消息总线，尽管非事务性(non-transactional)的消息总线总会受上面提到的双写(dual-writes)影响。但是，另一种实现方式，即Debezium，变得很直接：每个应用可以直接监控数据库的更改，并且响应更改。

四、数据集成(Data integration)
数据通常被存储在多个地方，尤其是当数据被用于不同的目的的时候，会有不同的形式。保持多系统的同步是很有挑战性的，但是可以通过使用Debezium加上简单的事件处理逻辑来实现简单的ETL类型的解决方案。

五、命令查询职责分离(CQRS)
在命令查询职责分离(Command Query Responsibility Separation,CQRS)架构模式中，更新数据使用了一种数据模型，读数据使用了一种或者多种数据模型。由于数据更改被记录在更新侧(update-side)，这些更改将被处理以更新各种读展示。所以CQRS应用通常更复杂，尤其是他们需要保证可靠性和全序(totally-ordered)处理。Debezium和CDC可以使这种方式更可行：写操作被正常记录，但是Debezium捕获数据更改，并且持久化到全序流里，然后供那些需要异步更新只读视图的服务消费。写侧(write-side)表可以表示面向领域的实体(domain-oriented entities)，或者当CQRS和Event Sourcing结合的时候，写侧表仅仅用做追加操作命令事件的日志。

3 debezium系统架构

Debezium是一个捕获数据更改(CDC)平台，并且利用Kafka和Kafka Connect实现了自己的持久性、可靠性和容错性。

每一个部署在Kafka Connect分布式的、可扩展的、容错性的服务中的connector监控一个上游数据库服务器，捕获所有的数据库更改，然后记录到一个或者多个Kafka topic(通常一个数据库表对应一个kafka topic)。

Kafka确保所有这些数据更改事件都能够多副本并且总体上有序(Kafka只能保证一个topic的单个分区内有序)，这样，更多的客户端可以独立消费同样的数据更改事件而对上游数据库系统造成的影响降到很小(如果N个应用都直接去监控数据库更改，对数据库的压力为N，而用debezium汇报数据库更改事件到kafka，所有的应用都去消费kafka中的消息，可以把对数据库的压力降到1)。

另外，客户端可以随时停止消费，然后重启，从上次停止消费的地方接着消费。每个客户端可以自行决定他们是否需要exactly-once或者at-least-once消息交付语义保证，并且所有的数据库或者表的更改事件是按照上游数据库发生的顺序被交付的。

对于不需要或者不想要这种容错级别、性能、可扩展性、可靠性的应用，他们可以使用内嵌的Debezium connector引擎来直接在应用内部运行connector。这种应用仍需要消费数据库更改事件，但更希望connector直接传递给它，而不是持久化到Kafka里。

部署架构

最常见的是，Debezium是通过Apache Kafka连接部署的。Kafka Connect是一个用于实现和操作的框架和运行时

源连接器，如Debezium，它将数据摄取到Kafka
接收连接器，它将数据从Kafka主题写入到其他系统。

下图显示了一个基于Debezium的CDC管道的架构:

Debezium Server
Debezium 的另一种部署方法是使用Debezium 服务器。Debezium Server是一个可配置、开箱即用的应用程序。通过Debezium server，可以将源数据库流式变化事件同步到各种不同的消息基础设施。

作为嵌入式引擎
使用Debezium连接器的另一种方法是嵌入式引擎。在这种情况下，Debezium不会通过Kafka Connect运行，而是作为一个嵌入到定制Java应用程序中的库运行。这对于在应用程序内部使用更改事件非常有用，而不需要部署完整的Kafka和Kafka连接集群，或者将更改流到其他消息传递代理(如Amazon Kinesis)。

4 debezium安装

kafka connect模式

Kafka Connect当前支持两种模式，standalone和distributed两种模式。

standalone主要用于入门测试，所以我们来实现distributed模式。

Distributed，分布式模式可以在处理工作中自动平衡，允许动态扩展或缩减，并在活动任务以及配置和偏移量提交数据中提供容错能力。和standalone模式非常类似，最大区别在于启动的类和配置参数，参数决定了Kafka Connect流程如果存储偏移量，如何分配工作，在分布式模式下，Kafka Connect将偏移量，配置和任务状态存储在topic中。建议手动创建topic指定分区数，也可以通过配置文件参数自动创建topic。

以Debezium支持MySQL为例（分布式模式）。

先决条件：Kafka集群已安装（包括ZooKeeper）、MySQL已安装并启动实例。

1）下载MySQL connector插件：https://debezium.io/documentation/reference/1.6/install.html

2）在kafka安装目录下创建connect插件目录并配置生效

创建目录connect。将插件解压缩到该目录下。
修改/config/connect-distributed.properties, 增减一行：

 plugin.path=/opt/kafka/connect

默认情况下，kafka connect会加载在该目录下的所有插件。

启动connect服务。（该服务可以独立启动，但由于分布式模式下状态信息都保存在kafka中，所以kafka server服务可以先启动）

/opt/kafka/bin/connect-distributed.sh  -daemon /opt/kafka/config/connect-distributed.properties

查看8083端口是否监听：

netstat -nltp | grep 8083

正常启动情况下，可以看到：

tcp6       0      0 :::8083                 :::*                    LISTEN      6744/java

利用Postman可以查看到一下结果：

5 debezium mysql插件使用

下面我们构造一个实际的使用实例来说明debezium mysql的使用方法。

1）安装mysql，本次用mariadb-10.5.8替代。关于Mariadb的安装方法参考：https://blog.csdn.net/sinat_39809957/article/details/118764511?spm=1001.2014.3001.5501

2）创建一个数据库：
连接mariadb：./mariadb -S ../mysql.sock -uzst -p 创建数据库：create database test；`

3). 创建表：

 CREATE TABLE IF NOT EXISTS `userinfo`(`userid` int(11) not null,`username` varchar(100) not null,`age` int(11) not null,`num` int(11) not null,PRIMARY KEY ( `userid` )
)ENGINE=InnoDB DEFAULT CHARSET=utf8;

4). 打开数据库的binlog

首先检查binlog日志选项是否已打开

mysql> SELECT variable_value as "BINARY LOGGING STATUS (log-bin) ::"
FROM information_schema.global_variables WHERE variable_name='log_bin';
或者show variables link 'log_bin'

如果显示为“off”，则需要修改my.cnf文件，在[mysqld]段增加配置项如下：

server-id         = 223344
log_bin           = mysql-bin
binlog_format     = ROW
binlog_row_image  = FULL
expire_logs_days  = 10

再次查看日志选项是否已打开：

mysql> SELECT variable_value as "BINARY LOGGING STATUS (log-bin) ::"
FROM information_schema.global_variables WHERE variable_name='log_bin';

binlog选项打开后，执行：show binlog events.，可以看到binlog列表。

5）启动debezium mysql connector插件：
配置如下

{"name": "debezium-mysql-connector","config": {"connector.class": "io.debezium.connector.mysql.MySqlConnector","database.hostname": "168.61.38.11","database.port": "3306","database.user": "zst","database.password": "123456","database.server.id": "223344","database.server.name": "conndbtest","database.include.list": "test","database.history.kafka.bootstrap.servers": "168.61.38.11:9092,168.61.38.12:9092,168.61.38.13:9092,168.61.38.15:9092,168.61.38.16:9092","database.history.kafka.topic": "debe-mysql","include.schema.changes": "true"}
}

需要注意的是，server.id可以通过 select @@server_id获取到，该id在my.cnf中配置。server.name则是在此处随便定义一个。通过Postman把该配置打进kafka中，完成启动。

6）查看kafka，debe-mysql已创建，与此同时，debezium connector会默认创建3个topic为
connect-configs，connect-offsets，connect-status分别存放配置、偏移、状态信息。
执行脚本，查看dbtest-history是否已创建。

./kafka-topics.sh --zookeeper 168.61.38.11:2181 168.61.38.12:2181 168.61.38.13:2181 168.61.38.15:2181 168.61.38.16:2181 --list

debezium默认情况下，会将同一表上的insert、update、delete等操作日志事件写入到一个kafka topic中，其创建的topic名字规则是：
serverName.databaseName.tableName

7）在数据库test中，对userinfo表做一些DDL操作。
如修改userinfo表的一列：

alter table userinfo modify username char(50);

先执行：

./kafka-console-consumer.sh --bootstrap-server 168.61.38.11:9092,168.61.38.12:9092,168.61.38.13:9092,168.61.38.15:9092,168.61.38.16:9092 --topic debe-mysql

然后修改表，可以看到如下结果

{"source" : {"server" : "conndbtest"},"position" : {"transaction_id" : null,"ts_sec" : 1637744120,"file" : "mysql-bin.000003","pos" : 47012,"server_id" : 223344},"databaseName" : "","ddl" : "# Dum","tableChanges" : [ ]
}
{"source" : {"server" : "conndbtest"},"position" : {"transaction_id" : null,"ts_sec" : 1637744120,"file" : "mysql-bin.000003","pos" : 47124,"server_id" : 223344},"databaseName" : "test","ddl" : "alter table userinfo modify username char(50)","tableChanges" : [ {"type" : "ALTER","id" : "\"test\".\"userinfo\"","table" : {"defaultCharsetName" : "latin1","primaryKeyColumnNames" : [ "userid" ],"columns" : [ {"name" : "userid","jdbcType" : 4,"typeName" : "INT","typeExpression" : "INT","charsetName" : null,"length" : 11,"position" : 1,"optional" : false,"autoIncremented" : false,"generated" : false}, {"name" : "username","jdbcType" : 1,"typeName" : "CHAR","typeExpression" : "CHAR","charsetName" : "latin1","length" : 50,"position" : 2,"optional" : false,"autoIncremented" : false,"generated" : false}, {"name" : "age","jdbcType" : 4,"typeName" : "INT","typeExpression" : "INT","charsetName" : null,"length" : 11,"position" : 3,"optional" : false,"autoIncremented" : false,"generated" : false}, {"name" : "num","jdbcType" : 4,"typeName" : "INT","typeExpression" : "INT","charsetName" : null,"length" : 11,"position" : 4,"optional" : false,"autoIncremented" : false,"generated" : false} ]}} ]
}

需要注意的是，结果的json格式是debezium定义好的格式，debezium json格式通常前面定义schema信息，最后才是实际的载荷（payload）信息，详细格式定义可以查看：https://debezium.io/documentation/reference/1.6/connectors/mysql.html

8）在userinfo表上做CUD操作语句，也可以看到相应的结果。

先执行消费者消费脚本：
./kafka-console-consumer.sh --bootstrap-server 168.61.38.11:9092,168.61.38.12:9092,168.61.38.13:9092,168.61.38.15:9092,168.61.38.16:9092 --topic conndbtest.test.userinfo
随便insert一条数据，如：

insert into userinfo(userid,username,age,num) values(54,'hhhhh',55,55656);

则可以消费到如下消息：

{"schema": {"type": "struct","fields": [{"type": "struct","fields": [{"type": "int32","optional": false,"field": "userid"},{"type": "string","optional": false,"field": "username"},{"type": "int32","optional": false,"field": "age"},{"type": "int32","optional": false,"field": "num"}],"optional": true,"name": "conndbtest.test.userinfo.Value","field": "before"},{"type": "struct","fields": [{"type": "int32","optional": false,"field": "userid"},{"type": "string","optional": false,"field": "username"},{"type": "int32","optional": false,"field": "age"},{"type": "int32","optional": false,"field": "num"}],"optional": true,"name": "conndbtest.test.userinfo.Value","field": "after"},{"type": "struct","fields": [{"type": "string","optional": false,"field": "version"},{"type": "string","optional": false,"field": "connector"},{"type": "string","optional": false,"field": "name"},{"type": "int64","optional": false,"field": "ts_ms"},{"type": "string","optional": true,"name": "io.debezium.data.Enum","version": 1,"parameters": {"allowed": "true,last,false"},"default": "false","field": "snapshot"},{"type": "string","optional": false,"field": "db"},{"type": "string","optional": true,"field": "sequence"},{"type": "string","optional": true,"field": "table"},{"type": "int64","optional": false,"field": "server_id"},{"type": "string","optional": true,"field": "gtid"},{"type": "string","optional": false,"field": "file"},{"type": "int64","optional": false,"field": "pos"},{"type": "int32","optional": false,"field": "row"},{"type": "int64","optional": true,"field": "thread"},{"type": "string","optional": true,"field": "query"}],"optional": false,"name": "io.debezium.connector.mysql.Source","field": "source"},{"type": "string","optional": false,"field": "op"},{"type": "int64","optional": true,"field": "ts_ms"},{"type": "struct","fields": [{"type": "string","optional": false,"field": "id"},{"type": "int64","optional": false,"field": "total_order"},{"type": "int64","optional": false,"field": "data_collection_order"}],"optional": true,"field": "transaction"}],"optional": false,"name": "conndbtest.test.userinfo.Envelope"},"payload": {"before": null,"after": {"userid": 54,"username": "hhhhh","age": 55,"num": 55656},"source": {"version": "1.6.1.Final","connector": "mysql","name": "conndbtest","ts_ms": 1637744766000,"snapshot": "false","db": "test","sequence": null,"table": "userinfo","server_id": 223344,"gtid": null,"file": "mysql-bin.000003","pos": 47627,"row": 0,"thread": null,"query": null},"op": "c","ts_ms": 1637744814011,"transaction": null}
}