Debezium同步之oracle数据到Kafka的同步

Oracle 数据库可以作为独立实例安装，也可以使用 Oracle Real Application Cluster (RAC) 安装。Debezium Oracle 连接器与这两种安装类型兼容。

此外，必须为捕获的表或数据库启用补充日志记录，以便数据更改能够捕获已更改数据库行的之前状态。下面说明了如何在特定表上进行配置，这是最大限度减少 Oracle 重做日志中捕获的信息量的理想选择。

必须在数据库级别启用最小补充日志记录，并且可以按如下方式进行配置。

根据数据库配置，重做日志的大小和数量可能不足以实现可接受的性能。在设置 Debezium Oracle 连接器之前，请确保重做日志的容量足以支持数据库。

数据库的重做日志容量必须足以存储其数据字典。一般来说，数据字典的大小随着数据库中表和列的数量而增加。如果重做日志容量不足，数据库和 Debezium 连接器都可能会遇到性能问题。

请咨询您的数据库管理员以评估数据库是否可能需要增加日志容量。

为了让 Debezium Oracle 连接器捕获更改事件，它必须以具有特定权限的 Oracle LogMiner 用户身份运行。以下示例显示了用于在多租户数据库模型中为连接器创建 Oracle 用户帐户的 SQL。

{
    "name": "inventory-connector",  
    "config": {
        "connector.class" : "io.debezium.connector.oracle.OracleConnector",  
        "database.hostname" : "<ORACLE_IP_ADDRESS>",  
        "database.port" : "1521",  
        "database.user" : "c##dbzuser",  
        "database.password" : "dbz",   
        "database.dbname" : "ORCLCDB",  
        "database.server.name" : "server1",  
        "tasks.max" : "1",  
        "database.pdb.name" : "ORCLPDB1",  
        "database.history.kafka.bootstrap.servers" : "kafka:9092", 
        "database.history.kafka.topic": "schema-changes.inventory"  
    }
}

{
    "name": "inventory-connector",
    "config": {
        "connector.class" : "io.debezium.connector.oracle.OracleConnector",
        "tasks.max" : "1",
        "database.server.name" : "server1",
        "database.user" : "c##dbzuser",
        "database.password" : "dbz",
        "database.url": "jdbc:oracle:thin:@(DESCRIPTION=(ADDRESS_LIST=(LOAD_BALANCE=OFF)(FAILOVER=ON)(ADDRESS=(PROTOCOL=TCP)(HOST=<oracle ip 1>)(PORT=1521))(ADDRESS=(PROTOCOL=TCP)(HOST=<oracle ip 2>)(PORT=1521)))(CONNECT_DATA=SERVICE_NAME=)(SERVER=DEDICATED)))",
        "database.dbname" : "ORCLCDB",
        "database.pdb.name" : "ORCLPDB1",
        "database.history.kafka.bootstrap.servers" : "kafka:9092",
        "database.history.kafka.topic": "schema-changes.inventory"
    }
}

{
  "config": {
    "connector.class" : "io.debezium.connector.oracle.OracleConnector",
    "tasks.max" : "1",
    "database.server.name" : "server1",
    "database.hostname" : "<oracle ip>",
    "database.port" : "1521",
    "database.user" : "c##dbzuser",
    "database.password" : "dbz",
    "database.dbname" : "ORCLCDB",
    "database.pdb.name" : "ORCLPDB1",
    "database.history.kafka.bootstrap.servers" : "kafka:9092",
    "database.history.kafka.topic": "schema-changes.inventory"
  }
}

{
    "config": {
        "connector.class" : "io.debezium.connector.oracle.OracleConnector",
        "tasks.max" : "1",
        "database.server.name" : "server1",
        "database.hostname" : "<oracle ip>",
        "database.port" : "1521",
        "database.user" : "c##dbzuser",
        "database.password" : "dbz",
        "database.dbname" : "ORCLCDB",
        "database.history.kafka.bootstrap.servers" : "kafka:9092",
        "database.history.kafka.topic": "schema-changes.inventory"
    }
}