canal+kafka订阅Mysql binlog将数据异构到elasticsearch(或其他存储方式)

canal本质就是"冒充"从库,通过订阅mysql bin-log来获取数据库的更改信息。

mysql配置(my.cnf)

mysql需要配置my.cnf开启bin-log日志并且将bin-log日志格式设置为row, 同时为了防止bin-log日志占用过多磁盘,可以设置一下过期时间,

[mysqld]
log-bin=mysql-bin # 打开binlog
binlog-format=ROW # ROW格式
server_id=1 # mysql Replication 需要设置 在mysql集群里唯一

expire_logs_days=7 # binlog文件保存7天
max_binlog_size = 500m # 每个binlog日志文件大小 

canal配置

除了kafka之外,canal还支持将数据库修改的消息投递到rocketMQ, 或者不经过消息队列直接投递到canal的客户端,然后再在客户端实现自己的代码(如写入其他存储/其他消息队列) ,但是只能选其一。而如果选择canal客户端的方式, 一个canal server也只能将消息投递到一个canal client。

但是可以开启多个canal服务端和客户端(同一个实例,即对mysql来说只算是一个从库), 他们通过zookeeper保证只有一个服务端和客户端是有效的，其他只是作为HA的冗余。

然后需要修改canal目录下(以下为近最小配置)

conf/example/instance.properties

## mysql serverId
canal.instance.mysql.slaveId = 1234

# 数据库address
canal.instance.master.address = 127.0.0.1:3306

# 数据库账号密码
canal.instance.dbUsername = canal  
canal.instance.dbPassword = canal

# 需要订阅的数据库.表名 默认全部
canal.instance.filter.regex = .*\\..*  # 去掉转义符其实就是 .*..*

# topic名 固定
canal.mq.topic=canal

# 动态topic
# canal.mq.dynamicTopic=mytest,.*,mytest.user,mytest\..*,.*\..*

# 库名.表名: 唯一主键，多个表之间用逗号分隔
# canal.mq.partitionHash=mytest.person:id,mytest.role:id

其中动态topic 和 主键hash看上去有点难理解,去看其他人的博客找到的解释和例子如下

动态topic

canal 1.1.3版本之后, 支持配置格式：schema 或 schema.table，多个配置之间使用逗号分隔

例子1：test.test 指定匹配的单表，发送到以 test_test为名字的topic上
例子2：.…* 匹配所有表，每个表都会发送到各自表名的topic上
例子3：test 指定匹配对应的库，一个库的所有表都会发送到库名的topic上
例子4：test.* 指定匹配的表达式，针对匹配的表会发送到各自表名的topic上
例子5：test,test1.test1，指定多个表达式，会将test库的表都发送到test的topic上，test1.test1的表发送到对应的test1_test1 topic上，其余的表发送到默认的canal.mq.topic值
支持指定topic名称匹配, 配置格式：topicName:schema 或 schema.table，多个配置之间使用逗号分隔, 多组之间使用 ; 分隔

例子：test:test,test1.test1;test2:test2,test3.test1 针对匹配的表会发送到指定的topic上
————————————————
版权声明：本文为CSDN博主「BillowX_」的原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/weixin_35852328/article/details/87600871

主键

canal 1.1.3版本之后, 支持配置格式：schema.table:pk1^pk2，多个配置之间使用逗号分隔

例子1：test.test:pk1^pk2 指定匹配的单表，对应的hash字段为pk1 + pk2
例子2：.…:id 正则匹配，指定所有正则匹配的表对应的hash字段为id
例子3：.…:pkpkpk 正则匹配，指定所有正则匹配的表对应的hash字段为表主键(自动查找)
例子4: 匹配规则啥都不写，则默认发到0这个partition上
例子5：.…* ，不指定pk信息的正则匹配，将所有正则匹配的表,对应的hash字段为表名
按表hash: 一张表的所有数据可以发到同一个分区，不同表之间会做散列 (会有热点表分区过大问题)
例子6: test.test:id,.…* , 针对test的表按照id散列,其余的表按照table散列
注意：大家可以结合自己的业务需求，设置匹配规则，多条匹配规则之间是按照顺序进行匹配(命中一条规则就返回)
————————————————
版权声明：本文为CSDN博主「BillowX_」的原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/weixin_35852328/article/details/87600871

最后实现消费kafka上canal topic上消息的代码

这里以go为例,可以写入到elasticsearch/redis/其他

package main

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/elastic/go-elasticsearch/esapi"
	"github.com/elastic/go-elasticsearch/v6"
	"os"
)

var esClient *elasticsearch.Client

func init() {
	var err error
	config := elasticsearch.Config{}
	config.Addresses = []string{"http://127.0.0.1:9200"}
	esClient, err = elasticsearch.NewClient(config)
	checkErr(err)
}

type Msg struct {
	Data []struct {
		Id string `json:"id"`
		A  string `json:"a"`
	} `json:"data"`
	Type     string `json:"type"`
	DataBase string `json:"database"`
	Table    string `json:"table"`
}

func checkErr(err error) {
	if err != nil {
		fmt.Println(err)
		os.Exit(-1)
	}
}

type Consumer struct{}

func (c *Consumer) Setup(sarama.ConsumerGroupSession) error {
	return nil
}

func (c *Consumer) Cleanup(sarama.ConsumerGroupSession) error {
	return nil
}

func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for message := range claim.Messages() {
    // fmt.Printf("Message claimed: value = %s, timestamp = %v, topic = %s", string(message.Value), message.Timestamp, message.Topic)
		msg := &Msg{}
		err := json.Unmarshal(message.Value, msg)
		checkErr(err)
		if msg.DataBase == "test" && msg.Table == "tbltest" {
			if msg.Type == "INSERT" {
				for k, _ := range msg.Data {
					// 写elasticsearch 逻辑
					body := map[string]interface{}{
						"id": msg.Data[k].Id,
						"a":  msg.Data[k].A,
					}
					jsonBody, _ := json.Marshal(body)
					req := esapi.IndexRequest{
						Index:      msg.DataBase,
						DocumentID: msg.Table + "_" + msg.Data[k].Id,
						Body:       bytes.NewReader(jsonBody),
					}
					res, err := req.Do(context.Background(), esClient)
					checkErr(err)
					fmt.Println(res.String())
					res.Body.Close()
					session.MarkMessage(message, "")
				}
			}
		}
	}
	return nil
}

func main() {
	consumer := &Consumer{}

	config := sarama.NewConfig()
	config.Version = sarama.MaxVersion
	client, err := sarama.NewConsumerGroup([]string{"127.0.0.1:9092"}, "tg", config)
	checkErr(err)
	ctx := context.Background()

	client.Consume(ctx, []string{"canal"}, consumer)
}