canal实战(三)|canal数据消费到kafka

叁歲伎倆 2022-12-22 14:28 2992阅读 0赞

> ***\#\# 本文章改编于阿里开源工具Canal，原版网址 [https://github.com/alibaba/canal][https_github.com_alibaba_canal]***

*  canal第一辑 [canal的配置安装与服务启动][canal] 
 *  canal第二辑 [canal java客户端][canal java]

canal的原理我们在第一辑里已经介绍，现在我们需要把canal采集到的日志数据消费到kafka，并通过kafka把数据进行进一步入库，上云等消费操作。

首先我们把Kafka进行安装配置，涉及到的有zookeeper，kafka，有的服务器还要装java环境等，kafka的安装我之前有讲过，来，给你一个[kafka实施传送站][kafka]

安装配置并开启kafka后，我们需要把kafka的相关信息补充到canal配置中，以便canal可以把数据发送到kafka

*  **vim ./conf/example/instance.properties**

#################################################
    ## mysql serverId , v1.0.26+ will autoGen
    # canal.instance.mysql.slaveId=0
    
    # enable gtid use true/false
    canal.instance.gtidon=false
    
    # position info
    # 需要改为MySQL实例所在的的IP,若为集群以逗号分隔
    canal.instance.master.address=192.168.xxx.xxx:3306
    canal.instance.master.journal.name=mysql-bin.000001
    canal.instance.master.position=123439266
    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/hhy_test
    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=canal
    canal.instance.dbPassword=canal
    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
    # topic改为实际
    canal.mq.topic=canal
    # 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,.*\\..*
    #################################################

*  **vim 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 = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
    
    # zookeeper的ip
    canal.zkServers = 192.168.xxx.xxx
    # flush data to zk
    canal.zookeeper.flush.period = 1000
    canal.withoutNetty = false
    # tcp, kafka, RocketMQ,这里改为kafka
    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.url = jdbc:mysql://127.0.0.1:3306/hhy_test
    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
    
    # aliyun ak/sk , support rds/mq
    canal.aliyun.accessKey =
    canal.aliyun.secretKey =
    
    #################################################
    ######### 		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 = 5
    
    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
    
    ##################################################
    ######### 		     MQ 		     #############
    ##################################################
    canal.mq.servers = 192.168.xxx.xxx:9092
    canal.mq.retries = 0
    canal.mq.batchSize = 16384
    canal.mq.maxRequestSize = 1048576
    canal.mq.lingerMs = 100
    canal.mq.bufferMemory = 33554432
    canal.mq.canalBatchSize = 50
    canal.mq.canalGetTimeout = 100
    canal.mq.flatMessage = true
    canal.mq.compressionType = none
    canal.mq.acks = all
    #canal.mq.properties. =
    canal.mq.producerGroup = canal
    # Set this value to "cloud", if you want open message trace feature in aliyun.
    canal.mq.accessChannel = local
    # aliyun mq namespace
    #canal.mq.namespace =
    
    ##################################################
    #########     Kafka Kerberos Info    #############
    ##################################################
    canal.mq.kafka.kerberos.enable = false
    canal.mq.kafka.kerberos.krb5FilePath = "../conf/kerberos/krb5.conf"
    canal.mq.kafka.kerberos.jaasFilePath = "../conf/kerberos/jaas.conf"

配置好之后重新启动canal，看日志不报错的话，离成功就很近了

这里我们打开kafka的消费

bin/kafka-console-consumer.sh --bootstrap-server 192.168.xxx.xxx:9092 --topic canal --from-beginning

然后运行MySQL

insert into hhy_test.canal_test VALUES(1,'test',now())

这里就会在消费端看到对应的数据出来![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21vY2hvdTExMQ_size_16_color_FFFFFF_t_70][]

这个时候kafka出来的数据是json格式的，我们可以在java里进行处理(好像这是下一篇kafka要讲的事情，不管了，一篇全写完了！！！)

配置kafka的依赖

> <dependency>  
>     <groupId>org.apache.kafka</groupId>  
>     <artifactId>kafka-clients</artifactId>  
>     <version>1.1.0</version>  
> </dependency>

然后直接输入kafka的java控制消费端代码，网上这一类的代码非常多

package com.hhy.kafka;
    
    import java.util.Collections;
    import java.util.Properties;
    
    import com.alibaba.fastjson.JSON;
    import com.alibaba.fastjson.JSONArray;
    import com.alibaba.fastjson.JSONObject;
    import org.apache.kafka.clients.consumer.*;
    import org.apache.kafka.common.serialization.StringDeserializer;
    
    public class Consumer {
        public static void main(String[] args) {
            Properties p = new Properties();
            p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.xxx.xxx:9092");
            p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            p.put(ConsumerConfig.GROUP_ID_CONFIG, "canal");
    
            KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(p);
            kafkaConsumer.subscribe(Collections.singletonList(Producer.topic));// 订阅消息
    
            while (true) {
                ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
                for (ConsumerRecord<String, String> record : records) {
                    String value = record.value();
                    JSONObject obj = JSON.parseObject(value);
                    if ("INSERT".equalsIgnoreCase(obj.getString("type"))
                            && "hhy_test".equalsIgnoreCase(obj.getString("database"))
                            && "canal_test".equalsIgnoreCase(obj.getString("table"))) {
                        JSONArray dataArry = obj.getJSONArray("data");
                        if (dataArry != null && dataArry.size() > 0) {
                            for (int i = 0; i < dataArry.size(); i++) {
                                System.out.println(dataArry.getJSONObject(i).toJSONString());
                            }
                        }
                    }
                }
            }
        }
    }

运行结果出现

![20201120183735192.jpg][]

即为正常消费

[https_github.com_alibaba_canal]: https://github.com/alibaba/canal
[canal]: https://blog.csdn.net/mochou111/article/details/109808581
[canal java]: https://blog.csdn.net/mochou111/article/details/109854641
[kafka]: https://blog.csdn.net/mochou111/article/details/103700433
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21vY2hvdTExMQ_size_16_color_FFFFFF_t_70]: /images/20221120/ab195f1736814477a7881f14d844251c.png
[20201120183735192.jpg]: /images/20221120/cd36901a6a844e59ba841a03ada693bc.png