同步mysql数据到redis

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

# 1.将msyql的配置文件挂载出来
docker run -d --name mysql 
-p 3306:3306 
-e MYSQL_ROOT_PASSWORD=123456
-v /etc/mysql:/etc/mysql  
-v /opt/mysql:/var/lib/mysql 
-v /etc/localtime:/etc/localtime
mysql:5.7

# 2.查看binlog日志是否开启
show variables like '%log_bin%';


# 3.设置binlog位置，
# ubuntu /etc/mysql/mysql.conf.d/mysqld.cnf
# centos /etc/my.cnf

log-bin=/var/lib/mysql/mysql-bin
server-id=123454

# 4.查询binlog文件和位置信息

show master status;


# Tip 对于已经存在的docker容器可以进入容器，安装vim编辑配置文件

# ubuntu
apt-get update
apt-get install vim

# centos
yum install vim

# 配置完成后重启容器

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

//监听数据记录
func (h *MyEventHandler) OnRow(ev *canal.RowsEvent) error {
	//库名，表名，行为，数据记录,同步位置
	//record := fmt.Sprintf("%v %v %s %v %v\n", ev.Table.Schema, ev.Table.Name, ev.Action, ev.Rows, ev.Header)

	// 1. 数据库/数据表过滤 只监听想要的表
	matchdb, err := regexp.Match(`client_\d+`, []byte(ev.Table.Schema))
	if err != nil {
		log.Fatal(err.Error())
	}
	matchtb, err := regexp.Match(`personnel_feature`, []byte(ev.Table.Name))
	if err != nil {
		log.Fatal(err.Error())
	}
	// 只监听空间库下面的personnel_feature表
	if !matchdb || !matchtb {
		return nil
	}

	// 增/删/改
	op := ev.Action
	if op == "delete" {
		var spacename, id string
		for columnIndex, currColumn := range ev.Table.Columns {
			spacename = ev.Table.Schema
			if currColumn.Name == "id" {
				tmp := ev.Rows[len(ev.Rows)-1][columnIndex].(int64)
				id = strconv.FormatInt(tmp, 10)
			}
		}
		redis.DelData(spacename, id)

	} else {
		// insert / update
		mid := model.Midlayer{}
		for columnIndex, currColumn := range ev.Table.Columns {
			mid.SpaceName = ev.Table.Schema
		.......
      
		redis.SetData(mid)
	}
	return nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

//参考 https://github.com/gitstliu/MysqlToAll 里面的获取字段和值的方法
func (this *CommonEventHandler) OnRow(e *canal.RowsEvent) error {
	log4go.Debug("OnRow")
	entity := &common.RawLogEntity{}
	entity.Action = e.Action 
	entity.Rows = e.Rows
	entity.TableName = e.Table.Name
	entity.Header = []string{}
	entity.HeaderMap = map[string]int{}
	entity.ValueMap = map[string]interface{}{}

	for columnIndex, currColumn := range e.Table.Columns {
		entity.Header = append(entity.Header, currColumn.Name)
		entity.HeaderMap[currColumn.Name] = columnIndex
		entity.ValueMap[currColumn.Name] = e.Rows[len(e.Rows)-1][columnIndex]
	}
	log4go.Debug(entity)
	this.CurrOutput.Write(entity)

	return nil
}