dbsync

README

dbsync

这是用来实现两个MySQL数据库中的具有相同字段表的增量同步。

• 支持按照某种格式增量获取表中的待同步数据。
• 支持按照列名称向数据库的表中批量插入待同步的数据。

安装

go get -u gitee.com/cuckoopark/dbsync

数据库配置

在每一张需要同步的表中，应该有一个update_time更新时间的非空字段(名字可以不一样，但是类型必须是时间相关类型)，用来按照更新时间获取最新的更新数据。

这个字段需要在数据更新时，自动更新为当前时间戳，用于记录数据更新的时间。

例如update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP的字段设置。

获取增量更新数据

可以批量获取一张表的最新更新的数据，方法如下：

func DoFetch(db SQLCommon, tableName string, options FetchOptions) (FetchResult, error)

其中参数说明：

• db：数据库操作句柄。
• tableName：表名称。
• options：获取时的配置信息，FetchOptions格式如下所示：
  • IgnoreFields：需要忽略的列名称，获取数据的结果不包含该列。
  • PageNumber：分页获取增量的页码，从1开始。
  • PageSize：分页获取增量的页大小，判断分页是否结束，只需要判断获取结果的数量是否小于页大小即可。
  • UpdateTimeFieldName：更新时间(即上面说明的update_time)所在列的列名称。
  • LastUpdateTime：上次更新的时间戳，大于这个时间戳开始查询，如果为0，则表示查询全部数据。
  • WhereSqlStmt：自定义SQL查询语句的Where子句，与更新时间的条件([UpdateTimeFieldName] > ?)是AND的关系。
  • WhereSqlArgs：自定义SQL查询语句的Where子句的参数列表。

获取的结果，是FetchResult格式的结构体：

• columns：列名称列表。
• column_types：列的数据类型列表，与列名称列表一一对应。
• data：最新更新的数据，二维数组，每一行是一条数据，里面的值与列名称是一一对应关系。

注：时间类型的列，获取的结果time.Time会被转换为时间戳传递，用于节省数据长度。

插入单条更新数据

接口为：

func DoUpdateOne(db SQLCommon, tableName string, data []interface{}, options UpdateOptions) error

其中参数说明：

• db：数据库操作句柄。
• tableName：表名称。
• data：一条数据，里面的值与options.Columns一一对应。
• options：插入时的配置信息，UpdateOptions格式如下所示：
  • Columns：列名称列表，参照DoFetch返回的结果。
  • ColumnTypes：列的数据类型列表，参照DoFetch返回的结果。
  • FixedFields：固定的插入列，因为在DoFetch中会配置忽略一些列，所以这里可以给这些列设置值。
  • UniqueFields：唯一键或主键的列名称列表，在插入失败时，更新操作不更新唯一键或主键。

该接口无返回值。

插入或更新会调用INSERT INTO ... VALUES (...) ON DUPLICATE KEY UPDATE ...这种SQL语句来执行，不采用REPLACE INTO的原因是它的更新会先删除旧数据，再插入新数据，可能导致一些忽略的字段被修改。

批量插入更新数据

接口为：

func DoUpdate(db SQLCommon, tableName string, data [][]interface{}, options UpdateMultiOptions) error

相比较插入单条更新数据的接口，只是数据变成了二维数组，options里面多了一个配置项：

• BatchCount：一次批量插入的条数，用于加快插入或更新的执行速度。

Documentation

Index

• func DoUpdate(db SQLCommon, tableName string, data [][]interface{}, ...) (err error)
• func DoUpdateOne(db SQLCommon, tableName string, data []interface{}, options UpdateOptions) (err error)
• type FetchOptions
• type FetchResult
  • func DoFetch(db SQLCommon, tableName string, options FetchOptions) (rsp FetchResult, err error)
• type SQLCommon
• type UpdateMultiOptions
• type UpdateOptions

Functions

func DoUpdate

func DoUpdate(db SQLCommon, tableName string, data [][]interface{}, options UpdateMultiOptions) (err error)

插入多条增量更新的数据

func DoUpdateOne

func DoUpdateOne(db SQLCommon, tableName string, data []interface{}, options UpdateOptions) (err error)

插入单条数据

Types

type FetchOptions

type FetchOptions struct {
	IgnoreFields        []string      // 忽略的列名称
	PageNumber          int           // 分页获取增量的页码，从1开始
	PageSize            int           // 分页获取增量的页大小
	UpdateTimeFieldName string        // 更新时间所在列的列名称
	LastUpdateTime      int64         // 从哪个时间戳开始查询，这是大于的关系
	WhereSqlStmt        string        // 自定义SQL查询语句的Where子句
	WhereSqlArgs        []interface{} // 自定义SQL查询语句的Where子句的参数列表
}

获取增量数据时的配置信息

type FetchResult

type FetchResult struct {
	Columns     []string        `json:"columns"`      // 列名称
	ColumnTypes []string        `json:"column_types"` // 列的数据类型
	Data        [][]interface{} `json:"data"`         // 待同步的数据，每一行是一条数据，与列名称一一对应
}

获取增量数据的返回结果

func DoFetch

func DoFetch(db SQLCommon, tableName string, options FetchOptions) (rsp FetchResult, err error)

获取增量更新的数据

type SQLCommon

type SQLCommon interface {
	Exec(query string, args ...interface{}) (sql.Result, error)
	Prepare(query string) (*sql.Stmt, error)
	Query(query string, args ...interface{}) (*sql.Rows, error)
	QueryRow(query string, args ...interface{}) *sql.Row
}

原始SQL的接口

type UpdateMultiOptions

type UpdateMultiOptions struct {
	UpdateOptions
	BatchCount int // 一次批量插入的条数，用于加快执行速度
}

插入多条数据时的配置信息

type UpdateOptions

type UpdateOptions struct {
	Columns      []string               // 列名称
	ColumnTypes  []string               // 列类型
	FixedFields  map[string]interface{} // 固定的插入列
	UniqueFields []string               // 唯一键或主键的列名称列表
}

插入数据时的配置信息