第SpringBoot整合Canal與RabbitMQ監(jiān)聽數(shù)據(jù)變更記錄目錄需求步驟環(huán)境搭建perties修改canal配置文件整合SpringBootCanal實(shí)現(xiàn)客戶端Canal整合RabbitMQSpringBoot整合RabbitMQ

需求

我想要在SpringBoot中采用一種與業(yè)務(wù)代碼解耦合的方式，來實(shí)現(xiàn)數(shù)據(jù)的變更記錄，記錄的內(nèi)容是新數(shù)據(jù)，如果是更新操作還得有舊數(shù)據(jù)內(nèi)容。

經(jīng)過調(diào)研發(fā)現(xiàn)，使用Canal來監(jiān)聽MySQL的binlog變化可以實(shí)現(xiàn)這個(gè)需求，可是在監(jiān)聽到變化后需要馬上保存變更記錄，除非再做一些邏輯處理，于是我又結(jié)合了RabbitMQ來處理保存變更記錄的操作。

步驟

啟動(dòng)MySQL環(huán)境，并開啟binlog啟動(dòng)Canal環(huán)境，為其創(chuàng)建一個(gè)MySQL賬號(hào)，然后以Slave的形式連接MySQLCanal服務(wù)模式設(shè)為TCP，用Java編寫客戶端代碼，監(jiān)聽MySQL的binlog修改Canal服務(wù)模式設(shè)為RabbitMQ，啟動(dòng)RabbitMQ環(huán)境，配置Canal和RabbitMQ的連接，用消息隊(duì)列去接收binlog修改事件

環(huán)境搭建

環(huán)境搭建基于docker-compose：

version:"3"

services:

mysql:

network_mode:mynetwork

container_name:mymysql

ports:

-3306:3306

restart:always

volumes:

-/etc/localtime:/etc/localtime

-/home/mycontainers/mymysql/data:/data

-/home/mycontainers/mymysql/mysql:/var/lib/mysql

-/home/mycontainers/mymysql/conf:/etc/mysql

environment:

-MYSQL_ROOT_PASSWORD=root

command:

--character-set-server=utf8mb4

--collation-server=utf8mb4_unicode_ci

--log-bin=/var/lib/mysql/mysql-bin

--server-id=1

--binlog-format=ROW

--expire_logs_days=7

--max_binlog_size=500M

image:mysql:5.7.20

rabbitmq:

container_name:myrabbit

ports:

-15672:15672

-5672:5672

restart:always

volumes:

-/etc/localtime:/etc/localtime

-/home/mycontainers/myrabbit/rabbitmq:/var/lib/rabbitmq

network_mode:mynetwork

environment:

-RABBITMQ_DEFAULT_USER=admin

-RABBITMQ_DEFAULT_PASS=123456

image:rabbitmq:3.8-management

canal-server:

container_name:canal-server

restart:always

ports:

-11110:11110

-11111:11111

-11112:11112

volumes:

-/home/mycontainers/canal-server/conf/perties:/home/admin/canal-server/conf/perties

-/home/mycontainers/canal-server/conf/perties:/home/admin/canal-server/conf/example/perties

-/home/mycontainers/canal-server/logs:/home/admin/canal-server/logs

network_mode:mynetwork

depends_on:

-mysql

-rabbitmq

#-canal-admin

image:canal/canal-server:v1.1.5

我們需要修改下Canal環(huán)境的配置文件：perties和perties，映射Canal中的以下兩個(gè)路徑：

/home/admin/canal-server/conf/perties：配置文件中，canal.destinations意思是server上部署的instance列表，/home/admin/canal-server/conf/example/perties：這里的/example是指instance即實(shí)例名，要和上面perties內(nèi)instance配置對(duì)應(yīng)，canal會(huì)為實(shí)例創(chuàng)建對(duì)應(yīng)的文件夾，一個(gè)Client對(duì)應(yīng)一個(gè)實(shí)例

以下是我們需要準(zhǔn)備的兩個(gè)配置文件具體內(nèi)容：

perties

#################################################

#########commonargument#############

#################################################

#tcpbindip

canal.ip=

#registeriptozookeeper

canal.register.ip=

canal.port=11111

canal.metrics.pull.port=11112

#canalinstanceuser/passwd

#canal.user=canal

#canal.passwd=E3619321C1A937C46A0D8BD1DAC39F93B27D4458

#canaladminconfig

#canal.admin.manager=canal-admin:8089

#canal.admin.port=11110

#canal.admin.user=admin

#canal.admin.passwd=6BB4837EB74329105EE4568DDA7DC67ED2CA2AD9

#adminautoregister自動(dòng)注冊(cè)

#canal.admin.register.auto=true

#集群名，單機(jī)則不寫

#canal.admin.register.cluster=

#CanalServer名字

#=canal-admin

canal.zkServers=

#flushdatatozk

canal.zookeeper.flush.period=1000

canal.withoutNetty=false

#tcp,kafka,rocketMQ,rabbitMQ,pulsarMQ

canal.serverMode=tcp

#flushmetacursor/parsepositiontofile

canal.file.data.dir=${canal.conf.dir}

canal.file.flush.period=1000

##memorystoreRingBuffersize,shouldbeMath.pow(2,n)

canal.instance.memory.buffer.size=16384

##memorystoreRingBufferusedmemoryunitsize,default1kb

canal.instance.memory.buffer.memunit=1024

##meorystoregetsmodeusedMEMSIZEorITEMSIZE

canal.instance.memory.batch.mode=MEMSIZE

canal.instance.memory.rawEntry=true

##detecingconfig

canal.instance.detecting.enable=false

#canal.instance.detecting.sql=insertintoretl.xdualvalues(1,now())onduplicatekeyupdatex=now()

canal.instance.detecting.sql=select1

erval.time=3

canal.instance.detecting.retry.threshold=3

canal.instance.detecting.heartbeatHaEnable=false

#supportmaximumtransactionsize,morethanthesizeofthetransactionwillbecutintomultipletransactionsdelivery

canal.instance.transaction.size=1024

#mysqlfallbackconnectedtonewmastershouldfallbacktimes

canal.instance.fallbackIntervalInSeconds=60

#networkconfig

work.receiveBufferSize=16384

work.sendBufferSize=16384

work.soTimeout=30

#binlogfilterconfig

canal.instance.filter.druid.ddl=true

canal.instance.filter.query.dcl=false

canal.instance.filter.query.dml=false

canal.instance.filter.query.ddl=false

canal.instance.filter.table.error=false

canal.instance.filter.rows=false

canal.instance.filter.transaction.entry=false

canal.instance.filter.dml.insert=false

canal.instance.filter.dml.update=false

canal.instance.filter.dml.delete=false

#binlogformat/imagecheck

canal.instance.binlog.format=ROW,STATEMENT,MIXED

canal.instance.binlog.image=FULL,MINIMAL,NOBLOB

#binlogddlisolation

canal.instance.get.ddl.isolation=false

#parallelparserconfig

canal.instance.parser.parallel=true

##concurrentthreadnumber,default60%availableprocessors,suggestnottoexceedRuntime.getRuntime().availableProcessors()

canal.instance.parser.parallelThreadSize=16

##disruptorringbuffersize,mustbepowerof2

canal.instance.parser.parallelBufferSize=256

#tablemetatsdbinfo

canal.instance.tsdb.enable=true

canal.instance.tsdb.dir=${canal.file.data.dir:../conf}/${canal.instance.destination:}

canal.instance.tsdb.url=jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;

canal.instance.tsdb.dbUsername=canal

canal.instance.tsdb.dbPassword=canal

#dumpsnapshotinterval,default24hour

erval=24

#purgesnapshotexpire,default360hour(15days)

canal.instance.tsdb.snapshot.expire=360

#################################################

#########destinations#############

#################################################

canal.destinations=canal-exchange

#confrootdir

canal.conf.dir=../conf

#autoscaninstancediradd/removeandstart/stopinstance

canal.auto.scan=true

erval=5

#setthisvalueto'true'meansthatwhenbinlogposnotfound,skiptolatest.

#WARN:plskeep'false'inproductionenv,orifyouknowwhatyouwant.

canal.auto.reset.latest.pos.mode=false

canal.instance.tsdb.spring.xml=classpath:spring/tsdb/h2-tsdb.xml

#canal.instance.tsdb.spring.xml=classpath:spring/tsdb/mysql-tsdb.xml

canal.instance.global.mode=spring

canal.instance.global.lazy=false

canal.instance.global.manager.address=${canal.admin.manager}

#canal.instance.global.spring.xml=classpath:spring/memory-instance.xml

canal.instance.global.spring.xml=classpath:spring/file-instance.xml

#canal.instance.global.spring.xml=classpath:spring/default-instance.xml

##################################################

#########

MQProperties

#############

##################################################

#aliyunak/sk,supportrds/mq

canal.aliyun.accessKey=

canal.aliyun.secretKey=

canal.aliyun.uid=

canal.mq.flatMessage=true

canal.mq.canalBatchSize=50

canal.mq.canalGetTimeout=100

#Setthisvalueto"cloud",ifyouwantopenmessagetracefeatureinaliyun.

canal.mq.accessChannel=local

canal.mq.database.hash=true

canal.mq.send.thread.size=30

canal.mq.build.thread.size=8

##################################################

#########

Kafka

#############

##################################################

kafka.bootstrap.servers=:9092

kafka.acks=all

pression.type=none

kafka.batch.size=16384

kafka.linger.ms=1

kafka.max.request.size=1048576

kafka.buffer.memory=33554432

kafka.max.in.flight.requests.per.connection=1

kafka.retries=0

kafka.kerberos.enable=false

kafka.kerberos.krb5.file="../conf/kerberos/krb5.conf"

kafka.kerberos.jaas.file="../conf/kerberos/jaas.conf"

##################################################

#########

RocketMQ

#############

##################################################

ducer.group=test

rocketmq.enable.message.trace=false

rocketmq.customized.trace.topic=

space=

srv.addr=:9876

rocketmq.retry.times.when.send.failed=0

rocketmq.vip.channel.enabled=false

rocketmq.tag=

##################################################

#########

RabbitMQ

#############

##################################################

rabbitmq.host=myrabbit

rabbitmq.virtual.host=/

rabbitmq.exchange=canal-exchange

rabbitmq.username=admin

rabbitmq.password=RabbitMQ密碼

rabbitmq.deliveryMode=

##################################################

#########

Pulsar

#############

##################################################

pulsarmq.serverUrl=

pulsarmq.roleToken=

pulsarmq.topicTenantPrefix=

此時(shí)canal.serverMode=tcp，即TCP直連，我們先開啟這個(gè)服務(wù)，然后手寫Java客戶端代碼去連接它，等下再改為RabbitMQ。

通過注釋可以看到，canal支持的服務(wù)模式有：tcp,kafka,rocketMQ,rabbitMQ,pulsarMQ，即主流的消息隊(duì)列都支持。

perties

#################################################

##mysqlserverId,v1.0.26+willautoGen

#canal.instance.mysql.slaveId=123

#enablegtidusetrue/false

canal.instance.gtidon=false

#positioninfo

canal.instance.master.address=mymysql:3306

=

canal.instance.master.position=

canal.instance.master.timestamp=

canal.instance.master.gtid=

#rdsossbinlog

canal.instance.rds.accesskey=

canal.instance.rds.secretkey=

canal.instance.rds.instanceId=

#tablemetatsdbinfo

canal.instance.tsdb.enable=true

#canal.instance.tsdb.url=jdbc:mysql://:3306/canal_tsdb

#canal.instance.tsdb.dbUsername=canal

#canal.instance.tsdb.dbPassword=canal

#canal.instance.standby.address=

#=

#canal.instance.standby.position=

#canal.instance.standby.timestamp=

#canal.instance.standby.gtid=

#username/password

canal.instance.dbUsername=canal

canal.instance.dbPassword=canal

canal.instance.connectionCharset=UTF-8

#enabledruidDecryptdatabasepassword

canal.instance.enableDruid=false

#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==

#tableregex

canal.instance.filter.regex=.*\..*

#tableblackregex

canal.instance.filter.black.regex=mysql\.slave_.*

#tablefieldfilter(format:schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)

#canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch

#tablefieldblackfilter(format:schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)

#canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch

#mqconfig

canal.mq.topic=canal-routing-key

#dynamictopicroutebyschemaortableregex

#canal.mq.dynamicTopic=mytest1.user,topic2:mytest2\..*,.*\..*

canal.mq.partition=0

#hashpartitionconfig

#canal.mq.enableDynamicQueuePartition=false

#canal.mq.partitionsNum=3

#canal.mq.dynamicTopicPartitionNum=test.*:4,mycanal:6

#canal.mq.partitionHash=test.table:id^name,.*\..*

#################################################

把這兩個(gè)配置文件映射好，再次提醒，注意實(shí)例的路徑名，默認(rèn)是：/example/perties

修改canal配置文件

我們需要修改這個(gè)實(shí)例配置文件，去連接MySQL，確保以下的配置正確：

canal.instance.master.address=mymysql:3306

canal.instance.dbUsername=canal

canal.instance.dbPassword=canal

mymysql是同為docker容器的MySQL環(huán)境，端口3306是指內(nèi)部端口。

這里多說明一下，docker端口配置時(shí)假設(shè)為：13306:3306，那么容器對(duì)外的端口就是13306，內(nèi)部是3306，在本示例中，MySQL和Canal都是容器環(huán)境，所以Canal連接MySQL需要滿足以下條件：

處于同一網(wǎng)段（docker-compose.yml中的mynetwork）訪問內(nèi)部端口（即3306，而非13306）

dbUsername和dbPassword為MySQL賬號(hào)密碼，為了開發(fā)方便可以使用root/root，但是我仍建議自行創(chuàng)建用戶并分配訪問權(quán)限：

#進(jìn)入docker中的mysql容器

dockerexec-itmymysqlbash

#進(jìn)入mysql指令模式

mysql-uroot-proot

#編寫MySQL語句并執(zhí)行

...

--選擇mysql

usemysql;

--創(chuàng)建canal用戶，賬密：canal/canal

createuser'canal'@'%'identifiedby'canal';

--分配權(quán)限，以及允許所有主機(jī)登錄該用戶

grantSELECT,INSERT,UPDATE,DELETE,REPLICATIONSLAVE,REPLICATIONCLIENTon*.*to'canal'@'%';

--刷新一下使其生效

flushprivileges;

--附帶一個(gè)刪除用戶指令

dropuser'canal'@'%';

用navicat或者shell去登錄canal這個(gè)用戶，可以訪問即創(chuàng)建成功

整合SpringBootCanal實(shí)現(xiàn)客戶端

Maven依賴：

canal.version1.1.5/canal.version

!--canal--

dependency

groupIdcom.alibaba.otter/groupId

artifactIdcanal.client/artifactId

version${canal.version}/version

/dependency

dependency

groupIdcom.alibaba.otter/groupId

artifactIdtocol/artifactId

version${canal.version}/version

/dependency

新增組件并啟動(dòng)：

importcom.alibaba.otter.canal.client.CanalConnector;

importcom.alibaba.otter.canal.client.CanalConnectors;

importtocol.CanalEntry;

importtocol.Message;

importorg.springframework.boot.CommandLineRunner;

importorg.springframework.stereotype.Component;

import.InetSocketAddress;

importjava.util.List;

@Component

publicclassCanalClient{

privatefinalstaticintBATCH_SIZE=1000;

publicvoidrun(){

//創(chuàng)建鏈接

CanalConnectorconnector=CanalConnectors.newSingleConnector(newInetSocketAddress("localhost",11111),"canal-exchange","canal","canal");

try{

//打開連接

connector.connect();

//訂閱數(shù)據(jù)庫表,全部表

connector.subscribe(".*\..*");

//回滾到未進(jìn)行ack的地方，下次fetch的時(shí)候，可以從最后一個(gè)沒有ack的地方開始拿

connector.rollback();

while(true){

//獲取指定數(shù)量的數(shù)據(jù)

Messagemessage=connector.getWithoutAck(BATCH_SIZE);

//獲取批量ID

longbatchId=message.getId();

//獲取批量的數(shù)量

intsize=message.getEntries().size();

//如果沒有數(shù)據(jù)

if(batchId==-1||size==0){

try{

//線程休眠2秒

Thread.sleep(2000);

}catch(InterruptedExceptione){

e.printStackTrace();

}

}else{

//如果有數(shù)據(jù),處理數(shù)據(jù)

printEntry(message.getEntries());

}

//進(jìn)行batchid的確認(rèn)。確認(rèn)之后，小于等于此batchId的Message都會(huì)被確認(rèn)。

connector.ack(batchId);

}

}catch(Exceptione){

e.printStackTrace();

}finally{

connector.disconnect();

}

/**

*打印canalserver解析binlog獲得的實(shí)體類信息

*/

privatestaticvoidprintEntry(ListCanalEntry.Entryentrys){

for(CanalEntry.Entryentry:entrys){

if(entry.getEntryType()==CanalEntry.EntryType.TRANSACTIONBEGIN||entry.getEntryType()==CanalEntry.EntryType.TRANSACTIONEND){

//開啟/關(guān)閉事務(wù)的實(shí)體類型，跳過

continue;

}

//RowChange對(duì)象，包含了一行數(shù)據(jù)變化的所有特征

//比如isDdl是否是ddl變更操作sql具體的ddlsqlbeforeColumnsafterColumns變更前后的數(shù)據(jù)字段等等

CanalEntry.RowChangerowChage;

try{

rowChage=CanalEntry.RowChange.parseFrom(entry.getStoreValue());

}catch(Exceptione){

thrownewRuntimeException("ERROR##parseroferomanga-eventhasanerror,data:"+entry.toString(),e);

}

//獲取操作類型：insert/update/delete類型

CanalEntry.EventTypeeventType=rowChage.getEventType();

//打印Header信息

System.out.println(String.format("================》;binlog[%s:%s],name[%s,%s],eventType:%s",

entry.getHeader().getLogfileName(),entry.getHeader().getLogfileOffset(),

entry.getHeader().getSchemaName(),entry.getHeader().getTableName(),

eventType));

//判斷是否是DDL語句

if(rowChage.getIsDdl()){

System.out.println("================》;isDdl:true,sql:"+rowChage.getSql());

}

//獲取RowChange對(duì)象里的每一行數(shù)據(jù)，打印出來

for(CanalEntry.RowDatarowData:rowChage.getRowDatasList()){

//如果是刪除語句

if(eventType==CanalEntry.EventType.DELETE){

printColumn(rowData.getBeforeColumnsList());

//如果是新增語句

}elseif(eventType==CanalEntry.EventType.INSERT){

printColumn(rowData.getAfterColumnsList());

//如果是更新的語句

}else{

//變更前的數(shù)據(jù)

System.out.println("-------before");

printColumn(rowData.getBeforeColumnsList());

//變更后的數(shù)據(jù)

System.out.println("-------after");

printColumn(rowData.getAfterColumnsList());

}

}

}

privatestaticvoidprintColumn(ListCanalEntry.Columncolumns){

for(CanalEntry.Columncolumn:columns){

System.out.println(column.getName()+":"+column.getValue()+"

update="+column.getUpdated());

}

}

啟動(dòng)類Application：

@SpringBootApplication

publicclassBaseApplicationimplementsCommandLineRunner{

@Autowired

privateCanalClientcanalClient;

@Override

publicvoidrun(String...args)throwsException{

canalClient.run();

}

啟動(dòng)程序，此時(shí)新增或修改數(shù)據(jù)庫中的數(shù)據(jù)，我們就能從客戶端中監(jiān)聽到

不過我建議監(jiān)聽的信息放到消息隊(duì)列中，在空閑的時(shí)候去處理，所以直接配置Canal整合RabbitMQ更好。

Canal整合RabbitMQ

修改perties中的serverMode：

canal.serverMode=rabbitMQ

修改perties中的topic：

canal.mq.topic=canal-routing-key

然后找到關(guān)于RabbitMQ的配置：

##################################################

#########

RabbitMQ

#############

##################################################

#連接rabbit，寫IP，因?yàn)橥瑐€(gè)網(wǎng)絡(luò)下，所以可以寫容器名

rabbitmq.host=myrabbit

rabbitmq.virtual.host=/

#交換器名稱，等等我們要去手動(dòng)創(chuàng)建

rabbitmq.exchange=canal-exchange

rabbitmq.username=admin

rabbitmq.password=123456

#暫不支持指定端口，使用的是默認(rèn)的5762，好在在本示例中適用

重新啟動(dòng)容器，進(jìn)入RabbitMQ管理頁面創(chuàng)建exchange交換器和隊(duì)列queue：

新建exchange，命名為：canal-exchange新建queue，命名為：canal-queue綁定exchange和queue，routing-key設(shè)置為：canal-routing-key，這里對(duì)應(yīng)上面perties的canal.mq.topic

順帶一提，上面這段可以忽略，因?yàn)樵赟pringBoot的RabbitMQ配置中，會(huì)自動(dòng)創(chuàng)建交換器exchange和隊(duì)列queue，不過手動(dòng)創(chuàng)建的話，可以在忽略SpringBoot的基礎(chǔ)上，直接在RabbitMQ的管理頁面上看到修改記錄的消息。

SpringBoot整合RabbitMQ

依賴：

amqp.version2.3.4.RELEASE/amqp.version

!--消息隊(duì)列--

dependency

groupIdorg.springframework.boot/groupId

artifactIdspring-boot-starter-amqp/artifactId

version${amqp.version}/version

/dependency

application.yml：

spring:

rabbitmq:

#

host:myserverhost

host:08

port:5672

username:admin

password:RabbitMQ密碼

#消息確認(rèn)配置項(xiàng)

#確認(rèn)消息已發(fā)送到交換機(jī)(Exchange)

publisher-confirm-type:correlated

#確認(rèn)消息已發(fā)送到隊(duì)列(Queue)

publisher-returns:true

RabbitMQ配置類：

@Configuration

publicclassRabbitConfig{

@Bean

publicRabbitTemplaterabbitTemplate(ConnectionFactoryconnectionFactory){

RabbitTemplatetemplate=newRabbitTemplate();

template.setConnectionFactory(connectionFactory);

template.setMessageConverter(newJackson2JsonMessageConverter());

returntemplate;

/**

*template.setMessageConverter(newJackson2JsonMessageConverter());

*這段和上面這行代碼解決RabbitListener循環(huán)報(bào)錯(cuò)的問題

*/

@Bean

publicSimpleRabbitListenerContainerFactoryrabbitListenerContainerFactory(ConnectionFactoryconnectionFactory){

SimpleRabbitListenerContainerFactoryfactory=newSimpleRabbitListenerContainerFactory();

factory.setConnectionFactory(connectionFactory);

factory.setMessageConverter(newJackson2JsonMessageConverter());

returnfactory;

}

Canal消息生產(chǎn)者：

publicstaticfinalStringCanalQueue="canal-queue";

publicstaticfinalStringCanalExchange="canal-exchange";

publicstaticfinalStringCanalRouting="canal-routing-key";

/**

*Canal消息提供者，canal-server生產(chǎn)的消息通過RabbitMQ消息隊(duì)列發(fā)送

@Configuration

publicclassCanalProvider{

/**

*