Spring

canal.deployer-1.1.7-SNAPSHOT.tar.gz为Canal软件压缩包需要安装在服务器上并根据下文进行配置文件的修改。

CanalClient.rar为用Spring

二、.

在配置Canal前需要确保Mysql的Binlog已经开启并且模式为ROW找到当前binlog的文件名和position。

配置文件路径canal/conf/canal.properties

配置文件路径

canal/conf/example/instance.properties

Spring

com.canal.canalclient.config;import

com.alibaba.otter.canal.client.CanalConnector;

import

com.alibaba.otter.canal.client.CanalConnectors;

import

com.alibaba.otter.canal.protocol.CanalEntry;

import

com.alibaba.otter.canal.protocol.CanalEntry.Column;

import

com.alibaba.otter.canal.protocol.CanalEntry.EntryType;

import

com.alibaba.otter.canal.protocol.CanalEntry.EventType;

import

com.alibaba.otter.canal.protocol.CanalEntry.RowChange;

import

com.alibaba.otter.canal.protocol.Message;

import

com.google.protobuf.InvalidProtocolBufferException;

import

org.springframework.jdbc.core.JdbcTemplate;import

import

java.util.concurrent.ConcurrentLinkedQueue;

/***

{//获取canalServer连接本机地址,端口号CanalConnector

connector

CanalConnectors.newSingleConnector(new

端口号),

{//连接canalServerconnector.connect();//订阅Desctinstionconnector.subscribe();connector.rollback();try

{while

{//尝试从master那边拉去数据batchSize条记录有多少取多少//轮询拉取数据

message

connector.getWithoutAck(batchSize);long

batchId

{dataHandle(message.getEntries());}connector.ack(batchId);System.out.println(aasize);//当队列里面堆积的sql大于一定数值的时候就模拟执行if

(SQL_QUEUE.size()

(InvalidProtocolBufferException

finally

{connector.disconnect();}}public

static

SQL_QUEUE.poll();jdbcTemplate.execute(sql);System.out.println([sql]----

sql);}}/***

dataHandle(ListCanalEntry.Entry

entrys)

RowChange.parseFrom(entry.getStoreValue());CanalEntry.EventType

eventType

RowChange.parseFrom(entry.getStoreValue());ListCanalEntry.RowData

rowDatasList

rowChange.getRowDatasList();for

(CanalEntry.RowData

rowData.getAfterColumnsList();StringBuffer

sql

entry.getHeader().getSchemaName()

entry.getHeader().getTableName()

set

newColumnList.get(i).getValue()

);if

rowData.getBeforeColumnsList();for

(Column

{//暂时只支持单一主键sql.append(column.getName()

column.getValue());break;}}SQL_QUEUE.add(sql.toString());}}

catch

(InvalidProtocolBufferException

保存删除语句**

RowChange.parseFrom(entry.getStoreValue());ListCanalEntry.RowData

rowDatasList

rowChange.getRowDatasList();for

(CanalEntry.RowData

rowData.getBeforeColumnsList();StringBuffer

sql

entry.getHeader().getSchemaName()

entry.getHeader().getTableName()

where

{//暂时只支持单一主键sql.append(column.getName()

column.getValue());break;}}SQL_QUEUE.add(sql.toString());}}

catch

(InvalidProtocolBufferException

保存插入语句**

RowChange.parseFrom(entry.getStoreValue());ListCanalEntry.RowData

rowDatasList

rowChange.getRowDatasList();for

(CanalEntry.RowData

rowData.getAfterColumnsList();StringBuffer

sql

entry.getHeader().getSchemaName()

entry.getHeader().getTableName()

();for

{sql.append(columnList.get(i).getName());if

columnList.size()

{sql.append(,);}}sql.append());SQL_QUEUE.add(sql.toString());}}

catch

(InvalidProtocolBufferException

}package

com.canal.canalclient.config.CanalClient;

import

org.springframework.beans.factory.annotation.Autowired;

import

org.springframework.beans.factory.annotation.Qualifier;

import

org.springframework.boot.ApplicationArguments;

import

org.springframework.boot.ApplicationRunner;

import

org.springframework.jdbc.core.JdbcTemplate;

import

org.springframework.stereotype.Component;Component

Slf4j

{AutowiredQualifier(test_master_energy)

static

Exception{log.info(开始监听同步数据库);CanalClient.jdbcTemplate

jdbcTemplate;CanalClient.startCanal();}

}?xml

xmlnshttp://maven.apache.org/POM/4.0.0

xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0

https://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersionparentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion3.0.3/versionrelativePath/

!--

--/parentgroupIdcom.canal/groupIdartifactIdCanalClient/artifactIdversion0.0.1-SNAPSHOT/versionnameCanalClient/namedescriptionCanalClient/descriptionpropertiesjava.version19/java.version/propertiesdependenciesdependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactIdscopetest/scope/dependencydependencygroupIdcom.alibaba.otter/groupIdartifactIdcanal.client/artifactIdversion1.1.4/version/dependencydependencygroupIdcom.alibaba/groupIdartifactIddruid-spring-boot-starter/artifactIdversion1.2.9/version/dependencydependencygroupIdmysql/groupIdartifactIdmysql-connector-java/artifactIdversion8.0.32/version/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactId/dependency/dependenciesbuildplugins!--

--plugingroupIdorg.apache.maven.plugins/groupIdartifactIdmaven-surefire-plugin/artifactIdversion2.12.4/versionconfigurationskipTeststrue/skipTests/configuration/pluginplugingroupIdorg.springframework.boot/groupIdartifactIdspring-boot-maven-plugin/artifactId/plugin/pluginsresourcesresourcedirectorysrc/main/resources/directory/resourceresourcedirectorysrc/main/java/directoryincludesinclude**/*.xml/include/includes/resource/resources/build/project