6 changed files with 576 additions and 201 deletions
@ -1,164 +1,461 @@ |
|||||
package main |
package main |
||||
|
|
||||
import ( |
import ( |
||||
|
"database/sql" |
||||
|
"encoding/csv" |
||||
"flag" |
"flag" |
||||
"fmt" |
"fmt" |
||||
"log" |
"log" |
||||
|
"os" |
||||
|
"path/filepath" |
||||
|
"strconv" |
||||
"strings" |
"strings" |
||||
|
"sync" |
||||
|
"time" |
||||
|
|
||||
"github.com/go-mysql-org/go-mysql/canal" |
_ "github.com/go-sql-driver/mysql" |
||||
"github.com/go-mysql-org/go-mysql/mysql" |
|
||||
) |
) |
||||
|
|
||||
// 自定义事件处理结构体
|
// BackupConfig 备份配置
|
||||
type MyEventHandler struct { |
type BackupConfig struct { |
||||
canal.DummyEventHandler |
Host string |
||||
|
Port int |
||||
|
User string |
||||
|
Password string |
||||
|
Database string |
||||
|
BackupDir string |
||||
|
KeepDays int |
||||
|
Async bool |
||||
} |
} |
||||
|
|
||||
// 处理行事件
|
// BackupResult 备份结果
|
||||
func (h *MyEventHandler) OnRow(e *canal.RowsEvent) error { |
type BackupResult struct { |
||||
table := e.Table |
Success bool |
||||
sql := "" |
Message string |
||||
|
StartTime time.Time |
||||
|
EndTime time.Time |
||||
|
Filename string |
||||
|
} |
||||
|
|
||||
|
// BackupManager 备份管理器
|
||||
|
type BackupManager struct { |
||||
|
config BackupConfig |
||||
|
db *sql.DB |
||||
|
mutex sync.Mutex |
||||
|
running bool |
||||
|
} |
||||
|
|
||||
|
// NewBackupManager 创建新的备份管理器
|
||||
|
func NewBackupManager(config BackupConfig) (*BackupManager, error) { |
||||
|
dsn := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?parseTime=true", |
||||
|
config.User, config.Password, config.Host, config.Port, config.Database) |
||||
|
|
||||
|
db, err := sql.Open("mysql", dsn) |
||||
|
if err != nil { |
||||
|
return nil, fmt.Errorf("无法连接数据库: %v", err) |
||||
|
} |
||||
|
|
||||
|
// 测试连接
|
||||
|
if err := db.Ping(); err != nil { |
||||
|
return nil, fmt.Errorf("数据库连接失败: %v", err) |
||||
|
} |
||||
|
|
||||
|
return &BackupManager{ |
||||
|
config: config, |
||||
|
db: db, |
||||
|
}, nil |
||||
|
} |
||||
|
|
||||
switch e.Action { |
// 获取数据库中所有表
|
||||
case canal.InsertAction: |
func (m *BackupManager) getTables() ([]string, error) { |
||||
// 处理插入事件
|
query := "SHOW TABLES" |
||||
columns := make([]string, len(table.Columns)) |
rows, err := m.db.Query(query) |
||||
for i, col := range table.Columns { |
if err != nil { |
||||
columns[i] = col.Name |
return nil, fmt.Errorf("查询表失败: %v", err) |
||||
} |
} |
||||
|
defer rows.Close() |
||||
|
|
||||
values := make([]string, len(e.Rows[0])) |
var tables []string |
||||
for i, val := range e.Rows[0] { |
for rows.Next() { |
||||
values[i] = formatValue(val) |
var table string |
||||
|
if err := rows.Scan(&table); err != nil { |
||||
|
return nil, fmt.Errorf("扫描表名失败: %v", err) |
||||
|
} |
||||
|
tables = append(tables, table) |
||||
} |
} |
||||
|
|
||||
sql = fmt.Sprintf("INSERT INTO `%s`.`%s` (%s) VALUES (%s);", |
return tables, nil |
||||
table.Schema, table.Name, |
} |
||||
strings.Join(columns, ", "), |
|
||||
strings.Join(values, ", ")) |
|
||||
|
|
||||
case canal.UpdateAction: |
// 获取表结构
|
||||
// 处理更新事件
|
func (m *BackupManager) getTableSchema(table string) (string, error) { |
||||
oldRow := e.Rows[0] |
query := fmt.Sprintf("SHOW CREATE TABLE `%s`", table) |
||||
newRow := e.Rows[1] |
rows, err := m.db.Query(query) |
||||
|
if err != nil { |
||||
|
return "", fmt.Errorf("查询表结构失败: %v", err) |
||||
|
} |
||||
|
defer rows.Close() |
||||
|
|
||||
sets := make([]string, 0) |
var ( |
||||
wheres := make([]string, 0) |
tbl string |
||||
|
sql string |
||||
|
) |
||||
|
|
||||
for i, col := range table.Columns { |
if rows.Next() { |
||||
if oldRow[i] != newRow[i] { |
if err := rows.Scan(&tbl, &sql); err != nil { |
||||
sets = append(sets, fmt.Sprintf("`%s` = %s", col.Name, formatValue(newRow[i]))) |
return "", fmt.Errorf("扫描表结构失败: %v", err) |
||||
} |
} |
||||
wheres = append(wheres, fmt.Sprintf("`%s` = %s", col.Name, formatValue(oldRow[i]))) |
|
||||
} |
} |
||||
|
|
||||
sql = fmt.Sprintf("UPDATE `%s`.`%s` SET %s WHERE %s;", |
return sql, nil |
||||
table.Schema, table.Name, |
} |
||||
strings.Join(sets, ", "), |
|
||||
strings.Join(wheres, " AND ")) |
|
||||
|
|
||||
case canal.DeleteAction: |
// 备份表数据到CSV文件
|
||||
// 处理删除事件
|
func (m *BackupManager) backupTableData(table string, dataFile string) error { |
||||
wheres := make([]string, len(table.Columns)) |
// 获取表列信息
|
||||
for i, col := range table.Columns { |
columnsQuery := fmt.Sprintf("DESCRIBE `%s`", table) |
||||
wheres[i] = fmt.Sprintf("`%s` = %s", col.Name, formatValue(e.Rows[0][i])) |
rows, err := m.db.Query(columnsQuery) |
||||
|
if err != nil { |
||||
|
return fmt.Errorf("查询表列信息失败: %v", err) |
||||
} |
} |
||||
|
defer rows.Close() |
||||
|
|
||||
sql = fmt.Sprintf("DELETE FROM `%s`.`%s` WHERE %s;", |
var columns []string |
||||
table.Schema, table.Name, |
for rows.Next() { |
||||
strings.Join(wheres, " AND ")) |
var ( |
||||
|
field, typ, null, key, extra string |
||||
|
defaultVal sql.NullString |
||||
|
) |
||||
|
if err := rows.Scan(&field, &typ, &null, &key, &defaultVal, &extra); err != nil { |
||||
|
return fmt.Errorf("扫描表列信息失败: %v", err) |
||||
|
} |
||||
|
columns = append(columns, field) |
||||
} |
} |
||||
|
|
||||
if sql != "" { |
// 打开数据文件
|
||||
fmt.Println(sql) |
file, err := os.Create(dataFile) |
||||
|
if err != nil { |
||||
|
return fmt.Errorf("创建数据文件失败: %v", err) |
||||
} |
} |
||||
|
defer file.Close() |
||||
|
|
||||
return nil |
writer := csv.NewWriter(file) |
||||
} |
defer writer.Flush() |
||||
|
|
||||
|
// 写入列名
|
||||
|
if err := writer.Write(columns); err != nil { |
||||
|
return fmt.Errorf("写入列名失败: %v", err) |
||||
|
} |
||||
|
|
||||
// string
|
// 分页查询数据
|
||||
func (h *MyEventHandler) String() string { return "MyEventHandler" } |
limit := 1000 |
||||
|
offset := 0 |
||||
|
columnsStr := "`" + strings.Join(columns, "`, `") + "`" |
||||
|
|
||||
|
for { |
||||
|
dataQuery := fmt.Sprintf("SELECT %s FROM `%s` LIMIT %d OFFSET %d", |
||||
|
columnsStr, table, limit, offset) |
||||
|
|
||||
|
dataRows, err := m.db.Query(dataQuery) |
||||
|
if err != nil { |
||||
|
return fmt.Errorf("查询表数据失败: %v", err) |
||||
|
} |
||||
|
|
||||
|
// 获取列类型信息
|
||||
|
columnTypes, err := dataRows.ColumnTypes() |
||||
|
if err != nil { |
||||
|
dataRows.Close() |
||||
|
return fmt.Errorf("获取列类型失败: %v", err) |
||||
|
} |
||||
|
|
||||
// 处理DDL事件
|
rowCount := 0 |
||||
// func (h *MyEventHandler) OnDDL(nextPos mysql.Position, queryEvent *replication.QueryEvent) error {
|
for dataRows.Next() { |
||||
// sql := string(queryEvent.Query)
|
rowCount++ |
||||
// if sql != "" {
|
// 创建一个接口切片来存储一行数据
|
||||
// fmt.Println(sql + ";")
|
values := make([]interface{}, len(columns)) |
||||
// }
|
valuePtrs := make([]interface{}, len(columns)) |
||||
// return nil
|
|
||||
// }
|
|
||||
|
|
||||
// 格式化值为SQL表示形式
|
for i := range values { |
||||
func formatValue(value interface{}) string { |
valuePtrs[i] = &values[i] |
||||
|
} |
||||
|
|
||||
|
// 扫描行数据
|
||||
|
if err := dataRows.Scan(valuePtrs...); err != nil { |
||||
|
dataRows.Close() |
||||
|
return fmt.Errorf("扫描表数据失败: %v", err) |
||||
|
} |
||||
|
|
||||
|
// 处理数据并写入CSV
|
||||
|
csvRow := make([]string, len(columns)) |
||||
|
for i, val := range values { |
||||
|
colType := columnTypes[i].DatabaseTypeName() |
||||
|
csvRow[i] = formatValue(val, colType) |
||||
|
} |
||||
|
|
||||
|
if err := writer.Write(csvRow); err != nil { |
||||
|
dataRows.Close() |
||||
|
return fmt.Errorf("写入数据到CSV失败: %v", err) |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
dataRows.Close() |
||||
|
|
||||
|
// 如果获取的行数小于limit，说明已经到最后一页
|
||||
|
if rowCount < limit { |
||||
|
break |
||||
|
} |
||||
|
|
||||
|
offset += limit |
||||
|
} |
||||
|
|
||||
|
return nil |
||||
|
} |
||||
|
|
||||
|
// 格式化值为字符串
|
||||
|
func formatValue(value interface{}, colType string) string { |
||||
if value == nil { |
if value == nil { |
||||
return "NULL" |
return "NULL" |
||||
} |
} |
||||
|
|
||||
switch v := value.(type) { |
switch v := value.(type) { |
||||
case int, int8, int16, int32, int64, uint, uint8, uint16, uint32, uint64, float32, float64: |
|
||||
return fmt.Sprintf("%v", v) |
|
||||
case []byte: |
case []byte: |
||||
// 处理二进制数据
|
// 处理二进制数据
|
||||
return fmt.Sprintf("X'%x'", v) |
return string(v) |
||||
case string: |
case string: |
||||
// 转义单引号
|
return v |
||||
return fmt.Sprintf("'%s'", strings.ReplaceAll(v, "'", "''")) |
case int64: |
||||
|
return strconv.FormatInt(v, 10) |
||||
|
case float64: |
||||
|
return strconv.FormatFloat(v, 'f', -1, 64) |
||||
|
case bool: |
||||
|
if v { |
||||
|
return "1" |
||||
|
} |
||||
|
return "0" |
||||
|
case time.Time: |
||||
|
return v.Format("2006-01-02 15:04:05") |
||||
default: |
default: |
||||
return fmt.Sprintf("'%v'", v) |
return fmt.Sprintf("%v", v) |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
// 创建备份目录
|
||||
|
func (m *BackupManager) createBackupDir(timestamp string) (string, error) { |
||||
|
backupPath := filepath.Join(m.config.BackupDir, |
||||
|
fmt.Sprintf("%s_%s", m.config.Database, timestamp)) |
||||
|
|
||||
|
if err := os.MkdirAll(backupPath, 0755); err != nil { |
||||
|
return "", fmt.Errorf("创建备份目录失败: %v", err) |
||||
|
} |
||||
|
|
||||
|
return backupPath, nil |
||||
|
} |
||||
|
|
||||
|
// 备份表结构到文件
|
||||
|
func (m *BackupManager) backupTableSchema(table string, schemaFile string) error { |
||||
|
schema, err := m.getTableSchema(table) |
||||
|
if err != nil { |
||||
|
return err |
||||
|
} |
||||
|
|
||||
|
// 写入表结构到文件
|
||||
|
if err := os.WriteFile(schemaFile, []byte(schema+";\n"), 0644); err != nil { |
||||
|
return fmt.Errorf("写入表结构失败: %v", err) |
||||
} |
} |
||||
|
|
||||
|
return nil |
||||
|
} |
||||
|
|
||||
|
// 清理旧备份
|
||||
|
func (m *BackupManager) cleanOldBackups() error { |
||||
|
if m.config.KeepDays <= 0 { |
||||
|
return nil |
||||
|
} |
||||
|
|
||||
|
cutoffTime := time.Now().AddDate(0, 0, -m.config.KeepDays) |
||||
|
|
||||
|
// 读取备份目录
|
||||
|
entries, err := os.ReadDir(m.config.BackupDir) |
||||
|
if err != nil { |
||||
|
return fmt.Errorf("读取备份目录失败: %v", err) |
||||
|
} |
||||
|
|
||||
|
for _, entry := range entries { |
||||
|
if !entry.IsDir() { |
||||
|
continue |
||||
|
} |
||||
|
|
||||
|
// 检查目录名是否符合备份命名规范
|
||||
|
if !strings.HasPrefix(entry.Name(), m.config.Database+"_") { |
||||
|
continue |
||||
|
} |
||||
|
|
||||
|
// 获取目录修改时间
|
||||
|
info, err := entry.Info() |
||||
|
if err != nil { |
||||
|
log.Printf("获取目录信息失败: %v", err) |
||||
|
continue |
||||
|
} |
||||
|
|
||||
|
// 如果目录超过保留天数，删除它
|
||||
|
if info.ModTime().Before(cutoffTime) { |
||||
|
path := filepath.Join(m.config.BackupDir, entry.Name()) |
||||
|
if err := os.RemoveAll(path); err != nil { |
||||
|
log.Printf("删除旧备份失败: %v", err) |
||||
|
} else { |
||||
|
log.Printf("已删除旧备份: %s", path) |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
return nil |
||||
|
} |
||||
|
|
||||
|
// 执行备份
|
||||
|
func (m *BackupManager) performBackup() BackupResult { |
||||
|
result := BackupResult{ |
||||
|
StartTime: time.Now(), |
||||
|
Success: false, |
||||
|
} |
||||
|
|
||||
|
// 检查是否已经在运行备份
|
||||
|
m.mutex.Lock() |
||||
|
if m.running { |
||||
|
m.mutex.Unlock() |
||||
|
result.Message = "备份正在进行中" |
||||
|
return result |
||||
|
} |
||||
|
m.running = true |
||||
|
m.mutex.Unlock() |
||||
|
|
||||
|
defer func() { |
||||
|
m.mutex.Lock() |
||||
|
m.running = false |
||||
|
m.mutex.Unlock() |
||||
|
result.EndTime = time.Now() |
||||
|
}() |
||||
|
|
||||
|
// 获取当前时间戳作为备份标识
|
||||
|
timestamp := result.StartTime.Format("20060102150405") |
||||
|
|
||||
|
// 创建备份目录
|
||||
|
backupDir, err := m.createBackupDir(timestamp) |
||||
|
if err != nil { |
||||
|
result.Message = fmt.Sprintf("创建备份目录失败: %v", err) |
||||
|
return result |
||||
|
} |
||||
|
result.Filename = filepath.Base(backupDir) |
||||
|
|
||||
|
// 获取所有表
|
||||
|
tables, err := m.getTables() |
||||
|
if err != nil { |
||||
|
result.Message = fmt.Sprintf("获取表列表失败: %v", err) |
||||
|
return result |
||||
|
} |
||||
|
|
||||
|
// 备份每个表
|
||||
|
for _, table := range tables { |
||||
|
log.Printf("开始备份表: %s", table) |
||||
|
|
||||
|
// 备份表结构
|
||||
|
schemaFile := filepath.Join(backupDir, table+".sql") |
||||
|
if err := m.backupTableSchema(table, schemaFile); err != nil { |
||||
|
result.Message = fmt.Sprintf("备份表 %s 结构失败: %v", table, err) |
||||
|
return result |
||||
|
} |
||||
|
|
||||
|
// 备份表数据
|
||||
|
dataFile := filepath.Join(backupDir, table+".csv") |
||||
|
if err := m.backupTableData(table, dataFile); err != nil { |
||||
|
result.Message = fmt.Sprintf("备份表 %s 数据失败: %v", table, err) |
||||
|
return result |
||||
|
} |
||||
|
|
||||
|
log.Printf("表 %s 备份完成", table) |
||||
|
} |
||||
|
|
||||
|
// 清理旧备份
|
||||
|
if err := m.cleanOldBackups(); err != nil { |
||||
|
log.Printf("清理旧备份时出错: %v", err) |
||||
|
} |
||||
|
|
||||
|
result.Success = true |
||||
|
result.Message = fmt.Sprintf("数据库 %s 备份成功，存储在: %s", m.config.Database, backupDir) |
||||
|
return result |
||||
|
} |
||||
|
|
||||
|
// StartBackup 启动备份（异步或同步）
|
||||
|
func (m *BackupManager) StartBackup() (BackupResult, chan BackupResult) { |
||||
|
if m.config.Async { |
||||
|
resultChan := make(chan BackupResult, 1) |
||||
|
go func() { |
||||
|
result := m.performBackup() |
||||
|
resultChan <- result |
||||
|
close(resultChan) |
||||
|
}() |
||||
|
return BackupResult{Success: true, Message: "异步备份已启动"}, resultChan |
||||
|
} |
||||
|
|
||||
|
result := m.performBackup() |
||||
|
return result, nil |
||||
|
} |
||||
|
|
||||
|
// Close 关闭数据库连接
|
||||
|
func (m *BackupManager) Close() error { |
||||
|
return m.db.Close() |
||||
} |
} |
||||
|
|
||||
func main() { |
func main() { |
||||
// 解析命令行参数
|
// 解析命令行参数
|
||||
host := flag.String("host", "localhost", "MySQL主机地址") |
host := flag.String("host", "localhost", "MySQL主机地址") |
||||
port := flag.Uint("port", 3306, "MySQL端口") |
port := flag.Int("port", 3306, "MySQL端口") |
||||
user := flag.String("user", "root", "MySQL用户名") |
user := flag.String("user", "root", "MySQL用户名") |
||||
password := flag.String("password", "", "MySQL密码") |
password := flag.String("password", "", "MySQL密码") |
||||
serverID := flag.Uint("server-id", 1001, "客户端服务器ID") |
database := flag.String("db", "", "要备份的数据库名") |
||||
flavor := flag.String("flavor", "mysql", "数据库类型 (mysql或mariadb)") |
backupDir := flag.String("dir", "backups", "备份文件存储目录") |
||||
startFile := flag.String("start-file", "", "开始读取的binlog文件名") |
keepDays := flag.Int("keep", 7, "备份保留天数") |
||||
startPos := flag.Uint("start-pos", 4, "开始读取的位置") |
async := flag.Bool("async", false, "是否异步执行备份") |
||||
|
|
||||
flag.Parse() |
flag.Parse() |
||||
|
|
||||
// 创建canal配置
|
if *database == "" { |
||||
cfg := canal.NewDefaultConfig() |
log.Fatal("必须指定要备份的数据库名 (-db 参数)") |
||||
cfg.Addr = fmt.Sprintf("%s:%d", *host, *port) |
} |
||||
cfg.User = *user |
|
||||
cfg.Password = *password |
|
||||
cfg.ServerID = uint32(*serverID) |
|
||||
cfg.Flavor = *flavor |
|
||||
|
|
||||
// 设置需要监听的数据库和表，默认监听所有
|
// 创建备份配置
|
||||
// cfg.Dump.TableDB = "test_db"
|
config := BackupConfig{ |
||||
// cfg.Dump.Tables = []string{"test_table"}
|
Host: *host, |
||||
|
Port: *port, |
||||
|
User: *user, |
||||
|
Password: *password, |
||||
|
Database: *database, |
||||
|
BackupDir: *backupDir, |
||||
|
KeepDays: *keepDays, |
||||
|
Async: *async, |
||||
|
} |
||||
|
|
||||
// 创建canal实例
|
// 创建备份管理器
|
||||
c, err := canal.NewCanal(cfg) |
manager, err := NewBackupManager(config) |
||||
if err != nil { |
if err != nil { |
||||
log.Fatalf("创建canal实例失败: %v", err) |
log.Fatalf("初始化备份管理器失败: %v", err) |
||||
} |
} |
||||
|
defer manager.Close() |
||||
|
|
||||
// 设置事件处理器
|
// 启动备份
|
||||
c.SetEventHandler(&MyEventHandler{}) |
log.Println("开始数据库备份...") |
||||
|
result, resultChan := manager.StartBackup() |
||||
|
log.Println(result.Message) |
||||
|
|
||||
// 设置起始位置
|
// 如果是异步备份，等待结果
|
||||
var pos mysql.Position |
if *async && resultChan != nil { |
||||
if *startFile != "" { |
log.Println("等待备份完成...") |
||||
pos = mysql.Position{Name: *startFile, Pos: uint32(*startPos)} |
backupResult := <-resultChan |
||||
|
|
||||
|
if backupResult.Success { |
||||
|
log.Printf("备份成功，耗时: %v", backupResult.EndTime.Sub(backupResult.StartTime)) |
||||
|
log.Println(backupResult.Message) |
||||
} else { |
} else { |
||||
// 如果未指定，从当前位置开始
|
log.Printf("备份失败: %s", backupResult.Message) |
||||
pos, err = c.GetMasterPos() |
|
||||
if err != nil { |
|
||||
log.Fatalf("获取主库位置失败: %v", err) |
|
||||
} |
} |
||||
fmt.Printf("从binlog位置 %s:%d 开始读取\n", pos.Name, pos.Pos) |
|
||||
} |
|
||||
|
|
||||
// 开始同步
|
|
||||
err = c.RunFrom(pos) |
|
||||
if err != nil { |
|
||||
log.Fatalf("同步失败: %v", err) |
|
||||
} |
} |
||||
} |
} |
||||
|
|||||
@ -0,0 +1,164 @@ |
|||||
|
package main |
||||
|
|
||||
|
import ( |
||||
|
"flag" |
||||
|
"fmt" |
||||
|
"log" |
||||
|
"strings" |
||||
|
|
||||
|
"github.com/go-mysql-org/go-mysql/canal" |
||||
|
"github.com/go-mysql-org/go-mysql/mysql" |
||||
|
) |
||||
|
|
||||
|
// 自定义事件处理结构体 |
||||
|
type MyEventHandler struct { |
||||
|
canal.DummyEventHandler |
||||
|
} |
||||
|
|
||||
|
// 处理行事件 |
||||
|
func (h *MyEventHandler) OnRow(e *canal.RowsEvent) error { |
||||
|
table := e.Table |
||||
|
sql := "" |
||||
|
|
||||
|
switch e.Action { |
||||
|
case canal.InsertAction: |
||||
|
// 处理插入事件 |
||||
|
columns := make([]string, len(table.Columns)) |
||||
|
for i, col := range table.Columns { |
||||
|
columns[i] = col.Name |
||||
|
} |
||||
|
|
||||
|
values := make([]string, len(e.Rows[0])) |
||||
|
for i, val := range e.Rows[0] { |
||||
|
values[i] = formatValue(val) |
||||
|
} |
||||
|
|
||||
|
sql = fmt.Sprintf("INSERT INTO `%s`.`%s` (%s) VALUES (%s);", |
||||
|
table.Schema, table.Name, |
||||
|
strings.Join(columns, ", "), |
||||
|
strings.Join(values, ", ")) |
||||
|
|
||||
|
case canal.UpdateAction: |
||||
|
// 处理更新事件 |
||||
|
oldRow := e.Rows[0] |
||||
|
newRow := e.Rows[1] |
||||
|
|
||||
|
sets := make([]string, 0) |
||||
|
wheres := make([]string, 0) |
||||
|
|
||||
|
for i, col := range table.Columns { |
||||
|
if oldRow[i] != newRow[i] { |
||||
|
sets = append(sets, fmt.Sprintf("`%s` = %s", col.Name, formatValue(newRow[i]))) |
||||
|
} |
||||
|
wheres = append(wheres, fmt.Sprintf("`%s` = %s", col.Name, formatValue(oldRow[i]))) |
||||
|
} |
||||
|
|
||||
|
sql = fmt.Sprintf("UPDATE `%s`.`%s` SET %s WHERE %s;", |
||||
|
table.Schema, table.Name, |
||||
|
strings.Join(sets, ", "), |
||||
|
strings.Join(wheres, " AND ")) |
||||
|
|
||||
|
case canal.DeleteAction: |
||||
|
// 处理删除事件 |
||||
|
wheres := make([]string, len(table.Columns)) |
||||
|
for i, col := range table.Columns { |
||||
|
wheres[i] = fmt.Sprintf("`%s` = %s", col.Name, formatValue(e.Rows[0][i])) |
||||
|
} |
||||
|
|
||||
|
sql = fmt.Sprintf("DELETE FROM `%s`.`%s` WHERE %s;", |
||||
|
table.Schema, table.Name, |
||||
|
strings.Join(wheres, " AND ")) |
||||
|
} |
||||
|
|
||||
|
if sql != "" { |
||||
|
fmt.Println(sql) |
||||
|
} |
||||
|
|
||||
|
return nil |
||||
|
} |
||||
|
|
||||
|
// string |
||||
|
func (h *MyEventHandler) String() string { return "MyEventHandler" } |
||||
|
|
||||
|
// 处理DDL事件 |
||||
|
// func (h *MyEventHandler) OnDDL(nextPos mysql.Position, queryEvent *replication.QueryEvent) error { |
||||
|
// sql := string(queryEvent.Query) |
||||
|
// if sql != "" { |
||||
|
// fmt.Println(sql + ";") |
||||
|
// } |
||||
|
// return nil |
||||
|
// } |
||||
|
|
||||
|
// 格式化值为SQL表示形式 |
||||
|
func formatValue(value interface{}) string { |
||||
|
if value == nil { |
||||
|
return "NULL" |
||||
|
} |
||||
|
|
||||
|
switch v := value.(type) { |
||||
|
case int, int8, int16, int32, int64, uint, uint8, uint16, uint32, uint64, float32, float64: |
||||
|
return fmt.Sprintf("%v", v) |
||||
|
case []byte: |
||||
|
// 处理二进制数据 |
||||
|
return fmt.Sprintf("X'%x'", v) |
||||
|
case string: |
||||
|
// 转义单引号 |
||||
|
return fmt.Sprintf("'%s'", strings.ReplaceAll(v, "'", "''")) |
||||
|
default: |
||||
|
return fmt.Sprintf("'%v'", v) |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
func main() { |
||||
|
// 解析命令行参数 |
||||
|
host := flag.String("host", "localhost", "MySQL主机地址") |
||||
|
port := flag.Uint("port", 3306, "MySQL端口") |
||||
|
user := flag.String("user", "root", "MySQL用户名") |
||||
|
password := flag.String("password", "", "MySQL密码") |
||||
|
serverID := flag.Uint("server-id", 1001, "客户端服务器ID") |
||||
|
flavor := flag.String("flavor", "mysql", "数据库类型 (mysql或mariadb)") |
||||
|
startFile := flag.String("start-file", "", "开始读取的binlog文件名") |
||||
|
startPos := flag.Uint("start-pos", 4, "开始读取的位置") |
||||
|
|
||||
|
flag.Parse() |
||||
|
|
||||
|
// 创建canal配置 |
||||
|
cfg := canal.NewDefaultConfig() |
||||
|
cfg.Addr = fmt.Sprintf("%s:%d", *host, *port) |
||||
|
cfg.User = *user |
||||
|
cfg.Password = *password |
||||
|
cfg.ServerID = uint32(*serverID) |
||||
|
cfg.Flavor = *flavor |
||||
|
|
||||
|
// 设置需要监听的数据库和表，默认监听所有 |
||||
|
// cfg.Dump.TableDB = "test_db" |
||||
|
// cfg.Dump.Tables = []string{"test_table"} |
||||
|
|
||||
|
// 创建canal实例 |
||||
|
c, err := canal.NewCanal(cfg) |
||||
|
if err != nil { |
||||
|
log.Fatalf("创建canal实例失败: %v", err) |
||||
|
} |
||||
|
|
||||
|
// 设置事件处理器 |
||||
|
c.SetEventHandler(&MyEventHandler{}) |
||||
|
|
||||
|
// 设置起始位置 |
||||
|
var pos mysql.Position |
||||
|
if *startFile != "" { |
||||
|
pos = mysql.Position{Name: *startFile, Pos: uint32(*startPos)} |
||||
|
} else { |
||||
|
// 如果未指定，从当前位置开始 |
||||
|
pos, err = c.GetMasterPos() |
||||
|
if err != nil { |
||||
|
log.Fatalf("获取主库位置失败: %v", err) |
||||
|
} |
||||
|
fmt.Printf("从binlog位置 %s:%d 开始读取\n", pos.Name, pos.Pos) |
||||
|
} |
||||
|
|
||||
|
// 开始同步 |
||||
|
err = c.RunFrom(pos) |
||||
|
if err != nil { |
||||
|
log.Fatalf("同步失败: %v", err) |
||||
|
} |
||||
|
} |
||||
Binary file not shown.
@ -0,0 +1,9 @@ |
|||||
|
安装依赖 |
||||
|
go get github.com/go-sql-driver/mysql |
||||
|
|
||||
|
|
||||
|
# 同步备份 |
||||
|
./mysql_backup -host=localhost -port=3306 -user=your_user -password=your_pass -db=your_database |
||||
|
|
||||
|
# 异步备份 |
||||
|
./mysql_backup -host=localhost -port=3306 -user=your_user -password=your_pass -db=your_database -async=true |
||||
Loading…
Reference in new issue