当前位置：首页 > news >正文

数据采集工具之Canal

news 2025/9/17 2:15:52

本文主要介绍canal采集mysql数据的tcp、datahub(kafka)模式如何实现

1、下载canal

https://aliyun-datahub.oss-cn-hangzhou.aliyuncs.com/tools/canal.deployer-1.1.5-SNAPSHOT.tar.gz

canal的原理类似于mysql的主从复制，canal模拟的是从节点拉取主节点的binlog数据

2、TCP模式的实现

a、canal.properties

打开看看即可，不需要调整

#################################################
#########               common argument         #############
#################################################
# tcp bind ip
canal.ip =
# register ip to zookeeper
canal.register.ip =
canal.port = 11111
canal.metrics.pull.port = 11112
# canal instance user/passwd
# canal.user = canal
# canal.passwd = E3619321C1A937C46A0D8BD1DAC39F93B27D4458# canal admin config
#canal.admin.manager = 127.0.0.1:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441canal.zkServers =
# flush data to zk
canal.zookeeper.flush.period = 1000
canal.withoutNetty = false
# tcp, kafka, rocketMQ, rabbitMQ
canal.serverMode = tcp
# flush meta cursor/parse position to file
canal.file.data.dir = ${canal.conf.dir}
canal.file.flush.period = 1000
## memory store RingBuffer size, should be Math.pow(2,n)
canal.instance.memory.buffer.size = 16384
## memory store RingBuffer used memory unit size , default 1kb
canal.instance.memory.buffer.memunit = 1024 
## meory store gets mode used MEMSIZE or ITEMSIZE
canal.instance.memory.batch.mode = MEMSIZE
canal.instance.memory.rawEntry = true## detecing config
canal.instance.detecting.enable = false
#canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now()
canal.instance.detecting.sql = select 1
canal.instance.detecting.interval.time = 3
canal.instance.detecting.retry.threshold = 3
canal.instance.detecting.heartbeatHaEnable = false# support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery
canal.instance.transaction.size =  1024
# mysql fallback connected to new master should fallback times
canal.instance.fallbackIntervalInSeconds = 60# network config
canal.instance.network.receiveBufferSize = 16384
canal.instance.network.sendBufferSize = 16384
canal.instance.network.soTimeout = 30# binlog filter config
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# binlog format/image check
canal.instance.binlog.format = ROW,STATEMENT,MIXED 
canal.instance.binlog.image = FULL,MINIMAL,NOBLOB# binlog ddl isolation
canal.instance.get.ddl.isolation = false# parallel parser config
canal.instance.parser.parallel = true
## concurrent thread number, default 60% available processors, suggest not to exceed Runtime.getRuntime().availableProcessors()
#canal.instance.parser.parallelThreadSize = 16
## disruptor ringbuffer size, must be power of 2
canal.instance.parser.parallelBufferSize = 256# table meta tsdb info
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
# dump snapshot interval, default 24 hour
canal.instance.tsdb.snapshot.interval = 24
# purge snapshot expire , default 360 hour(15 days)
canal.instance.tsdb.snapshot.expire = 360#################################################
#########               destinations            #############
#################################################
canal.destinations = example  ##这里可以设置多个逗号分开
# conf root dir
canal.conf.dir = ../conf
# auto scan instance dir add/remove and start/stop instance
canal.auto.scan = true
canal.auto.scan.interval = 5canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml
#canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xmlcanal.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##################################################
#########             MQ Properties      #############
##################################################
canal.mq.flat.message = true
canal.mq.database.hash = true
canal.mq.parallel.thread.size = 8
canal.mq.canal.batch.size = 50
canal.mq.canal.fetch.timeout = 100
# Set this value to "cloud", if you want open message trace feature in aliyun.
canal.mq.access.channel = local# aliyun ak/sk , support rds/mq
canal.aliyun.accessKey =
canal.aliyun.secretKey =
canal.aliyun.uid=##################################################
#########                    Kafka                   #############
##################################################
kafka.bootstrap.servers = 127.0.0.1:9092
kafka.acks = all
kafka.compression.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 = 0canal.mq.kafka.kerberos.enable = false
canal.mq.kafka.kerberos.krb5.file = "../conf/kerberos/krb5.conf"
canal.mq.kafka.kerberos.jaas.file = "../conf/kerberos/jaas.conf"##################################################
#########                   RocketMQ         #############
##################################################
rocketmq.producer.group = test
rocketmq.enable.message.trace = false
rocketmq.customized.trace.topic =
rocketmq.namespace =
rocketmq.namesrv.addr = 127.0.0.1:9876
rocketmq.retry.times.when.send.failed = 0
rocketmq.vip.channel.enabled = false##################################################
#########                   RabbitMQ         #############
##################################################
rabbitmq.host =
rabbitmq.virtual.host =
rabbitmq.exchange =
rabbitmq.username =

b、example/instance.properties

canal.instance.master.address=192.168.140.1:3306 ###修改为自己的mysql信息

#################################################
## mysql serverId , v1.0.26+ will autoGen
# canal.instance.mysql.slaveId=0# enable gtid use true/false
canal.instance.gtidon=false# position info
canal.instance.master.address=192.168.140.1:3306  ###修改为自己的mysql信息
canal.instance.master.journal.name=
canal.instance.master.position=
canal.instance.master.timestamp=
canal.instance.master.gtid=# rds oss binlog
canal.instance.rds.accesskey=
canal.instance.rds.secretkey=
canal.instance.rds.instanceId=# table meta tsdb info
canal.instance.tsdb.enable=true
#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
#canal.instance.tsdb.dbUsername=canal
#canal.instance.tsdb.dbPassword=canal#canal.instance.standby.address =
#canal.instance.standby.journal.name =
#canal.instance.standby.position =
#canal.instance.standby.timestamp =
#canal.instance.standby.gtid=# username/password
canal.instance.dbUsername=flink
canal.instance.dbPassword=flink
canal.instance.connectionCharset = UTF-8
# enable druid Decrypt database password
canal.instance.enableDruid=false
#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==# table regex
canal.instance.filter.regex=.*\\..*
# table black regex
canal.instance.filter.black.regex=
# table field filter(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
# table field black filter(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# mq config
canal.mq.topic=example
# dynamic topic route by schema or table regex
#canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..*
canal.mq.partition=0
# hash partition config
#canal.mq.partitionsNum=3
#canal.mq.partitionHash=test.table:id^name,.*\\..*
#################################################

启动：bin/startup.sh

jps:

到此，canal服务端配置完成

c、canal客户端开发

依赖

<dependency><groupId>com.alibaba.otter</groupId><artifactId>canal.client</artifactId><version>1.1.2</version></dependency>

开发代码：

package com.tbea;import com.alibaba.fastjson.JSONObject;
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.Message;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;import java.net.InetSocketAddress;
import java.util.List;public class CanalClient {public static void main(String[] args) throws InterruptedException, InvalidProtocolBufferException {//获取链接CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.140.129", 11111), "example", "", "");//尝试获取新数据while (true){//todo 连接canalConnector.connect();//todo 订阅数据库canalConnector.subscribe("flinkcdc.*");//todo 批量拉取数据Message message = canalConnector.get(100);//todo 获取entryList<CanalEntry.Entry> entries = message.getEntries();//todo 遍历 判断集合状态if (entries.size()<=0){System.out.println("当次未抓取到数据，休息一会~~~~~~~");Thread.sleep(1000);}else {//遍历解析for (CanalEntry.Entry entry:entries){//1.获取表名String tableName = entry.getHeader().getTableName();//2.获取类型CanalEntry.EntryType entryType = entry.getEntryType();//3.获取序列化数据ByteString storeValue = entry.getStoreValue();//4.判断entry是否rowdata类型if (CanalEntry.EntryType.ROWDATA.equals(entryType)){//5.反序列化数据CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);//6.获取当前事件操作类型CanalEntry.EventType eventType = rowChange.getEventType();//7.获取数据集List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();//8.遍历rowdatalist，并打印数据集for (CanalEntry.RowData rowData:rowDatasList){JSONObject beforeData = new JSONObject();JSONObject afterData = new JSONObject();List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();for (CanalEntry.Column column:beforeColumnsList){beforeData.put(column.getName(),column.getValue());}for (CanalEntry.Column column:afterColumnsList){afterData.put(column.getName(),column.getValue());}System.out.println("Table:"+tableName+",EventType:"+eventType+",Before:"+beforeData+",After:"+afterData);}}else {System.out.println("数据类型不是所需要的");}}}}}
}

执行：

到此，我们可以实时获取到mysql数据库的各种操作日志，接下来需要将数据写到哪里可以按需实现。

3、kafka模式的实现

a.canal.properties

修改：canal.serverMode = kafka

kafka信息：kafka.bootstrap.servers = 192.168.140.128:9092

#################################################
#########               common argument         #############
#################################################
# tcp bind ip
canal.ip =
# register ip to zookeeper
canal.register.ip =
canal.port = 11111
canal.metrics.pull.port = 11112
# canal instance user/passwd
# canal.user = canal
# canal.passwd = E3619321C1A937C46A0D8BD1DAC39F93B27D4458# canal admin config
#canal.admin.manager = 127.0.0.1:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441canal.zkServers =
# flush data to zk
canal.zookeeper.flush.period = 1000
canal.withoutNetty = false
# tcp, kafka, rocketMQ, rabbitMQ
canal.serverMode = kafka
# flush meta cursor/parse position to file
canal.file.data.dir = ${canal.conf.dir}
canal.file.flush.period = 1000
## memory store RingBuffer size, should be Math.pow(2,n)
canal.instance.memory.buffer.size = 16384
## memory store RingBuffer used memory unit size , default 1kb
canal.instance.memory.buffer.memunit = 1024 
## meory store gets mode used MEMSIZE or ITEMSIZE
canal.instance.memory.batch.mode = MEMSIZE
canal.instance.memory.rawEntry = true## detecing config
canal.instance.detecting.enable = false
#canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now()
canal.instance.detecting.sql = select 1
canal.instance.detecting.interval.time = 3
canal.instance.detecting.retry.threshold = 3
canal.instance.detecting.heartbeatHaEnable = false# support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery
canal.instance.transaction.size =  1024
# mysql fallback connected to new master should fallback times
canal.instance.fallbackIntervalInSeconds = 60# network config
canal.instance.network.receiveBufferSize = 16384
canal.instance.network.sendBufferSize = 16384
canal.instance.network.soTimeout = 30# binlog filter config
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# binlog format/image check
canal.instance.binlog.format = ROW,STATEMENT,MIXED 
canal.instance.binlog.image = FULL,MINIMAL,NOBLOB# binlog ddl isolation
canal.instance.get.ddl.isolation = false# parallel parser config
canal.instance.parser.parallel = true
## concurrent thread number, default 60% available processors, suggest not to exceed Runtime.getRuntime().availableProcessors()
#canal.instance.parser.parallelThreadSize = 16
## disruptor ringbuffer size, must be power of 2
canal.instance.parser.parallelBufferSize = 256# table meta tsdb info
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
# dump snapshot interval, default 24 hour
canal.instance.tsdb.snapshot.interval = 24
# purge snapshot expire , default 360 hour(15 days)
canal.instance.tsdb.snapshot.expire = 360#################################################
#########               destinations            #############
#################################################
canal.destinations = example
# conf root dir
canal.conf.dir = ../conf
# auto scan instance dir add/remove and start/stop instance
canal.auto.scan = true
canal.auto.scan.interval = 5canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml
#canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xmlcanal.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##################################################
#########             MQ Properties      #############
##################################################
canal.mq.flat.message = true
canal.mq.database.hash = true
canal.mq.parallel.thread.size = 8
canal.mq.canal.batch.size = 50
canal.mq.canal.fetch.timeout = 100
# Set this value to "cloud", if you want open message trace feature in aliyun.
canal.mq.access.channel = local# aliyun ak/sk , support rds/mq
canal.aliyun.accessKey =
canal.aliyun.secretKey =
canal.aliyun.uid=##################################################
#########                    Kafka                   #############
##################################################
kafka.bootstrap.servers = 192.168.140.128:9092
kafka.acks = all
kafka.compression.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 = 0canal.mq.kafka.kerberos.enable = false
canal.mq.kafka.kerberos.krb5.file = "../conf/kerberos/krb5.conf"
canal.mq.kafka.kerberos.jaas.file = "../conf/kerberos/jaas.conf"##################################################
#########                   RocketMQ         #############
##################################################
rocketmq.producer.group = test
rocketmq.enable.message.trace = false
rocketmq.customized.trace.topic =
rocketmq.namespace =
rocketmq.namesrv.addr = 127.0.0.1:9876
rocketmq.retry.times.when.send.failed = 0
rocketmq.vip.channel.enabled = false##################################################
#########                   RabbitMQ         #############
##################################################
rabbitmq.host =
rabbitmq.virtual.host =
rabbitmq.exchange =
rabbitmq.username =
rabbitmq.password =

b、example/instance.properties

修改：canal.instance.master.address=192.168.140.1:3306

canal.instance.dbUsername=flink

canal.instance.dbPassword=flink

canal.mq.topic=mysql_binlogs

#################################################
## mysql serverId , v1.0.26+ will autoGen
# canal.instance.mysql.slaveId=0# enable gtid use true/false
canal.instance.gtidon=false# position info
canal.instance.master.address=192.168.140.1:3306
canal.instance.master.journal.name=
canal.instance.master.position=
canal.instance.master.timestamp=
canal.instance.master.gtid=# rds oss binlog
canal.instance.rds.accesskey=
canal.instance.rds.secretkey=
canal.instance.rds.instanceId=# table meta tsdb info
canal.instance.tsdb.enable=true
#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
#canal.instance.tsdb.dbUsername=canal
#canal.instance.tsdb.dbPassword=canal#canal.instance.standby.address =
#canal.instance.standby.journal.name =
#canal.instance.standby.position =
#canal.instance.standby.timestamp =
#canal.instance.standby.gtid=# username/password
canal.instance.dbUsername=flink
canal.instance.dbPassword=flink
canal.instance.connectionCharset = UTF-8
# enable druid Decrypt database password
canal.instance.enableDruid=false
#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==# table regex
canal.instance.filter.regex=.*\\..*
# table black regex
canal.instance.filter.black.regex=
# table field filter(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
# table field black filter(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# mq config
canal.mq.topic=mysql_binlogs
# dynamic topic route by schema or table regex
#canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..*
canal.mq.partition=0
# hash partition config
#canal.mq.partitionsNum=3
#canal.mq.partitionHash=test.table:id^name,.*\\..*
#################################################

c、启动

startup.sh

d、查看数据生产情况

kafka-console-consumer.sh --bootstrap-server 192.168.140.128:9092 --topic mysql_binlogs --from-beginning

4、datahub兼容kafka的实现

配置如何实现啊

什么是Canal插件,如何使用Canal插件_数据总线 DataHub(DataHub)-阿里云帮助中心

看样子需要datahub兼容kafka的ip:port （私有云需要联系运维）

5、TCP模式+Datahub SDK实现

TCP配置及基本实现参考2

a、依赖

    <dependency><groupId>com.aliyun.datahub</groupId><artifactId>aliyun-sdk-datahub</artifactId><version>2.25.1</version></dependency>

b、代码实现

package com.tbea;import com.alibaba.fastjson.JSONObject;
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.Message;
import com.aliyun.datahub.client.DatahubClient;
import com.aliyun.datahub.client.DatahubClientBuilder;
import com.aliyun.datahub.client.auth.AliyunAccount;
import com.aliyun.datahub.client.common.DatahubConfig;
import com.aliyun.datahub.client.example.examples.Constant;
import com.aliyun.datahub.client.exception.*;
import com.aliyun.datahub.client.http.HttpConfig;
import com.aliyun.datahub.client.model.*;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;import static com.aliyun.datahub.client.example.examples.Constant.*;public class CanalClient {static DatahubClient datahubClient = DatahubClientBuilder.newBuilder().setDatahubConfig(//Protocol可不设置，不设置默认使用PROTOBUF传输协议new DatahubConfig("https://datahub.cn-beijing-tbdg-d01.dh.res.bigdata.tbea.com",new AliyunAccount("2Z8tAOpDPBm5LEkA", "Tlupsw2G0PdKGCRyPLucHjeESqoCla"))).setHttpConfig(new HttpConfig().setCompressType(CompressType.LZ4)).build();public static void main(String[] args) throws InterruptedException, InvalidProtocolBufferException {//获取链接CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.140.129", 11111), "example", "", "");//尝试获取新数据while (true){//todo 连接canalConnector.connect();//todo 订阅数据库canalConnector.subscribe("flinkcdc.*");//todo 批量拉取数据Message message = canalConnector.get(100);//todo 获取entryList<CanalEntry.Entry> entries = message.getEntries();//todo 遍历 判断集合状态if (entries.size()<=0){System.out.println("当次未抓取到数据，休息一会~~~~~~~");Thread.sleep(1000);}else {//遍历解析for (CanalEntry.Entry entry:entries){//1.获取表名String tableName = entry.getHeader().getTableName();String schemaName = entry.getHeader().getSchemaName();//2.获取类型CanalEntry.EntryType entryType = entry.getEntryType();//3.获取序列化数据ByteString storeValue = entry.getStoreValue();//4.判断entry是否rowdata类型if (CanalEntry.EntryType.ROWDATA.equals(entryType)){//5.反序列化数据CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);//6.获取当前事件操作类型CanalEntry.EventType eventType = rowChange.getEventType();//7.获取数据集List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();JSONObject beforeData = new JSONObject();JSONObject afterData = new JSONObject();//8.遍历rowdatalist，并打印数据集for (CanalEntry.RowData rowData:rowDatasList){List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();for (CanalEntry.Column column:beforeColumnsList){beforeData.put(column.getName(),column.getValue());}for (CanalEntry.Column column:afterColumnsList){afterData.put(column.getName(),column.getValue());}System.out.println("Table:"+tableName+",EventType:"+eventType+",Before:"+beforeData+",After:"+afterData);}Binlog binlog = new Binlog(eventType.toString(), tableName, schemaName, beforeData.toJSONString(), afterData.toJSONString());tupleExample("bigdata","tcp_canal",3,binlog);}else {System.out.println("数据类型不是所需要的");}}}}}// 写入Tuple型数据public static void tupleExample(String project,String topic,int retryTimes,Binlog binlog) {DatahubClient datahubClient = DatahubClientBuilder.newBuilder().setDatahubConfig(//Protocol可不设置，不设置默认使用PROTOBUF传输协议new DatahubConfig("https://datahub.cn-*****************.com",new AliyunAccount("************", "*************"))).setHttpConfig(new HttpConfig().setCompressType(CompressType.LZ4)).build();// 获取schemaRecordSchema recordSchema = datahubClient.getTopic(project,topic).getRecordSchema();// 生成十条数据List<RecordEntry> recordEntries = new ArrayList<>();RecordEntry recordEntry = new RecordEntry();// 对每条数据设置额外属性，例如ip 机器名等。可以不设置额外属性，不影响数据写入recordEntry.addAttribute("key1", "value1");TupleRecordData data = new TupleRecordData(recordSchema);data.setField("operate", binlog.getOperater());data.setField("tablename", binlog.getTableName());data.setField("databasename", binlog.getDatabaseName());data.setField("before", binlog.getBefore());data.setField("after", binlog.getAfter());recordEntry.setRecordData(data);recordEntries.add(recordEntry);try {PutRecordsResult result = datahubClient.putRecords(project, topic, recordEntries);int i = result.getFailedRecordCount();if (i > 0) {retry(datahubClient, result.getFailedRecords(), retryTimes, project, topic);}}  catch (DatahubClientException e) {System.out.println("requestId:" + e.getRequestId() + "\tmessage:" + e.getErrorMessage());}}//重试机制public static void retry(DatahubClient client, List<RecordEntry> records, int retryTimes, String project, String topic) {boolean suc = false;while (retryTimes != 0) {retryTimes = retryTimes - 1;PutRecordsResult recordsResult = client.putRecords(project, topic, records);if (recordsResult.getFailedRecordCount() > 0) {retry(client,recordsResult.getFailedRecords(),retryTimes,project,topic);}suc = true;break;}if (!suc) {System.out.println("retryFailure");}}
}

注意topic和代码的映射关系

数据采集工具之Canal

本文主要介绍canal采集mysql数据的tcp、datahub(kafka)模式如何实现 1、下载canal https://aliyun-datahub.oss-cn-hangzhou.aliyuncs.com/tools/canal.deployer-1.1.5-SNAPSHOT.tar.gz canal的原理类似于mysql的主从复制，canal模拟的是从节点拉取主节点的binlog数…...

编程日记 2024/8/9 23:36:38

【后端】消息中间件小册

1.RabbitMQ RabbitMQ 是一个流行的消息中间件系统，采用 AMQP（高级消息队列协议）来管理消息的传递。它的工作原理涉及多个组件和机制来确保消息的可靠性和完整性。以下是 RabbitMQ 的基本工作原理以及如何保证消息不丢失的机制： R…...

编程日记 2024/8/9 23:34:36

【进阶篇-Day14：JAVA中IO流之转换流、序列化流、打印流、Properties集合的介绍】

目录 1、转换流1.1 转换流分类：1.2 转换流的作用（1）按照指定的字符编码读写操作：（2）将字节流转换为字符流进行操作： 2、序列化流2.1 序列化的基本使用：2.2 序列化的操作流程&#xf…...

编程日记 2024/8/9 23:33:34

【Material-UI】Checkbox 组件中的 Label Placement 设置详解

文章目录一、Checkbox 组件简介1. 组件概述2. labelPlacement 属性二、labelPlacement 属性的使用方法三、各标签位置的效果与应用场景1. Top（顶部）2. Start（左侧）3. Bottom（底部）4. End（右侧…...

编程日记 2024/8/9 23:30:30

XJTUSE-离散数学-集合

基本概念集合的包含与相等，如子集幂集：以A的所有子集组成的集合称为A的幂集AB <> 集合的基本运算基本运算证明会考交运算并运算补运算差运算：A \ B 环和运算：环积运算： 集合的其他表示方法文图表示法 …...

编程日记 2024/8/9 23:28:28

安徽省消防设施操作员题库

1.()是做人的基本准则，也是社会道德和职业道德的一项基本规范。 A.诚实守信(正确答案) B.爱岗敬业 C.以人为本 D.钻研业务 2.()是指为了追求完美，坚持工匠精神，在工作中不放松对自己的要求。 A.爱岗敬业 B.精益求精(正确答案) C.勤奋刻苦 D.专…...

编程日记 2024/8/9 23:27:27

Singularity容器安装与使用

Singularity容器技术 docker的缺点: 资源限制问题:Slurm利用cgroups实现资源分配，Docker通过ocker daemon无法实现。权限问题:Docker daemon使用 root用户启动，HPC场录期望使用普通用户运行容器。 singuiarily主要是适合HPC中的普通用户，…...

编程日记 2024/8/9 23:26:25

Linux 文件、重定向、缓冲区

个人主页：仍有未知等待探索-CSDN博客专题分栏： Linux 目录一、文件 1、文件的理解（浅层） 1.文件是什么？ 2.文件操作的前提 3.文件的存储 4.一个进程可以打开多个文件吗？如果可以怎么管理的&#xf…...

编程日记 2024/8/9 23:22:21

WEB漏洞-SQL注入之MYSQL注入

跨库注入的原理：针对同一IP下的不同域名同一服务器下网站A对应数据库A 网站B对应数据库B 网站C对应数据库C 如果某网站的存在注入点，注入点的权限恰好是root权限，也就是最高权限，那么可以通过跨库注入获取其他网站的数据库…...

编程日记 2024/8/9 23:20:17

mysql 查询 from a, b 和 a left join b 有什么区别

在MySQL中，from a, b 和 a left join b 有显著的区别，主要体现在查询结果和使用场景上。基本语法与返回结果： from a, b：这种写法实际上是将两个表作为一个整体来处理，即假设这两个表是同一个表。因此，它…...

编程日记 2024/8/9 23:17:13

禁用ssh 22端口

在Linux系统中，要关闭SSH端口，可以通过修改SSH配置文件或防火墙规则来实现。方法一：修改SSH配置文件 1. 使用root用户登录Linux系统。 2. 打开SSH配置文件，可以使用任何文本编辑器，如vi或nano。在大多数Linux发行版上…...

编程日记 2024/8/9 23:16:12

C++基础编程的学习3

nullptr关键字在C11之前，空指针通常用NULL或0表示。然而，这些表示方法存在类型安全问题。C11引入了nullptr关键字，它提供了一个明确的、类型安全的空指针值。 Lambda表达式 Lambda表达式是C11引入的一种便捷的匿名函数定义方式。当Lambda…...

编程日记 2024/8/9 23:15:11

Java中的Optional类：解锁优雅编程的秘密

引言在Java开发的世界里，空指针异常（NullPointerException）一直是让无数程序员头疼的问题之一。它不仅打断了程序的正常执行流程，还可能隐藏在复杂的业务逻辑之中，难以定位。自Java 8起，一个新的类——Op…...

编程日记 2024/8/9 23:14:10

聆思CSK6大模型开发板语音控制风扇（上）

前言本文介绍如何实现大模型语音交互对风扇实现开关控制，示例使用的硬件是聆思CSK6大模型开发板，大模型接的是星火大模型。本文将先详细介绍大模型云端从获取交互语音到下发控制指令的全过程，包括整合上下文、配置提示词、语义意图分类、提取…...

编程日记 2024/8/9 23:13:09

代码随想录算法训练营第四十一天 | 121. 买卖股票的最佳时机、122. 买卖股票的最佳时机II、123. 买卖股票的最佳时机III

一、121. 买卖股票的最佳时机题目链接：121. 买卖股票的最佳时机 - 力扣（LeetCode） 文章讲解：代码随想录 (programmercarl.com)——121. 买卖股票的最佳时机视频讲解：动态规划之 LeetCode：121.买卖股票的最…...

编程日记 2024/8/9 23:12:05

延时队列与redis and rabbitmq

延时队列是什么延时队列（Delay Queue）是一种特殊的消息队列，它允许你在添加消息时设置一个延时时间，消息只有在延时时间到达后才能被消费。这种机制在分布式系统中非常有用，常用于处理需要在指定时间后执行的任务&am…...

编程日记 2024/8/9 23:10:03

数据结构--单链

#include "link.h" plink get_head() { plink pmalloc(sizeof(Link)); if(pNULL) { printf("申情节点失败\n"); return NULL; } p->len0; p->nextNULL; return p; } void head_insert(plink L,int a) {…...

编程日记 2024/8/9 23:07:57

春秋云镜CVE-2023-38836

打开靶场环境点击发现一个登陆框，弱口令试一下发现账号密码为admin,password 随便点击点击 Media发现这里可以上传文件上传木马试试 <?php eval($_POST["wjq"]); ?> 发现不能上传php文件 php内容修改他的格式抓包绕过一下 302就可以其实已经…...

编程日记 2024/8/9 23:06:55

Linux 进程概念

Linux 进程概念硬件理解冯诺依曼体系结构五大组成部件强调存储引子操作系统（Operator System）概念作用认识为什么要有操作系统？ 结构示意图理解操作系统system call库函数概念进程什么是进程概念误区认识描述进程 - PCBtask_struct - P…...

编程日记 2024/8/9 23:05:54

【秋招突围】2024届校招-米哈游笔试题-第二套

🍭 大家好这里是春秋招笔试突围，一起备战大厂笔试 💻 ACM金牌团队🏅️ | 多次AK大厂笔试｜编程一对一辅导 ✨ 本系列打算持续跟新春秋招笔试题 👏 感谢大家的订阅➕ 和喜欢💗 和手里的小花花🌸 ✨ 笔试合集传送们 -> 🧷春秋招笔试合集 🌰 明晚又有米…...

编程日记 2024/8/9 23:03:52

Qt/C++开发监控GB28181系统/取流协议/同时支持udp/tcp被动/tcp主动

一、前言说明在2011版本的gb28181协议中，拉取视频流只要求udp方式，从2016开始要求新增支持tcp被动和tcp主动两种方式，udp理论上会丢包的，所以实际使用过程可能会出现画面花屏的情况，而tcp肯定不丢包，起码…...

编程新知 2025/9/15 13:18:21

如何在看板中体现优先级变化

在看板中有效体现优先级变化的关键措施包括：采用颜色或标签标识优先级、设置任务排序规则、使用独立的优先级列或泳道、结合自动化规则同步优先级变化、建立定期的优先级审查流程。其中，设置任务排序规则尤其重要，因为它让看板视觉上直观地体…...

编程新知 2025/9/14 23:47:25

基于Flask实现的医疗保险欺诈识别监测模型

基于Flask实现的医疗保险欺诈识别监测模型项目截图项目简介社会医疗保险是国家通过立法形式强制实施，由雇主和个人按一定比例缴纳保险费，建立社会医疗保险基金，支付雇员医疗费用的一种医疗保险制度， 它是促进社会文明和进步的…...

编程新知 2025/9/7 2:01:42

Qt Widget类解析与代码注释

#include "widget.h" #include "ui_widget.h"Widget::Widget(QWidget *parent): QWidget(parent), ui(new Ui::Widget) {ui->setupUi(this); }Widget::~Widget() {delete ui; }//解释这串代码，写上注释当然可以！这段代码是 Qt …...

编程新知 2025/6/15 21:47:30

测试markdown--肇兴

day1： 1、去程：7:04 --11:32高铁高铁右转上售票大厅2楼，穿过候车厅下一楼，上大巴车 ￥10/人 **2、到达：**12点多到达寨子，买门票，美团/抖音：￥78人 3、中饭&a…...

编程新知 2025/8/8 2:13:29

高等数学（下）题型笔记（八）空间解析几何与向量代数

目录 0 前言 1 向量的点乘 1.1 基本公式 1.2 例题 2 向量的叉乘 2.1 基础知识 2.2 例题 3 空间平面方程 3.1 基础知识 3.2 例题 4 空间直线方程 4.1 基础知识 4.2 例题 5 旋转曲面及其方程 5.1 基础知识 5.2 例题 6 空间曲面的法线与切平面 6.1 基础知识 6.2…...

编程新知 2025/8/30 19:47:38

C# 类和继承(抽象类)

抽象类抽象类是指设计为被继承的类。抽象类只能被用作其他类的基类。不能创建抽象类的实例。抽象类使用abstract修饰符声明。抽象类可以包含抽象成员或普通的非抽象成员。抽象类的成员可以是抽象成员和普通带实现的成员的任意组合。抽象类自己可以派生自另一个抽象类。例…...

编程新知 2025/8/26 6:51:02

Rust 异步编程

Rust 异步编程引言 Rust 是一种系统编程语言，以其高性能、安全性以及零成本抽象而著称。在多核处理器成为主流的今天，异步编程成为了一种提高应用性能、优化资源利用的有效手段。本文将深入探讨 Rust 异步编程的核心概念、常用库以及最佳实践。异步编程基础什么是异步…...

编程新知 2025/8/8 11:04:42

微信小程序云开发平台MySQL的连接方式

注：微信小程序云开发平台指的是腾讯云开发先给结论：微信小程序云开发平台的MySQL，无法通过获取数据库连接信息的方式进行连接，连接只能通过云开发的SDK连接，具体要参考官方文档： 为什么？ 因为…...

编程新知 2025/9/13 23:29:05

3-11单元格区域边界定位(End属性)学习笔记

返回一个Range 对象，只读。该对象代表包含源区域的区域上端下端左端右端的最后一个单元格。等同于按键 End 向上键(End(xlUp))、End向下键(End(xlDown))、End向左键(End(xlToLeft)End向右键(End(xlToRight)) 注意：它移动的位置必须是相连的有内容的单元格…...

编程新知 2025/9/7 8:51:17

1、下载canal

2、TCP模式的实现

a、canal.properties

b、example/instance.properties

c、canal客户端开发

3、kafka模式的实现

a.canal.properties

b、example/instance.properties

c、启动

d、查看数据生产情况

4、datahub兼容kafka的实现

5、TCP模式+Datahub SDK实现

a、依赖

b、代码实现

相关文章：