当今互联网时代，数据作为数字时代最重要的资产，不论是海量大数据还是小而精细的数据，都非常宝贵。在企业级应用中，多数据库间数据的同步一直是一个难题。本文详细讲解了如何采用MySQL实时同步到Oracle的解决方案。

详解MySQL实时同步到Oracle解决方案

前言

当今互联网时代，数据作为数字时代最重要的资产，不论是海量大数据还是小而精细的数据，都非常宝贵。在企业级应用中，多数据库间数据的同步一直是一个难题。本文详细讲解了如何采用MySQL实时同步到Oracle的解决方案。

解决方案

MySQL与Oracle都是业内常用的数据库系统。MySQL的优势在于易安装、易配置、易扩展，而Oracle则更善于处理复杂商业逻辑、拥有更强大的高可用性和高性能等特点。

同步MySQL到Oracle的解决方案，常常采用以下两种方式。

利用第三方工具进行同步

借助第三方工具，利用其提供API或配置文件来实现MySQL数据同步到Oracle。

常用第三方工具：

Apache Nifi：Web UI可视化管理，支持多种数据源，支持批量/流式处理；
Debezium：开源工具，基于Apache Kafka，支持MySQL、PostgreSQL等数据源多样化（也支持Oracle），支持增量更新；
Maxwell：开源工具，依赖于MySQL二进制日志（binlog），支持Kafka和Kinesis等订阅器；

示例：使用Maxwell同步MySQL数据到Oracle。

使用如下命令下载并安装Maxwell。

$ curl -L https://github.com/zendesk/maxwell/releases/download/v1.19.1/maxwell-1.19.1.tar.gz | tar zxvf -

编辑maxwell.properties文件。

# 同步MySQL连接配置
maxwell.host = 127.0.0.1
maxwell.port = 3306
maxwell.user = maxwell
maxwell.password = XXXXX
maxwell.schemas = pcmall

# Oracle连接配置
maxwell.target.type = oracle
maxwell.target.host = 127.0.0.1
maxwell.target.port = 1521
maxwell.target.user = maxwell
maxwell.target.password = XXXXX
maxwell.target.database = maxwell
maxwell.target.sid = sid

# ID字段设置，将MySQL的id字段，在Oracle中命名为syc_id
producer.request.required.acks = 0
producer.request.timeout.ms = 5000
producer.flush.interval.ms = 1000
#bootstrap.servers=localhost:9092
producer.type = sync
producer.kafka.bootstrap.servers = localhost:9092
producer.kafka.topic = maxwell
producer.kafka.compression.type = snappy
producer.kafka.client.id = maxwell-1
producer.kafka.max.request.size = 31457280
#producer.ddl.kafka.bootstrap.servers = localhost:9092
producer.ddl = true
producer.include-tables = taskflow_chokes,taskflows
producer.exclude-tables = amx_task_instances,offline_task_flows
column.includes=.*
# 经过重命名的列到列名映射关系
column.rename.id = sync_id
filter.ddl = true

运行以下命令启动同步服务。

./bin/maxwell --config ./config/config.properties

自定义同步插件

MySQL的binlog技术，可以轻松地获取MySQL中的数据变化，同时维护了数据传输的可靠性和高效性。针对mysql的binlog，社区已经提供很多基于binlog的工具，如Canal、开源的Ali-DataX，业界都有一些企业级同步插件，如阿里云的DTS、腾讯云的DTX、华为云的DMS SQL server等。

基于Canal进行自定义插件同步的好处是：

针对自己的业务可以做定制化开发；
依托于Canal已经做好的binlog解析，不需要重复造轮子；
Canal官方提供丰富的插件应用案例，可直接参考学习。

示例：使用Canal+自定义插件同步MySQL数据到Oracle。

首先使用以下命令拉取Canal项目源码。

$ git clone https://github.com/alibaba/canal.git canal
$ cd canal && mvn clean install -DskipTests

编辑Canal的启动脚本bin/startup.sh，将其修改为如下配置

#!/usr/bin/env bash

cd `dirname $0`/..
CURRENT_DIR=`pwd`
MODULES_DIR=${CURRENT_DIR}/modules

# pidfile do not change line
pidfile=${CURRENT_DIR}/logs/canal.pid

# JAVA_OPTS do not change line
JAVA_OPTS="-server -Xms2g -Xmx4g -Djava.awt.headless=true -Dfile.encoding=UTF-8"
JAVA_DEBUG_OPTS=""
if [ "$1" = "debug" ]; then
    JAVA_DEBUG_OPTS=" -Xdebug -Xrunjdwp:transport=dt_socket,address=8787,server=y,suspend=n "
fi
CANAL_CONF_DIR=${CURRENT_DIR}/conf/
if [ -z "$LOG_HOME" ];then
    LOG_HOME=${CURRENT_DIR}/logs
fi

# rm log soft link
if [ -L ${CURRENT_DIR}/logs/canal.log ];then
    rm -f ${CURRENT_DIR}/logs/canal.log
fi

# root 处理
if [ "`whoami`" = "root" ]; then
    echo "canal can't be start as root!"
    echo "canal can't be start as root!" > ${LOG_HOME}/canal.log 2>&1
    exit 1
fi

# set env
if [ -f /etc/profile ];then
    . /etc/profile
fi
. ${CURRENT_DIR}/bin/env.sh

#set JAVA_HOME directory,if JAVA_HOME have set in environment variable,it will use it.
if [ "$JAVA_HOME" = "" ];then
    JAVA_HOME=${HOME}/public/app/jdk1.8.0_121
fi

if [ ! -d "$JAVA_HOME" ]; then
    echo "JAVA_HOME $JAVA_HOME not exists, please set right."
    echo "JAVA_HOME $JAVA_HOME not exists, please set right." > ${LOG_HOME}/canal.log 2>&1
    exit 1
fi

if [ "$1" = "start" ]; then
    if [ -f "${pidfile}" ]; then
            if kill -0 `cat "${pidfile}"` > /dev/null 2>&1; then
                echo $command already running as process `cat "${pidfile}"`.
                echo $command already running as process `cat "${pidfile}"`. > ${LOG_HOME}/canal.log 2>&1
                exit 1
            fi
    fi

    if [ ! -d "$LOG_HOME" ]; then
        mkdir "$LOG_HOME"
    fi

    # check dir ,if not exists create dir
    if [ ! -d "$MODULES_DIR" ]; then
        mkdir "$MODULES_DIR"
    fi

    if [ -z "$CANAL_CONF_DIR" ];then
        CANAL_CONF_DIR=${CURRENT_DIR}/conf/
    fi

    CLASSPATH="."
    for i in "$CANAL_CONF_DIR"/lib/*.jar;do
        CLASSPATH="$CLASSPATH":"$i"
    done

    for i in "$CURRENT_DIR"/lib/*.jar;do
        CLASSPATH="$CLASSPATH":"$i"
    done

    for i in "$MODULES_DIR"/*/lib/*.jar;do
        CLASSPATH="$CLASSPATH":"$i"
    done

    echo "using CLASSPATH:":$CLASSPATH

    export CLASSPATH

    # 启动server
    nohup ${JAVA_HOME}/bin/java ${JAVA_OPTS} ${JAVA_DEBUG_OPTS} \
    -Dlog4j.configurationFile=file:${CANAL_CONF_DIR}/log4j2.xml \
    -Dcanal.instance.global.mode=${mode} \
    -Dcanal.conf.dir=${CANAL_CONF_DIR} \
    -Dcanal.instance.mysql.slaveId=${slaveId} \
    -Dcanal.instance.master.address=${master} \
    -Dcanal.instance.dbUsername=${dbUsername} \
    -Dcanal.instance.dbPassword=${dbPassword} \
    -Dcanal.instance.connectionCharset=${connectionCharset} \
    -Dcanal.instance.tsdb.enable=${tsdb_enable} \
    -Dcanal.instance.gtidon=${gtidon} \
    -Dcanal.instance.filter.regex=${filter_regex} \
    -Dcanal.instance.filter.black.regex=${filter_black_regex} \
    -Dcanal.instance.binlog.sniffer=false \
    com.alibaba.otter.canal.deployer.CanalLauncher > "$LOG_HOME"/canal.log 2>&1 &

    if [ $? -eq 0 ]; then
        if /bin/echo -n $! > "${pidfile}"; then
            sleep 1
            echo "start OK"
            echo "start OK" > ${LOG_HOME}/canal.log 2>&1
        else
            echo "start failed"
            echo "start failed" > ${LOG_HOME}/canal.log 2>&1
            exit 1
        fi
    else
        echo "start failed"
        echo "start failed" > ${LOG_HOME}/canal.log 2>&1
        exit 1
    fi
fi

Canal启动后默认使用文件存储方式存储instance配置信息。但该方式不利于在生产环境中对配置进行修改和维护。因此，选择使用Canal Admin管理Canal实例。

使用以下命令下载并安装Canal Admin。

$ curl -L https://github.com/yinanfang/canal-admin-bin/releases/download/1.0.11/canal.admin-1.0.11.tar.gz | tar zxvf -
$ cd canal.admin-1.0.11/bin/
$ sh startup.sh

在Canal Admin页面创建实例，将MySQL和Oracle的相关配置填写。

创建完成后添加相应的自定义插件进行同步操作。