Databus for Oracle

2022-11-03 16:07:50 浏览数 (1)

Databus组成

Relay ：数据抓取端

  • 读取数据源变更行（ROW），并将变更行数据序列化到内存缓冲区。
  • 监听客户端请求并传输缓冲区的数据。

Client：数据客户端

  • 检查Relay端的数据变更，并同步数据。
  • 如果与Relay之间数据变更相差太大、会执行追溯功能。

注：单个客户端既可以处理全部Databus数据流，也可以作为集群的一部分处理一小部分数据流。

bootstrap-producer：追溯数据生产端

  1. 这是一个特殊客户端，用于将Relay上的变更数据存储到Mysql数据库。
  2. Mysql数据库用于追溯数据。

bootstrap-server：追溯数据服务端

  1. 监听Client端执行追溯的功能请求。

databus-example for Oracle

这里记录运行databus自带的person的example的过程。

  1. 首先准备Oracle环境，这里我安装的oracle-xe 11g 。管理员密码orale，sid：XE，端口1521。
  2. databus是用gradle构建的java应用，所以本地首先需要安装gradle环境和java环境。gradle版本>=1.8即可。java版本>=1.6即可。
  3. 在github上下载databus源代码，解压解压好目录是databus-master。
  4. 在oracle官网下载ojdbc6.jar重命名放到如下目录：
代码语言：javascript复制
databus-master/sandbox-repo/com/oracle/ojdbc6/11.2.0.2.0/ojdbc6-11.2.0.2.0.jar

其实做好如上操作即可开始编译了，databus wiki是这么介绍了，不过看有人编译会指定utf-8编码，其实不需要，因为这里不涉及中文。

  1. 关于gradle版本，如果本地的gradle环境是5以前的，那么无所谓，如果是gradle5需要更改build.gradle文件。
代码语言：javascript复制
task wrapper(type: Wrapper) {
    gradleVersion = '1.8'
}

改为如下，否则编译的时候会报错：Cannot add task ‘wrapper’ as a task with that name already exists

代码语言：javascript复制
wrapper {
    gradleVersion = '1.8'
}

我本次编译使用版本是gradle5.1（试了最新的5.4.1编译太慢直接放弃了）

  1. 在com.linkedin.databus2.core.container.netty.ServerContainer的initializeContainerJmx()方法中添加如下内容。这里如果不更改在运行relay的时候会产生Cannot bind to URL rmi://localhost:1099 ServiceUnavailableException错误。
代码语言：javascript复制
LocateRegistry.createRegistry(_containerStaticConfig.getJmx().getRmiRegistryPort());

最后完整结果：

代码语言：javascript复制
  protected void initializeContainerJmx()
{ 

if (_containerStaticConfig.getJmx().isRmiEnabled())
{ 

try
{ 

JMXServiceURL jmxServiceUrl =
new JMXServiceURL("service:jmx:rmi://"  
_containerStaticConfig.getJmx().getJmxServiceHost()   ":"  
_containerStaticConfig.getJmx().getJmxServicePort()  "/jndi/rmi://"  
_containerStaticConfig.getJmx().getRmiRegistryHost()   ":"  
_containerStaticConfig.getJmx().getRmiRegistryPort()   "/jmxrmi"  
_containerStaticConfig.getJmx().getJmxServicePort());
LocateRegistry.createRegistry(_containerStaticConfig.getJmx().getRmiRegistryPort())
_jmxConnServer = JMXConnectorServerFactory.newJMXConnectorServer(jmxServiceUrl, null,
getMbeanServer());
}
catch (Exception e)
{ 

LOG.warn("Unable to instantiate JMX server", e);
}
}
}

注意首行导包：import java.rmi.registry.LocateRegistry;

  1. 修改reateUser在第51行
代码语言：javascript复制
 sqlplus sys/oracle@${DBNAME} as sysdba << __EOF__

这里我改为sys/oracle ，这是我的Oracle数据源的sysdba用户名密码。

  1. 修改sources-person.json,更改第4行，改为我们要连接的数据源信息。这里如果写错了那么运行clien的时候会报数据库连接错误。
代码语言：javascript复制
{ 

"name" : "person",
"id"  : 1,
"uri" : "jdbc:oracle:thin:person/person@localhost:1521:XE",
"slowSourceQueryThreshold" : 2000,
"sources" :
[
{ 
"id" : 101,
"name" : "com.linkedin.events.example.person.Person",
"uri": "person.person",
"partitionFunction" : "constant:1"
}
]
}

以上都操作好了即可编译了：

代码语言：javascript复制
gradle -Dopen_source=true assemble

java是向下兼容的，我用的jdk8，编译的时候只是会警告有些方法过期而已，是可以成功编译的。 编译好了会在build文件夹下生成很多编译后的文件。

在运行实例之前先要执行数据库脚本以及生成avro序列化文件

  1. 运行createUser
代码语言：javascript复制
 sh ./db/oracle/bin/createUser.sh person person xe tbs_person /u01/app/oracle/oradata/XE/person/
  1. 运行createSchema
代码语言：javascript复制
cd ./db/oracle/fwk
sh /home/databus-master/db/oracle/bin/createSchema.sh person/person@xe /home/databus-master/databus2-example/database/person/
  1. 生成avro序列化文件
代码语言：javascript复制
cd /home/databus-master/build/databus2-cmdline-tools-pkg/distributions
tar -zxvf databus2-cmdline-tools-pkg-2.0.0.tar.gz
cd bin
sh dbus2-avro-schema-gen.sh -namespace com.linkedin.events.example.person -recordName Person -viewName "sy$person" -avroOutDir /home/databus-master/myperson-avro/avroOut -avroOutVersion 1 -javaOutDir /home/databus-master/myperson-avro/javaOut -userName person -password person -database jdbc:oracle:thin:@localhost:1521:xe

接下即可运行测试了。

  1. 运行relay
代码语言：javascript复制
tar -zxvf build/databus2-example-relay-pkg/distributions/databus2-example-relay-pkg-2.0.0.tar.gz
sh start-example-relay.sh person
  1. 运行client
代码语言：javascript复制
tar -zxvf build/databus2-example-client-pkg/distributions/databus2-example-client-pkg-2.0.0.tar.gz
sh start-example-client.sh person
  1. 开始测试，登录Oracle的person用户执行插入或者更新语句：
代码语言：javascript复制
INSERT INTO person(id,first_name, last_name,birth_date) VALUES(1,'Trump', 'Donald',sysdate);
update person p set  p.BIRTH_DATE = sysdate -100 where p.ID = 1;
  1. 可以在clien.log看到捕获的信息

但是我发现执行delete语句是无法捕获的，查看数据库脚本发现，databus-example在建立触发器的时候，只建立了inser/update的触发器，没有delete相关的触发器。 继续查文档，看到这个文档Databus-2.0-Client-Event-Model-and-Consumer-API. 其中Data Change Events章节中有提到可以捕获update（包含insert）和delete两个操作应该也是可以捕获的。

到此为止，我们能够捕获Oracle数据库变更的行信息。但需要针对每个表新加字段txn，对每个表建立触发器。但捕获到了的数据不像ogg一样可以直接同步到目标库，目前来看这部分需要写代码单独处理。

网络上关于databus的文章资料比较少，我整理了一下有价值的资料地址，我的实验也参考了下面的文章：

  1. databus/wiki
  2. inkedin 的 databus 部署
  3. databus bootstrap 部署
  4. Databus架构分析与初步实践（for mysql）（上篇）
  5. Databus架构分析与初步实践（for mysql）（下篇）
  6. Databus部分翻译

版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。

发布者：全栈程序员栈长，转载请注明出处：https://javaforall.cn/181285.html原文链接：https://javaforall.cn

打包 gradle android java 编程算法

0 人点赞