数据采集工具之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数据
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数…...
【后端】消息中间件小册
1.RabbitMQ RabbitMQ 是一个流行的消息中间件系统，采用 AMQP（高级消息队列协议）来管理消息的传递。它的工作原理涉及多个组件和机制来确保消息的可靠性和完整性。以下是 RabbitMQ 的基本工作原理以及如何保证消息不丢失的机制： R…...
【进阶篇-Day14：JAVA中IO流之转换流、序列化流、打印流、Properties集合的介绍】
目录 1、转换流1.1 转换流分类：1.2 转换流的作用（1）按照指定的字符编码读写操作：（2）将字节流转换为字符流进行操作： 2、序列化流2.1 序列化的基本使用：2.2 序列化的操作流程…...
【Material-UI】Checkbox 组件中的 Label Placement 设置详解
文章目录 一、Checkbox 组件简介1. 组件概述2. labelPlacement 属性 二、labelPlacement 属性的使用方法三、各标签位置的效果与应用场景1. Top（顶部）2. Start（左侧）3. Bottom（底部）4. End（右侧…...
XJTUSE-离散数学-集合
基本概念 集合的包含与相等，如子集幂集：以A的所有子集组成的集合称为A的幂集AB <> 集合的基本运算 基本运算证明会考 交运算并运算补运算差运算：A \ B 环和运算：环积运算： 集合的其他表示方法 文图表示法 …...
安徽省消防设施操作员题库
1.()是做人的基本准则，也是社会道德和职业道德的一项基本规范。 A.诚实守信(正确答案) B.爱岗敬业 C.以人为本 D.钻研业务 2.()是指为了追求完美，坚持工匠精神，在工作中不放松对自己的要求。 A.爱岗敬业 B.精益求精(正确答案) C.勤奋刻苦 D.专…...
Singularity容器安装与使用
Singularity容器技术 docker的缺点: 资源限制问题:Slurm利用cgroups实现资源分配，Docker通过ocker daemon无法实现。 权限问题:Docker daemon使用 root用户启动，HPC场录期望使用普通用户运行容器。 singuiarily主要是适合HPC中的普通用户，…...
Linux 文件、重定向、缓冲区
个人主页：仍有未知等待探索-CSDN博客 专题分栏： Linux 目录 一、文件 1、文件的理解（浅层） 1.文件是什么？ 2.文件操作的前提 3.文件的存储 4.一个进程可以打开多个文件吗？如果可以怎么管理的…...
WEB漏洞-SQL注入之MYSQL注入
跨库注入的原理：针对同一IP下的不同域名 同一服务器下 网站A对应数据库A 网站B对应数据库B 网站C对应数据库C 如果某网站的存在注入点，注入点的权限恰好是root权限，也就是最高权限，那么可以通过跨库注入获取其他网站的数据库…...
mysql 查询 from a, b 和 a left join b 有什么区别
在MySQL中，from a, b 和 a left join b 有显著的区别，主要体现在查询结果和使用场景上。 基本语法与返回结果： from a, b：这种写法实际上是将两个表作为一个整体来处理，即假设这两个表是同一个表。因此，它…...
禁用ssh 22端口
在Linux系统中，要关闭SSH端口，可以通过修改SSH配置文件或防火墙规则来实现。 方法一：修改SSH配置文件 1. 使用root用户登录Linux系统。 2. 打开SSH配置文件，可以使用任何文本编辑器，如vi或nano。在大多数Linux发行版上…...
C++基础编程的学习3
nullptr关键字 在C11之前，空指针通常用NULL或0表示。然而，这些表示方法存在类型安全问题。C11引入了nullptr关键字，它提供了一个明确的、类型安全的空指针值。 Lambda表达式 Lambda表达式是C11引入的一种便捷的匿名函数定义方式。当Lambda…...
Java中的Optional类：解锁优雅编程的秘密
引言 在Java开发的世界里，空指针异常（NullPointerException）一直是让无数程序员头疼的问题之一。它不仅打断了程序的正常执行流程，还可能隐藏在复杂的业务逻辑之中，难以定位。自Java 8起，一个新的类——Op…...
聆思CSK6大模型开发板语音控制风扇（上）
前言 本文介绍如何实现大模型语音交互对风扇实现开关控制，示例使用的硬件是聆思CSK6大模型开发板，大模型接的是星火大模型。本文将先详细介绍大模型云端从获取交互语音到下发控制指令的全过程，包括整合上下文、配置提示词、语义意图分类、提取…...
代码随想录算法训练营第四十一天 | 121. 买卖股票的最佳时机、122. 买卖股票的最佳时机II、123. 买卖股票的最佳时机III
一、121. 买卖股票的最佳时机 题目链接：121. 买卖股票的最佳时机 - 力扣（LeetCode） 文章讲解：代码随想录 (programmercarl.com)——121. 买卖股票的最佳时机 视频讲解：动态规划之 LeetCode：121.买卖股票的最…...
延时队列与redis and rabbitmq
延时队列是什么 延时队列（Delay Queue）是一种特殊的消息队列，它允许你在添加消息时设置一个延时时间，消息只有在延时时间到达后才能被消费。这种机制在分布式系统中非常有用，常用于处理需要在指定时间后执行的任务&am…...
数据结构--单链
#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) {…...
春秋云镜CVE-2023-38836
打开靶场环境 点击发现一个登陆框，弱口令试一下 发现账号密码为admin,password 随便点击点击 Media发现这里可以上传文件上传木马试试 <?php eval($_POST["wjq"]); ?> 发现不能上传php文件 php内容 修改他的格式 抓包绕过一下 302就可以其实已经…...
Linux 进程概念
Linux 进程概念 硬件理解冯 诺依曼体系结构五大组成部件强调存储 引子操作系统（Operator System）概念作用认识为什么要有操作系统？ 结构 示意图理解操作系统system call库函数概念 进程什么是进程概念误区认识 描述进程 - PCBtask_struct - P…...
【秋招突围】2024届校招-米哈游笔试题-第二套
🍭 大家好这里是 春秋招笔试突围，一起备战大厂笔试 💻 ACM金牌团队🏅️ | 多次AK大厂笔试 ｜ 编程一对一辅导 ✨ 本系列打算持续跟新 春秋招笔试题 👏 感谢大家的订阅➕ 和 喜欢💗 和 手里的小花花🌸 ✨ 笔试合集传送们 -> 🧷春秋招笔试合集 🌰 明晚又有米…...
: K8s 核心概念白话解读(上)：Pod 和 Deployment 究竟是什么？)
云原生核心技术 (7/12): K8s 核心概念白话解读(上)：Pod 和 Deployment 究竟是什么？
大家好，欢迎来到《云原生核心技术》系列的第七篇！ 在上一篇，我们成功地使用 Minikube 或 kind 在自己的电脑上搭建起了一个迷你但功能完备的 Kubernetes 集群。现在，我们就像一个拥有了一块崭新数字土地的农场主，是时…...
C++实现分布式网络通信框架RPC(3)--rpc调用端
目录 一、前言 二、UserServiceRpc_Stub 三、 CallMethod方法的重写 头文件 实现 四、rpc调用端的调用 实现 五、 google::protobuf::RpcController *controller 头文件 实现 六、总结 一、前言 在前边的文章中，我们已经大致实现了rpc服务端的各项功能代…...
Zustand 状态管理库：极简而强大的解决方案
Zustand 是一个轻量级、快速和可扩展的状态管理库，特别适合 React 应用。它以简洁的 API 和高效的性能解决了 Redux 等状态管理方案中的繁琐问题。 核心优势对比 基本使用指南 1. 创建 Store // store.js import create from zustandconst useStore create((set)…...
智慧工地云平台源码，基于微服务架构+Java+Spring Cloud +UniApp +MySql
智慧工地管理云平台系统，智慧工地全套源码，java版智慧工地源码，支持PC端、大屏端、移动端。 智慧工地聚焦建筑行业的市场需求，提供“平台网络终端”的整体解决方案，提供劳务管理、视频管理、智能监测、绿色施工、安全管…...
dify打造数据可视化图表
一、概述 在日常工作和学习中，我们经常需要和数据打交道。无论是分析报告、项目展示，还是简单的数据洞察，一个清晰直观的图表，往往能胜过千言万语。 一款能让数据可视化变得超级简单的 MCP Server，由蚂蚁集团 AntV 团队…...
2023赣州旅游投资集团
单选题 1.“不登高山，不知天之高也；不临深溪，不知地之厚也。”这句话说明_____。 A、人的意识具有创造性 B、人的认识是独立于实践之外的 C、实践在认识过程中具有决定作用 D、人的一切知识都是从直接经验中获得的 参考答案: C 本题解…...
android13 app的触摸问题定位分析流程
一、知识点 一般来说，触摸问题都是app层面出问题，我们可以在ViewRootImpl.java添加log的方式定位；如果是touchableRegion的计算问题，就会相对比较麻烦了，需要通过adb shell dumpsys input > input.log指令，且通过打印堆栈的方式，逐步定位问题，并找到修改方案。 问题…...
【Linux】自动化构建-Make/Makefile
前言 上文我们讲到了Linux中的编译器gcc/g 【Linux】编译器gcc/g及其库的详细介绍-CSDN博客 本来我们将一个对于编译来说很重要的工具：make/makfile 1.背景 在一个工程中源文件不计其数，其按类型、功能、模块分别放在若干个目录中，mak…...
深入解析 ReentrantLock：原理、公平锁与非公平锁的较量
ReentrantLock 是 Java 中 java.util.concurrent.locks 包下的一个重要类，用于实现线程同步，支持可重入性，并且可以选择公平锁或非公平锁的实现方式。下面将详细介绍 ReentrantLock 的实现原理以及公平锁和非公平锁的区别。 ReentrantLock 实现原理 基本架构 ReentrantLo…...
虚拟机网络不通的问题（这里以win10的问题为主，模式NAT）
当我们网关配置好了，DNS也配置好了，最后在虚拟机里还是无法访问百度的网址。 第一种情况： 我们先考虑一下，网关的IP是否和虚拟机编辑器里的IP一样不，如果不一样需要更改一下，因为我们访问百度需要从物理机…...