mirror of
https://github.com/casdoor/casdoor.git
synced 2025-07-02 19:40:19 +08:00
feat: add basic MySQL sync functionality (#1575)
* feat: add sync module to bi-sync mysql * feat: fix the delay problem * feat: fix go mod * feat: fix the varchar(100) parse error * fix: fix go.mod space inconsistency * fix: fix go.mod space inconsistency * fix: use sql builder instead of concatenation * fix: remove serverId * fix: fix file is not `gofumpt`-ed (gofumpt) error
This commit is contained in:
longxu0509
187
sync/bi_sync_mysql.go
Normal file
187
sync/bi_sync_mysql.go
Normal file
@ -0,0 +1,187 @@
|
||||
// Copyright 2023 The Casdoor Authors. All Rights Reserved.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package sync
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"github.com/go-mysql-org/go-mysql/canal"
|
||||
"github.com/go-mysql-org/go-mysql/mysql"
|
||||
"github.com/go-mysql-org/go-mysql/replication"
|
||||
"github.com/siddontang/go-log/log"
|
||||
"github.com/xorm-io/xorm"
|
||||
)
|
||||
|
||||
var (
|
||||
dataSourceName1 string
|
||||
dataSourceName2 string
|
||||
engin1 *xorm.Engine
|
||||
engin2 *xorm.Engine
|
||||
)
|
||||
|
||||
func InitConfig() *canal.Config {
|
||||
// init dataSource
|
||||
dataSourceName1 = fmt.Sprintf("%s:%s@tcp(%s:%d)/%s", username1, password1, host1, port1, database1)
|
||||
dataSourceName2 = fmt.Sprintf("%s:%s@tcp(%s:%d)/%s", username2, password2, host2, port2, database2)
|
||||
|
||||
// create engine
|
||||
engin1, _ = CreateEngine(dataSourceName1)
|
||||
engin2, _ = CreateEngine(dataSourceName2)
|
||||
log.Info("init engine success…")
|
||||
|
||||
// config canal
|
||||
cfg := canal.NewDefaultConfig()
|
||||
cfg.Addr = fmt.Sprintf("%s:%d", host1, port1)
|
||||
cfg.Password = password1
|
||||
cfg.User = username1
|
||||
// We only care table in database1
|
||||
cfg.Dump.TableDB = database1
|
||||
// cfg.Dump.Tables = []string{"user"}
|
||||
log.Info("config canal success…")
|
||||
return cfg
|
||||
}
|
||||
|
||||
func StartBinlogSync() error {
|
||||
// init config
|
||||
config := InitConfig()
|
||||
|
||||
c, err := canal.NewCanal(config)
|
||||
pos, err := c.GetMasterPos()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Register a handler to handle RowsEvent
|
||||
c.SetEventHandler(&MyEventHandler{})
|
||||
|
||||
// Start canal
|
||||
c.RunFrom(pos)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
type MyEventHandler struct {
|
||||
canal.DummyEventHandler
|
||||
}
|
||||
|
||||
func OnTableChanged(header *replication.EventHeader, schema string, table string) error {
|
||||
log.Info("table changed event")
|
||||
return nil
|
||||
}
|
||||
|
||||
func (h *MyEventHandler) onDDL(header *replication.EventHeader, nextPos mysql.Position, queryEvent *replication.QueryEvent) error {
|
||||
log.Info("into DDL event")
|
||||
return nil
|
||||
}
|
||||
|
||||
func (h *MyEventHandler) OnRow(e *canal.RowsEvent) error {
|
||||
length := len(e.Table.Columns)
|
||||
columnNames := make([]string, length)
|
||||
oldColumnValue := make([]interface{}, length)
|
||||
newColumnValue := make([]interface{}, length)
|
||||
isChar := make([]bool, len(e.Table.Columns))
|
||||
|
||||
for i, col := range e.Table.Columns {
|
||||
columnNames[i] = col.Name
|
||||
if col.Type <= 2 {
|
||||
isChar[i] = false
|
||||
} else {
|
||||
isChar[i] = true
|
||||
}
|
||||
}
|
||||
|
||||
switch e.Action {
|
||||
case canal.UpdateAction:
|
||||
for i, row := range e.Rows {
|
||||
for j, item := range row {
|
||||
if i%2 == 0 {
|
||||
if isChar[j] == true {
|
||||
oldColumnValue[j] = fmt.Sprintf("%s", item)
|
||||
} else {
|
||||
oldColumnValue[j] = fmt.Sprintf("%d", item)
|
||||
}
|
||||
} else {
|
||||
if isChar[j] == true {
|
||||
newColumnValue[j] = fmt.Sprintf("%s", item)
|
||||
} else {
|
||||
newColumnValue[j] = fmt.Sprintf("%d", item)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if i%2 == 1 {
|
||||
updateSql, args, err := GetUpdateSql(e.Table.Schema, e.Table.Name, columnNames, newColumnValue, oldColumnValue)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
res, err := engin2.DB().Exec(updateSql, args...)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
log.Info(updateSql, args, res)
|
||||
}
|
||||
}
|
||||
case canal.DeleteAction:
|
||||
for _, row := range e.Rows {
|
||||
for j, item := range row {
|
||||
if isChar[j] == true {
|
||||
oldColumnValue[j] = fmt.Sprintf("%s", item)
|
||||
} else {
|
||||
oldColumnValue[j] = fmt.Sprintf("%d", item)
|
||||
}
|
||||
}
|
||||
|
||||
deleteSql, args, err := GetDeleteSql(e.Table.Schema, e.Table.Name, columnNames, oldColumnValue)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
res, err := engin2.DB().Exec(deleteSql, args...)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
log.Info(deleteSql, args, res)
|
||||
}
|
||||
case canal.InsertAction:
|
||||
for _, row := range e.Rows {
|
||||
for j, item := range row {
|
||||
if isChar[j] == true {
|
||||
newColumnValue[j] = fmt.Sprintf("%s", item)
|
||||
} else {
|
||||
newColumnValue[j] = fmt.Sprintf("%d", item)
|
||||
}
|
||||
}
|
||||
|
||||
insertSql, args, err := GetInsertSql(e.Table.Schema, e.Table.Name, columnNames, newColumnValue)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
res, err := engin2.DB().Exec(insertSql, args...)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
log.Info(insertSql, args, res)
|
||||
}
|
||||
default:
|
||||
log.Infof("%v", e.String())
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (h *MyEventHandler) String() string {
|
||||
return "MyEventHandler"
|
||||
}
|
||||
17
sync/conf.go
Normal file
17
sync/conf.go
Normal file
@ -0,0 +1,17 @@
|
||||
package sync
|
||||
|
||||
var (
|
||||
host1 = "127.0.0.1"
|
||||
port1 = 3306
|
||||
username1 = "root"
|
||||
password1 = "123456"
|
||||
database1 = "db"
|
||||
)
|
||||
|
||||
var (
|
||||
host2 = "127.0.0.1"
|
||||
port2 = 3306
|
||||
username2 = "root"
|
||||
password2 = "123456"
|
||||
database2 = "db"
|
||||
)
|
||||
72
sync/utils.go
Normal file
72
sync/utils.go
Normal file
@ -0,0 +1,72 @@
|
||||
// Copyright 2023 The Casdoor Authors. All Rights Reserved.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package sync
|
||||
|
||||
import (
|
||||
"log"
|
||||
|
||||
"github.com/Masterminds/squirrel"
|
||||
"github.com/xorm-io/xorm"
|
||||
)
|
||||
|
||||
func GetUpdateSql(schemaName string, tableName string, columnNames []string, newColumnVal []interface{}, oldColumnVal []interface{}) (string, []interface{}, error) {
|
||||
updateSql := squirrel.Update(schemaName + "." + tableName)
|
||||
for i, columnName := range columnNames {
|
||||
updateSql = updateSql.Set(columnName, newColumnVal[i])
|
||||
}
|
||||
|
||||
for i, columnName := range columnNames {
|
||||
updateSql = updateSql.Where(squirrel.Eq{columnName: oldColumnVal[i]})
|
||||
}
|
||||
|
||||
sql, args, err := updateSql.ToSql()
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
|
||||
return sql, args, nil
|
||||
}
|
||||
|
||||
func GetInsertSql(schemaName string, tableName string, columnNames []string, columnValue []interface{}) (string, []interface{}, error) {
|
||||
insertSql := squirrel.Insert(schemaName + "." + tableName).Columns(columnNames...).Values(columnValue...)
|
||||
|
||||
return insertSql.ToSql()
|
||||
}
|
||||
|
||||
func GetDeleteSql(schemaName string, tableName string, columnNames []string, columnValue []interface{}) (string, []interface{}, error) {
|
||||
deleteSql := squirrel.Delete(schemaName + "." + tableName)
|
||||
|
||||
for i, columnName := range columnNames {
|
||||
deleteSql = deleteSql.Where(squirrel.Eq{columnName: columnValue[i]})
|
||||
}
|
||||
|
||||
return deleteSql.ToSql()
|
||||
}
|
||||
|
||||
func CreateEngine(dataSourceName string) (*xorm.Engine, error) {
|
||||
engine, err := xorm.NewEngine("mysql", dataSourceName)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// ping mysql
|
||||
err = engine.Ping()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
log.Println("mysql connection success……")
|
||||
return engine, nil
|
||||
}
|
||||
Reference in New Issue
Block a user