Databus架构分析与初步实践（for mysql） | 网易乐得技术团队

标签： | 发表时间：2018-09-21 14:40 | 作者：
出处：http://tech.lede.com

1. 简介

Databus是一个低延迟、可靠的、支持事务的、保持一致性的数据变更抓取系统。由LinkedIn于2013年开源。Databus通过挖掘数据库日志的方式，将数据库变更实时、可靠的从数据库拉取出来，业务可以通过定制化client实时获取变更并进行其他业务逻辑。

Databus有以下特点：

  • 数据源和消费者之间的隔离。
  • 数据传输能保证顺序性和至少一次交付的高可用性。
  • 从变化流的任意时间点进行消费，包括通过bootstrap获取所有数据。
  • 分区消费
  • 源一致性保存，消费不成功会一直消费直到消费成功

2. 功能&特性

  • 来源独立：Databus支持多种数据来源的变更抓取，包括Oracle和MySQL。
  • 可扩展、高度可用：Databus能扩展到支持数千消费者和事务数据来源，同时保持高度可用性。
  • 事务按序提交：Databus能保持来源数据库中的事务完整性，并按照事务分组和来源的提交顺寻交付变更事件。
  • 低延迟、支持多种订阅机制：数据源变更完成后，Databus能在毫秒级内将事务提交给消费者。同时，消费者使用Databus中的服务器端过滤功能，可以只获取自己需要的特定数据。
  • 无限回溯：对消费者支持无限回溯能力，例如当消费者需要产生数据的完整拷贝时，它不会对数据库产生任何额外负担。当消费者的数据大大落后于来源数据库时，也可以使用该功能。

3. 使用场景举例

BUSSINESS1 和 BUSSINESS2 是两个不同的业务逻辑，他们的变更需要同时写入到 DB 和 CACHE ，那么当他们同时修改同一个数据的时候是否能保证数据的一致性呢？可以发现如果按照下图标明的顺序进行操作并不能保证数据的一致性！

1-1

还有一个问题是变更完DB之后，更新CACHE失败怎么办？如果忽略的话，会造成后续读取到CACHE中旧的数据，如果重试的话，业务代码会写得更加复杂。针对这些场景，如果没有一个强一致协议是很难解决掉的。如果要业务逻辑去实现这些晦涩的一致性协议，却又是不现实的。

现在，有了Databus，上面提到的这些一致性问题就都没有了，并且那些冗长的双写逻辑也可以去掉了，如下图所示:

1-1

4. 系统整体架构与主要组件

4.1 系统整体架构

1-1

上图中介绍了Databus系统的构成，包括Relays、bootstrap服务和Client lib等。Bootstrap服务中包括Bootstrap Producer和Bootstrap Server。快速变化的消费者直接从Relay中取事件。如果一个消费者的数据更新大幅落后，它要的数据就不在Relay的日志中，而是需要请求Bootstrap服务，返回的将会是自消费者上次处理变更之后的所有数据变更快照。

  • Source Databases：MySQL以及Oracle数据源
  • Relays：负责抓取和存储数据库变更，全内存存储，也可配置使用mmap内存映射文件方式
  • Schema Registry：数据库数据类型到Databus数据类型的一个转换表
  • Bootstrap Service：一个特殊的客户端，功能和Relays类似，负责存储数据库变更，主要是磁盘存储
  • Application：数据库变更消费逻辑，从Relay中拉取变更，并消费变更
  • Client Lib：提供挑选关注变更的API给消费逻辑
  • Consumer Code：变更消费逻辑，可以是自身消费或者再将变更发送至下游服务

4.2 主要组件及功能

上图系统整体架构图画的比较简略，下载源码观察项目结构后不难发现databus的主要由以下四个组件构成：

  • Databus Relay：
    • 从源数据库中的Databus源中读取变化的行并序列化为Databus变化事件保存到内存缓冲区中。
    • 监听Databus客户端的请求(包括引导程序的请求)并传输Databus数据变化事件。
  • Databus Client：
    • 在Relay上检查新的数据变化事件和处理特定的业务逻辑的回调。
    • 如果它们在relay后面落下太远，到引导程序服务运行一个追溯查询。
    • 单独的客户端可以处理全部的Databus流，它们也可以作为集群的一部分而每个客户端处理一部分流。
  • Databus Bootstrap Producer：
    • 只是一个特殊的客户端。
    • 检查Relay上的新的数据变化事件。
    • 保存数据变化事件到Mysql数据库，Mysql数据库用于引导程序和为了客户端追溯数据。
  • Databus Bootstrap Server：
    • 监听来自Databus客户端的请求并为了引导和追溯返回一个超长的回溯的数据变化事件。

5. Databus Relay和Databus Client详细分析

5.1 Databus Relay

5.1.1 架构与组件功能

1-1

  • Databus Event Producer(DBEP)：定期从数据库中查询变更，如果检测到变更，它将读取数据库中的所有已更改的行，并将其转换为Avro记录。因为数据库数据类型和Databus数据类型不一致，因此需要 Schema Registry 做转换。

  • SCN（System Change Number）：系统改变号，是数据库中非常重要的一个数据结构。SCN用以标识数据库在某个确切时刻提交的版本。在事务提交时，它被赋予一个唯一的标识事务的SCN。

  • Event Buffers：按照SCN的顺序存储databus事件，buffer可以是纯内存的，也可以是mmap到文件系统的。每个buffer在内存中还有一个对应的SCN Index和一个MaxSCN reader/writer，SCN Index可以加快查询指定事件的速度。

  • Request Processor：通过监听Netty的channel，实现收发client的请求。

  • MaxSCN Reader/Writer：用于跟踪DBEP的处理进度；Reader在Databus启动的时候会读取存储的文件上一次DBEP处理的位置，当Databus从DBEP中读取变更存储到Event Buffers时，Writer就会最后一个SCN写入到文件中存储，这样就能保证下次启动可以从正确的位置读取数据库变更。

  • JMX（Java Management Extensions）：支持标准的Jetty容器，databus提供了多个Mbean来监控relay

    • ContainerStatsMBean
    • DbusEventsTotalStatsMBean
    • DbusEventsStatisticsCollectorMBean

  • RESTFul Interface:Realy提供了相关http接口供外部调用，Client与Relay建立http长连接，并从Relay拉取Event。

5.1.2 源码分析

  • ServerContainer._globalStatsThread：统计信息的线程

  • OpenReplicatorEventProducer.EventProducerThread：针对mysql binlog日志的Event生产者线程，每个source一个线程，持有_orListener，管理和数据库的连接，将变更写入到Event Buffer里。

  • EventProducerThread启动后会初始化类型为OpenReplicator的日志解析对象开始解析日志，同时初始化类型为ORListener的_orListener开始监听，代码如下：

                    
    1
                    
    2
                    
    3
                    
    4
                    
    5
                    
    6
                    
    7
                    
    8
                    
    9
                    
    10
                    
    11
                    
    12
                    
    13
                    
    14
                    
    15
                    
    16
                    
    17
                    
    18
                    
    19
                    
    20
                    
    21
                    
    22
                    
    23
                    
    24
                    
    25
                    
    26
                    
    27
                    
    28
    		 
                    
    @Override
                    
    publicvoidrun()
                    
    {
                    
    _eventBuffer.start(_sinceScn);
                    
    _startPrevScn.set(_sinceScn);
                    
                    
    initOpenReplicator(_sinceScn);
                    
    try
                    
    {
                    
    booleanstarted =false;
                    
    while(!started) {
                    
    try{
                    
    _or.start();
                    
    started =true;
                    
    }
                    
    catch(Exception e) {
                    
    _log.error("Failed to start OpenReplicator: "+ e);
                    
    _log.warn("Sleeping for 1000 ms");
                    
    Thread.sleep(1000);
                    
    }
                    
    }
                    
    _orListener.start();
                    
    }catch(Exception e)
                    
    {
                    
    _log.error("failed to start open replicator: "+ e.getMessage(), e);
                    
    return;
                    
    }
                    
    }

    初始化方法如下：

                    
    1
                    
    2
                    
    3
                    
    4
                    
    5
                    
    6
                    
    7
                    
    8
                    
    9
                    
    10
                    
    11
                    
    12
                    
    13
                    
    14
                    
    15
                    
    16
                    
    17
                    
    18
                    
    19
                    
    20
                    
    21
    		 
                    
    voidinitOpenReplicator(longscn)
                    
    {
                    
    intoffset = offset(scn);
                    
    intlogid = logid(scn);
                    
                    
    String binlogFile = String.format("%s.%06d", _binlogFilePrefix, logid);
                    
    // we should use a new ORListener to drop the left events in binlogEventQueue and the half processed transaction.
                    
    _orListener =newORListener(_sourceName, logid, _log, _binlogFilePrefix, _producerThread, _tableUriToSrcIdMap,
                    
    _tableUriToSrcNameMap, _schemaRegistryService,200,100L);
                    
                    
    _or.setBinlogFileName(binlogFile);
                    
    _or.setBinlogPosition(offset);
                    
    _or.setBinlogEventListener(_orListener);
                    
                    
    //must set transport and binlogParser to null to drop the old connection environment in reinit case
                    
    _or.setTransport(null);
                    
    _or.setBinlogParser(null);
                    
                    
    _log.info("Connecting to OpenReplicator "+ _or.getUser() +"@"+ _or.getHost() +":"+ _or.getPort() +"/"
                    
    + _or.getBinlogFileName() +"#"+ _or.getBinlogPosition());
                    
    }

    EventProducerThread._orListener：监听数据库变更，将变更转换为Avro记录，写入到transaction里面，最终调用_producerThread的onEndTransaction()方法将事务里的事件写入到Event Buffer里，代码如下：

                    
    1
                    
    2
                    
    3
                    
    4
                    
    5
                    
    6
                    
    7
                    
    8
                    
    9
                    
    10
                    
    11
                    
    12
                    
    13
                    
    14
                    
    15
                    
    16
                    
    17
                    
    18
                    
    19
    		 
                    
    @Override
                    
    publicvoidonEndTransaction(Transaction txn)throwsDatabusException
                    
    {
                    
    try
                    
    {
                    
    addTxnToBuffer(txn);
                    
    _maxSCNReaderWriter.saveMaxScn(txn.getIgnoredSourceScn()!=-1? txn.getIgnoredSourceScn() : txn.getScn());
                    
    }
                    
    catch(UnsupportedKeyException e)
                    
    {
                    
    _log.fatal("Got UnsupportedKeyException exception while adding txn ("+ txn +") to the buffer", e);
                    
    thrownewDatabusException(e);
                    
    }
                    
    catch(EventCreationException e)
                    
    {
                    
    _log.fatal("Got EventCreationException exception while adding txn ("+ txn +") to the buffer", e);
                    
    thrownewDatabusException(e);
                    
    }
                    
    }

    FileMaxSCNHandler负责读写SCN，注意在写入文件时会将原有文件重命名为XXX.temp，原因是为了防止在更新文件的时候发生错误，导致SCN丢失，代码如下：

                    
    1
                    
    2
                    
    3
                    
    4
                    
    5
                    
    6
                    
    7
                    
    8
                    
    9
                    
    10
                    
    11
                    
    12
                    
    13
                    
    14
                    
    15
                    
    16
                    
    17
                    
    18
                    
    19
                    
    20
                    
    21
                    
    22
                    
    23
                    
    24
                    
    25
                    
    26
                    
    27
                    
    28
                    
    29
                    
    30
                    
    31
                    
    32
                    
    33
                    
    34
    		 
                    
    privatevoidwriteScnToFile()throwsIOException
                    
    {
                    
    longscn = _scn.longValue();
                    
                    
    File dir = _staticConfig.getScnDir();
                    
    if(! dir.exists() && !dir.mkdirs())
                    
    {
                    
    thrownewIOException("unable to create SCN file parent:"+ dir.getAbsolutePath());
                    
    }
                    
                    
    // delete the temp file if one exists
                    
    File tempScnFile =newFile(_scnFileName + TEMP);
                    
    if(tempScnFile.exists() && !tempScnFile.delete())
                    
    {
                    
    LOG.error("unable to erase temp SCN file: "+ tempScnFile.getAbsolutePath());
                    
    }
                    
                    
    File scnFile =newFile(_scnFileName);
                    
    if(scnFile.exists() && !scnFile.renameTo(tempScnFile))
                    
    {
                    
    LOG.error("unable to backup scn file");
                    
    }
                    
                    
    if(!scnFile.createNewFile())
                    
    {
                    
    LOG.error("unable to create new SCN file:"+ scnFile.getAbsolutePath());
                    
    }
                    
    FileWriter writer =newFileWriter(scnFile);
                    
    writer.write(Long.toString(scn));
                    
    writer.write(SCN_SEPARATOR +newDate().toString());
                    
    writer.flush();
                    
    writer.close();
                    
    LOG.debug("scn persisted: "+ scn);
                    
    }

  • 以源码例子中PersonRelayServer的主类启动为起点，大致的启动流程如下：

    PersonRelayServer主方法 -> new DatabusRelayMain实例 -> 调用initProducers方法初始化生产者->根据配置调用addOneProducer增加生产者->new DbusEventBufferAppendable获得Event Buffer->new EventProducerServiceProvider实例->
    调用createProducer获得OpenReplicatorEventProducer->OpenReplicatorEventProducer中包含
    EventProducerThread->启动线程开始获取Event

5.2 Databus Client

5.2.1 架构与组件功能

1-1

  • Relay Puller：负责从relay拉取数据，具体工作有挑选relay，请求source，请求Register，校验schema，设置dispatcher等。

  • Dispatcher：从event buffers中读取事件，调用消费逻辑的回调，主要职责有：

    • 判断回调是否正确，回调失败后会进行重试，重试次数超限后抛出异常
    • 监控错误和超时
    • 持久化checkpoint

  • Checkpoint persistence Provider：checkpoint是消费者消费变更记录点的位置，负责将checkpoint持久化到本地，保证下次启动后可以从正常的位置pull event。

  • Event Callback：调用消费者自定义业务逻辑代码。

  • Bootstrap Puller：负责从Bootstrap servers拉取数据，功能类似Relay Puller。

5.2.2 源码分析

执行Client的启动脚本后会调用main方法，main方法会根据命令行参数中指定的属性文件创建StaticConfig类，然后配置类创建dbusHttpClient实例来与Relay进行通信，参数defaultConfigBuilder为默认配置类信息，可以为空，代码如下：

            
1
            
2
            
3
            
4
            
5
            
6
            
7
            
8
            
9
            
10
            
11
            
12
            
13
            
14
 
            
publicstaticDatabusHttpClientImplcreateFromCli(String[] args, Config defaultConfigBuilder)throwsException
            
{
            
Properties startupProps =  ServerContainer.processCommandLineArgs(args);
            
if(null== defaultConfigBuilder) defaultConfigBuilder =newConfig();
            
            
ConfigLoader<StaticConfig> staticConfigLoader =
            
newConfigLoader<StaticConfig>("databus.client.", defaultConfigBuilder);
            
            
StaticConfig staticConfig = staticConfigLoader.loadConfig(startupProps);
            
            
DatabusHttpClientImpl dbusHttpClient =newDatabusHttpClientImpl(staticConfig);
            
            
returndbusHttpClient;
            
}

设置要连接的Relay信息，然后通过参数defaultConfigBuilder传递给dbusHttpClient，代码如下：

            
1
            
2
            
3
            
4
            
5
 
            
DatabusHttpClientImpl.Config configBuilder =newDatabusHttpClientImpl.Config();
            
configBuilder.getRuntime().getRelay("1").setHost("localhost");
            
configBuilder.getRuntime().getRelay("1").setPort(11115);
            
configBuilder.getRuntime().getRelay("1").setSources(PERSON_SOURCE);
            
}

启动databus client过程如下:

            
1
            
2
            
3
            
4
            
5
            
6
            
7
            
8
            
9
            
10
            
11
            
12
            
13
            
14
            
15
            
16
            
17
            
18
            
19
            
20
            
21
            
22
            
23
            
24
            
25
            
26
            
27
            
28
            
29
            
30
            
31
            
32
            
33
            
34
            
35
            
36
            
37
            
38
            
39
            
40
            
41
            
42
            
43
            
44
            
45
            
46
            
47
            
48
            
49
            
50
            
51
            
52
            
53
            
54
            
55
            
56
            
57
            
58
            
59
            
60
            
61
            
62
            
63
            
64
            
65
            
66
            
67
            
68
            
69
            
70
            
71
            
72
            
73
            
74
            
75
            
76
            
77
            
78
            
79
            
80
            
81
            
82
            
83
            
84
            
85
            
86
            
87
 
            
protectedvoiddoStart()
            
{
            
_controlLock.lock();
            
try
            
{
            
// 绑定并开始接收来到的连接
            
intportNum = getContainerStaticConfig().getHttpPort();
            
_tcpChannelGroup =newDefaultChannelGroup();
            
_httpChannelGroup =newDefaultChannelGroup();
            
            
_httpServerChannel = _httpBootstrap.bind(newInetSocketAddress(portNum));
            
InetSocketAddress actualAddress = (InetSocketAddress)_httpServerChannel.getLocalAddress();
            
_containerPort = actualAddress.getPort();
            
            
// 持久化端口号 (文件名对容器来说必须唯一)
            
File portNumFile =newFile(getHttpPortFileName());
            
portNumFile.deleteOnExit();
            
try{
            
FileWriter portNumFileW =newFileWriter(portNumFile);
            
portNumFileW.write(Integer.toString(_containerPort));
            
portNumFileW.close();
            
LOG.info("Saving port number in "+ portNumFile.getAbsolutePath());
            
}catch(IOException e) {
            
thrownewRuntimeException(e);
            
}
            
            
_httpChannelGroup.add(_httpServerChannel);
            
LOG.info("Serving container "+ getContainerStaticConfig().getId() +
            
" HTTP listener on port "+ _containerPort);
            
            
if(_containerStaticConfig.getTcp().isEnabled())
            
{
            
inttcpPortNum = _containerStaticConfig.getTcp().getPort();
            
_tcpServerChannel = _tcpBootstrap.bind(newInetSocketAddress(tcpPortNum));
            
_tcpChannelGroup.add(_tcpServerChannel);
            
            
LOG.info("Serving container "+ getContainerStaticConfig().getId() +
            
" TCP listener on port "+ tcpPortNum);
            
}
            
            
_nettyShutdownThread =newNettyShutdownThread();
            
Runtime.getRuntime().addShutdownHook(_nettyShutdownThread);
            
            
// 5秒后开始producer线程
            
if(null!= _jmxConnServer && _containerStaticConfig.getJmx().isRmiEnabled())
            
{
            
try
            
{
            
_jmxShutdownThread =newJmxShutdownThread(_jmxConnServer);
            
Runtime.getRuntime().addShutdownHook(_jmxShutdownThread);
            
            
_jmxConnServer.start();
            
LOG.info("JMX server listening on port "+ _containerStaticConfig.getJmx().getJmxServicePort());
            
}
            
catch(IOException ioe)
            
{
            
if(ioe.getCause() !=null&& ioe.getCause()instanceofNameAlreadyBoundException)
            
{
            
LOG.warn("Unable to bind JMX server connector. Likely cause is that the previous instance was not cleanly shutdown: killed in Eclipse?");
            
if(_jmxConnServer.isActive())
            
{
            
LOG.warn("JMX server connector seems to be running anyway. ");
            
}
            
else
            
{
            
LOG.warn("Unable to determine if JMX server connector is running");
            
}
            
}
            
else
            
{
            
LOG.error("Unable to start JMX server connector", ioe);
            
}
            
}
            
}
            
            
_globalStatsThread.start();
            
}
            
catch(RuntimeException ex)
            
{
            
LOG.error("Got runtime exception :"+ ex, ex);
            
throwex;
            
}
            
finally
            
{
            
_controlLock.unlock();
            
}
            
}

6. Databus for Mysql实践

6.1 相关解释

  • 实现原理：通过解析mysql的binlog日志来获取变更事件，解析过程利用Java开源工具OpenReplicator，Open Replicator 首先连接到MySQL（就像一个普通的MySQL Slave一样），然后接收和分析binlog，最终将分析得出的binlog events以回调的方式通知应用，所有的Event实现了BinlogEventV4接口。

  • binlog 格式：Databus设计为针对Row格式日志进行解析

    • Statement：基于SQL语句的复制(statement-based replication,SBR)
    • Row：基于行的复制(row-based replication,RBR)
    • Mixed：混合模式复制(mixed-based replication,MBR)

  • SCN的确定：64bits组成，高32位表示binlog的文件序号，低32位代表event在binlog文件的offset,例如在 mysql-bin.000001文件中 offset为 4的scn表示为(1 << 32) | 4 = 4294967300

6.2 数据库环境配置

  • 安装mysql数据库，本次使用mysql-5.5.56版本。

  • 查看数据库是否开启binlog,如果binlog没有开启，可以通过set sql_log_bin=1命令来启用;如果想停用binlog,可以使用set sql_log_bin=0。

    1-1

  • 配置数据库binlog_format=ROW， show variables like ‘binlog_format‘可查看日志格式， set globle binlog_format=ROW’可设置，通过修改my.cnf文件也可以，增加或修改行binlog_format=ROW即可。

    1-1

  • binlog_checksum设置为空，show global variables like ‘binlog_checksum’命令可查看，set binlog_checksum=none可设置。

    1-1

  • 在mysql上创建名为or_test的数据库，or_test上创建表名为person的表，定义如下：

    1-1

6.3 Demo配置与运行

6.3.1 下载源码

  • Databus官网下载源码，下载地址 https://github.com/linkedin/databus.git，我们需要用到databus目录下的databus2-example文件夹，在此基础上改造并运行，目录结构及介绍如下：

    1-1

    • database：数据库模拟相关的脚本和工具
    • databus2-example-bst-producer-pkg：bootstrap producer的属性配置文件夹，包括bootstrap producer和log4j属性文件，build脚本以及bootstrap producer的启动和停止脚本。
    • databus2-example-client-pkg：client的属性配置文件夹，包括各种属性文件和启动和停止脚本。
    • databus2-example-client：client源代码，包含启动主类和消费者代码逻辑。
    • databus2-example-relay-pkg：relay的属性配置文件夹，包含监控的表的source信息和Avro schema。
    • databus2-example-relay：relay的启动主类。
    • schemas_registry：存放表的avsc文件。

6.3.2 Relay端的操作

  • 配置Relay属性文件：databus2-example-relay-pkg/conf/relay-or-person.properties的内容如下配置，包括端口号，buffer存储策略，maxScn存放地址等信息：

                    
    1
                    
    2
                    
    3
                    
    4
                    
    5
                    
    6
                    
    7
                    
    8
                    
    9
                    
    10
                    
    11
                    
    12
                    
    13
                    
    14
    		 
                    
    databus.relay.container.httpPort=11115
                    
    databus.relay.container.jmx.rmiEnabled=false
                    
    databus.relay.eventBuffer.allocationPolicy=DIRECT_MEMORY
                    
    databus.relay.eventBuffer.queuePolicy=OVERWRITE_ON_WRITE
                    
    databus.relay.eventLogReader.enabled=false
                    
    databus.relay.eventLogWriter.enabled=false
                    
    databus.relay.schemaRegistry.type=FILE_SYSTEM
                    
    databus.relay.schemaRegistry.fileSystem.schemaDir=./schemas_registry
                    
    databus.relay.eventBuffer.maxSize=1024000000
                    
    databus.relay.eventBuffer.readBufferSize=10240
                    
    databus.relay.eventBuffer.scnIndexSize=10240000
                    
    databus.relay.physicalSourcesConfigsPattern=../../databus2-example/databus2-example-relay-pkg/conf/sources-or-person.json
                    
    databus.relay.dataSources.sequenceNumbersHandler.file.scnDir=/tmp/maxScn
                    
    databus.relay.startDbPuller=true

  • 配置被监控表的source信息：databus2-example-relay-pkg/conf/sources-or-person.json的内容如下配置，其中URI format:mysql://username/password@mysql_host[:mysql_port]/mysql_serverid/binlog_prefix，注意%2F为转义字符，用户名为root,数据库密码为123。

                    
    1
                    
    2
                    
    3
                    
    4
                    
    5
                    
    6
                    
    7
                    
    8
                    
    9
                    
    10
                    
    11
                    
    12
                    
    13
                    
    14
                    
    15
    		 
                    
    {
                    
    "name" : "person",
                    
    "id"  : 1,
                    
    "uri" : "mysql://root%2F123@localhost:3306/1/mysql-bin",
                    
    "slowSourceQueryThreshold" : 2000,
                    
    "sources" :
                    
    [
                    
    {
                    
    "id" : 40,
                    
    "name" : "com.linkedin.events.example.or_test.Person",
                    
    "uri": "or_test.person",
                    
    "partitionFunction" : "constant:1"
                    
    }
                    
    ]
                    
    }

  • databus2-example-relay-pkg/schemas_registry/下定义person的Avro schema文件
    com.linkedin.events.example.or_test.Person.1.avsc，其中1表示版本(Databus目前没有针对mysql提供生成Avro schema文件的工具，所以只能手工编写)具体内容如下所示：

                    
    1
                    
    2
                    
    3
                    
    4
                    
    5
                    
    6
                    
    7
                    
    8
                    
    9
                    
    10
                    
    11
                    
    12
                    
    13
                    
    14
                    
    15
                    
    16
                    
    17
                    
    18
                    
    19
                    
    20
                    
    21
                    
    22
                    
    23
                    
    24
                    
    25
                    
    26
                    
    27
                    
    28
    		 
                    
    {
                    
    "name" : "Person_V1",
                    
    "doc" : "Auto-generated Avro schema for sy$person. Generated at Dec 04, 2012 05:07:05 PM PST",
                    
    "type" : "record",
                    
    "meta" : "dbFieldName=person;pk=id;",
                    
    "namespace" : "com.linkedin.events.example.or_test",
                    
    "fields" : [ {
                    
    "name" : "id",
                    
    "type" : [ "long", "null" ],
                    
    "meta" : "dbFieldName=ID;dbFieldPosition=0;"
                    
    }, {
                    
    "name" : "firstName",
                    
    "type" : [ "string", "null" ],
                    
    "meta" : "dbFieldName=FIRST_NAME;dbFieldPosition=1;"
                    
    }, {
                    
    "name" : "lastName",
                    
    "type" : [ "string", "null" ],
                    
    "meta" : "dbFieldName=LAST_NAME;dbFieldPosition=2;"
                    
    }, {
                    
    "name" : "birthDate",
                    
    "type" : [ "long", "null" ],
                    
    "meta" : "dbFieldName=BIRTH_DATE;dbFieldPosition=3;"
                    
    }, {
                    
    "name" : "deleted",
                    
    "type" : [ "string", "null" ],
                    
    "meta" : "dbFieldName=DELETED;dbFieldPosition=4;"
                    
    } ]
                    
    }

  • 注册Avro schema到index.schemas_registry文件,databus2-example-relay-pkg/schemas_registry/index.schemas_registry文件中添加行com.linkedin.events.example.or_test.Person.1.avsc ，每定义一个Avro schema都需要添加进去，relay运行时会到此文件中查找表对应的定义的Avro schema。

6.3.3 Client端的操作

  • 配置Client属性文件：databus2-example-client-pkg/conf/client-person.properties的内容如下配置，包括端口号，buffer存储策略，checkpoint持久化等信息：

                    
    1
                    
    2
                    
    3
                    
    4
                    
    5
                    
    6
                    
    7
                    
    8
                    
    9
                    
    10
                    
    11
                    
    12
    		 
                    
    databus.relay.container.httpPort=11125
                    
    databus.relay.container.jmx.rmiEnabled=false
                    
    databus.relay.eventBuffer.allocationPolicy=DIRECT_MEMORY
                    
    databus.relay.eventBuffer.queuePolicy=BLOCK_ON_WRITE
                    
    databus.relay.schemaRegistry.type=FILE_SYSTEM
                    
    databus.relay.eventBuffer.maxSize=10240000
                    
    databus.relay.eventBuffer.readBufferSize=1024000
                    
    databus.relay.eventBuffer.scnIndexSize=1024000
                    
    databus.client.connectionDefaults.pullerRetries.initSleep=1
                    
    databus.client.checkpointPersistence.fileSystem.rootDirectory=./personclient-checkpoints
                    
    databus.client.checkpointPersistence.clearBeforeUse=false
                    
    databus.client.connectionDefaults.enablePullerMessageQueueLogging=true

  • databus2-example-client/src/main/java下的PersonConsumer类是消费逻辑回调代码，主要是取出每一个event后依次打印每个字段的名值对，主要代码如下：

                    
    1
                    
    2
                    
    3
                    
    4
                    
    5
                    
    6
                    
    7
                    
    8
                    
    9
                    
    10
                    
    11
                    
    12
                    
    13
                    
    14
                    
    15
                    
    16
                    
    17
                    
    18
                    
    19
                    
    20
    		 
                    
    privateConsumerCallbackResultprocessEvent(DbusEvent event, DbusEventDecoder eventDecoder)
                    
    {
                    
    GenericRecord decodedEvent = eventDecoder.getGenericRecord(event,null);
                    
    try{
                    
    Utf8 firstName = (Utf8)decodedEvent.get("firstName");
                    
    Utf8 lastName = (Utf8)decodedEvent.get("lastName");
                    
    Long birthDate = (Long)decodedEvent.get("birthDate");
                    
    Utf8 deleted = (Utf8)decodedEvent.get("deleted");
                    
                    
    LOG.info("firstName: "+ firstName.toString() +
                    
    ", lastName: "+ lastName.toString() +
                    
    ", birthDate: "+ birthDate +
                    
    ", deleted: "+ deleted.toString());
                    
    }catch(Exception e) {
                    
    LOG.error("error decoding event ", e);
                    
    returnConsumerCallbackResult.ERROR;
                    
    }
                    
                    
    returnConsumerCallbackResult.SUCCESS;
                    
    }

  • databus2-example-client/src/main/java下的PersonClient类是relay的启动主类，主要是设置启动Client的配置信息，将消费者实例注册到监听器中，后续可对其进行回调，主要代码如下：

                    
    1
                    
    2
                    
    3
                    
    4
                    
    5
                    
    6
                    
    7
                    
    8
                    
    9
                    
    10
                    
    11
                    
    12
                    
    13
                    
    14
                    
    15
                    
    16
                    
    17
                    
    18
                    
    19
                    
    20
    		 
                    
    publicstaticvoidmain(String[] args)throwsException
                    
    {
                    
    DatabusHttpClientImpl.Config configBuilder =newDatabusHttpClientImpl.Config();
                    
                    
    //Try to connect to a relay on localhost
                    
    configBuilder.getRuntime().getRelay("1").setHost("localhost");
                    
    configBuilder.getRuntime().getRelay("1").setPort(11115);
                    
    configBuilder.getRuntime().getRelay("1").setSources(PERSON_SOURCE);
                    
                    
    //Instantiate a client using command-line parameters if any
                    
    DatabusHttpClientImpl client = DatabusHttpClientImpl.createFromCli(args, configBuilder);
                    
                    
    //register callbacks
                    
    PersonConsumer personConsumer =newPersonConsumer();
                    
    client.registerDatabusStreamListener(personConsumer,null, PERSON_SOURCE);
                    
    client.registerDatabusBootstrapListener(personConsumer,null, PERSON_SOURCE);
                    
                    
    //fire off the Databus client
                    
    client.startAndBlock();
                    
    }

6.3.4 build-启动-测试

  • Build:Databus采用gradle进行编译，所以需要安装gradle环境，安装安成后进入databus根目录执行命令gradle -Dopen_source=true assemble即可完成build,成功后在databus根目录下生成名为build的文件夹

  • 启动Relay:

    1. cd build/databus2-example-relay-pkg/distributions
    2. tar -zxvf databus2-example-relay-pkg.tar.gz解压
    3. 执行启动脚本 ./bin/start-example-relay.sh or_person -Y ./conf/sources-or-person.json

    4. 执行命令 curl -s http://localhost:11115/sources返回如下内容说明启动成功：

      1-1

  • 启动Client:

    1. cd build/databus2-example-client-pkg/distributions
    2. tar -zxvf databus2-example-client-pkg.tar.gz解压
    3. 执行启动脚本 ./bin/start-example-client.sh person

    4. 执行命令 curl http://localhost:11115/relayStats/outbound/http/clients返回如下内容说明启动成功：

      1-1

  • 测试:

    Relay和Client启动成功后，就已经开始对person表进行数据变更捕获了，现在向person表插入一条如下记录：

    1-1

    databus2-example-relay-pkg/distributions/logs下的relay.log记录如下：

    1-1

    databus2-example-client-pkg/distributions/logs下的client.log记录如下：

    1-1

    可以看到已经可以抓取到改变的数据了！

7. 总结

遇到的问题：

  • 主要是属性文件的配置问题，包括source-or-person.json, schemas_registry的文件缺失或配置错误。
  • 脚本方式启动时JVM无法创建，由于脚本启动时包含了自定义的JVM参数，与系统环境不符导致启动失败，去掉相关参数后正常启动。
  • Relay可以获取增删改查的Event,但Client只能解析到更新操作的Event，主要原因是Mysql默认的binlog_format=MIXED,而databus的设计是针对ROW格式的binlog，修改格式后可正常解析。
  • Windows平台无法使用，启动方式是用脚本启动，脚本启动时包含命令行参数较多，启动后无法进行调试，只能通过对日志观察的方式来进行。

需要进一步实验：

  • 使用bootstrap produces和bootstrap servers模式来进行大批量事件的获取
  • 配置多个relay进行事件抓取
  • 结合zookeeper来配置客户端集群进行消费

参考资料

相关 [databus 架构 分析] 推荐：

Databus架构分析与初步实践（for mysql） | 网易乐得技术团队

- -
Databus是一个低延迟、可靠的、支持事务的、保持一致性的数据变更抓取系统. 由LinkedIn于2013年开源. Databus通过挖掘数据库日志的方式，将数据库变更实时、可靠的从数据库拉取出来，业务可以通过定制化client实时获取变更并进行其他业务逻辑. 数据传输能保证顺序性和至少一次交付的高可用性.

Databus for MySQL 同步 · linkedin/databus Wiki · GitHub

- -
A frequently asked question on the Databus open source mailing list is about the possibility of capturing changes in MySQL through Databus. $ (cd bin && ./create_person.sh): The script assumes that MySQL is started on port 33066; please change it appropriately for your setup.

监听mysql的binlog日志工具分析：canal、Maxwell、Databus、DTS - 程序员大本营

- -
阿里云的数据传输服务DTS. 定位：基于数据库增量日志解析，提供增量数据订阅&消费，目前主要支持了mysql. canal模拟mysql slave的交互协议，伪装自己为mysql slave，向mysql master发送dump协议. mysql master收到dump请求，开始推送binary log给slave(也就是canal).

GFS架构分析

- zou - NOSQL Notes
Google文件系统（Google File System，GFS）是构建在廉价的服务器之上的大型分布式系统. 它将服务器故障视为正常现象，通过软件的方式自动容错，在保证系统可靠性和可用性的同时，大大减少了系统的成本. GFS是Google云存储的基石，其它存储系统，如Google Bigtable，Google Megastore，Google Percolator均直接或者间接地构建在GFS之上.

Instagram 架构分析笔记

- Yousri - DBA Notes
Instagram 团队上个月才迎来第 7 名员工，是的，7个人的团队. 作为 iPhone 上最火爆的图片类工具，instagram 用户数量已经超过 1400 万，图片数量超过 1.5 亿张. 不得不说，这真他妈是个业界奇迹. 几天前，只有三个人的 Instagram 工程师团队发布了一篇文章：What Powers Instagram: Hundreds of Instances, Dozens of Technologies，披露了 Instagram 架构的一些信息，足够勾起大多数人的好奇心.

Android 系统架构分析

- - CSDN博客移动开发推荐文章
Android：开源的 Linux + Google 的封闭软件 + 私有的基带 + 运营商锁定 = 开放的 Android 手机. iPhone：开源的 BSD + 苹果的闭源软件 + 私有的基带 + 运营商锁定 = 封闭的苹果 iPhone. 一个平庸的应用商店，开发者依靠广告赚钱，商店并非独此一家，用户找不到好软件.

twitter系统架构分析

- - 企业架构 - ITeye博客
twitter系统架构分析. （一）twitter的核心业务. twitter的核心业务，在于following和be followed：. （1）following-关注. 进入个人主页，会看到你follow的人发表的留言（不超过140个字），这是following的过程；. （2）followed-被关注.

LinkedIn实时低延迟数据抓取系统Databus开源

- - InfoQ cn
去年的架构师峰会上，来自LinkedIn的高级软件工程师 Lei Gao做了一场名为 《LinkedIn的数据处理架构》的演讲，着重介绍LinkedIn内部的数据基础设施的演变，其中提到Databus数据总线项目，当时就引起大家诸多好奇. 前不久，LinkedIn工程团队官方博客 发布消息：Databus项目开源.

LinkedIn开源低延时变化数据捕获系统Databus

- - CSDN最新资讯
LinkedIn于2月26日开源了其低延时变化数据捕获系统Databus，该系统在低延时情况下仍然具备高有效性. 能服务于LinkedIn生态系统数年之久，除上述特性以外当然还有其不可忽视的独到之处——无限制的lookback能力及丰富的subscription功能. 文章在最后公布了Databus源码及相关下载.

基于Binlog的实时同步功能——debezium、canel、databus技术选型 | holmofy

- -
去年的一篇文章大致地讲了我对MQ的一些认识，事实上Kafka在内的现代MQ，功能远不止这些. 后面整理好自己的思路，肯定会再写一篇文章来讲讲. 这篇文章的主角就是与MQ息息相关的CDC技术. CDC全称叫：change data capture，是一种基于数据库数据变更的事件型软件设计模式. 比如有一张订单表trade，订单每一次变更录入到一张trade_change的队列表.