数据采集工具之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大厂笔试 ｜ 编程一对一辅导 ✨ 本系列打算持续跟新 春秋招笔试题 👏 感谢大家的订阅➕ 和 喜欢💗 和 手里的小花花🌸 ✨ 笔试合集传送们 -> 🧷春秋招笔试合集 🌰 明晚又有米…...
＜6＞-MySQL表的增删查改
目录 一，create（创建表） 二，retrieve（查询表） 1，select列 2，where条件 三，update（更新表） 四，delete（删除表…...
ESP32读取DHT11温湿度数据
芯片：ESP32 环境：Arduino 一、安装DHT11传感器库 红框的库，别安装错了 二、代码 注意，DATA口要连接在D15上 #include "DHT.h" // 包含DHT库#define DHTPIN 15 // 定义DHT11数据引脚连接到ESP32的GPIO15 #define D…...
【Go】3、Go语言进阶与依赖管理
前言 本系列文章参考自稀土掘金上的 【字节内部课】公开课，做自我学习总结整理。 Go语言并发编程 Go语言原生支持并发编程，它的核心机制是 Goroutine 协程、Channel 通道，并基于CSP（Communicating Sequential Processes࿰…...
linux 下常用变更-8
1、删除普通用户 查询用户初始UID和GIDls -l /home/ ###家目录中查看UID cat /etc/group ###此文件查看GID删除用户1.编辑文件 /etc/passwd 找到对应的行，YW343:x:0:0::/home/YW343:/bin/bash 2.将标红的位置修改为用户对应初始UID和GID： YW3…...
现代密码学 | 椭圆曲线密码学—附py代码
Elliptic Curve Cryptography 椭圆曲线密码学（ECC）是一种基于有限域上椭圆曲线数学特性的公钥加密技术。其核心原理涉及椭圆曲线的代数性质、离散对数问题以及有限域上的运算。 椭圆曲线密码学是多种数字签名算法的基础，例如椭圆曲线数字签…...
【Java_EE】Spring MVC
目录 Spring Web MVC 编辑注解 RestController RequestMapping RequestParam RequestParam RequestBody PathVariable RequestPart 参数传递 注意事项 编辑参数重命名 RequestParam 编辑编辑传递集合 RequestParam 传递JSON数据 编辑RequestBody …...
Java入门学习详细版（一）
大家好，Java 学习是一个系统学习的过程，核心原则就是“理论 实践 坚持”，并且需循序渐进，不可过于着急，本篇文章推出的这份详细入门学习资料将带大家从零基础开始，逐步掌握 Java 的核心概念和编程技能。 …...
3403. 从盒子中找出字典序最大的字符串 I
3403. 从盒子中找出字典序最大的字符串 I 题目链接：3403. 从盒子中找出字典序最大的字符串 I 代码如下： class Solution { public:string answerString(string word, int numFriends) {if (numFriends 1) {return word;}string res;for (int i 0;i &…...
【开发技术】.Net使用FFmpeg视频特定帧上绘制内容
目录 一、目的 二、解决方案 2.1 什么是FFmpeg 2.2 FFmpeg主要功能 2.3 使用Xabe.FFmpeg调用FFmpeg功能 2.4 使用 FFmpeg 的 drawbox 滤镜来绘制 ROI 三、总结 一、目的 当前市场上有很多目标检测智能识别的相关算法，当前调用一个医疗行业的AI识别算法后返回…...
AI书签管理工具开发全记录（十九）：嵌入资源处理
1.前言 📝 在上一篇文章中，我们完成了书签的导入导出功能。本篇文章我们研究如何处理嵌入资源，方便后续将资源打包到一个可执行文件中。 2.embed介绍 🎯 Go 1.16 引入了革命性的 embed 包，彻底改变了静态资源管理的…...