从内存到sql的upsert

业务的upsert

在写业务时，大家一开始都会以顺序流程的方式开始着手写代码，CR时再看代码，会有不一样的感觉。

1. 需求描述

现有一张数据库表，表字段结构如下：

字段名称	类型	描述
`uuid`	string	数据的唯一键
`data`	string	业务数据
`version`	int	请求时业务的版本好
`checksum`	string	`data`内容的哈希值
`updated_at`	time.Time	更新时间

需求：若干个请求方会调用接口并带有该次请求的版本号来更新数据

更新规则：

数据的checksum不一致时，更新数据
当version版本号为0时，默认要更新数据；当version版本号大于db中版本号时，更新数据

2. 需求实现

该部分会根据业务场景不断演化

2.1 简单实现

一开始看到这个需求，很简单啊，有手就行！

type BusinessData struct {ID        int64     `gorm:"column:id;type:int(11);primary_key;AUTO_INCREMENT"`UUID      string    `gorm:"column:uuid;type:varchar(255);uniqueIndex:uniq_uuid"`Checksum  string    `gorm:"column:checksum;type:varchar(255)"`Version   int     `gorm:"column:version;type:int(11);default:0"`CreatedAt time.Time `gorm:"column:created_at;type:datetime(3);default:CURRENT_TIMESTAMP(3)"`UpdatedAt time.Time `gorm:"column:updated_at;type:datetime(3);update:CURRENT_TIMESTAMP(3)"`
}func GetDbData(uuids []string) ([]*BusinessData, error) {var (res []*BusinessDataerr error)dsn := "user:password@tcp(localhost:3306)/database_name?charset=utf8mb4&parseTime=True&loc=Local"db, oErr := gorm.Open(mysql.Open(dsn), &gorm.Config{})if oErr != nil {fmt.Println(oErr)return nil, oErr}err = db.Table("business").Where("uuid IN ?", uuids).Find(&res).Errorif err != nil {fmt.Println(err)return nil, err}return res, nil
}func UpsertData(data []*BusinessData) error {dsn := "user:password@tcp(localhost:3306)/database_name?charset=utf8mb4&parseTime=True&loc=Local"db, oErr := gorm.Open(mysql.Open(dsn), &gorm.Config{})if oErr != nil {fmt.Println(oErr)return oErr}err := db.Table("business").Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "uuid"},},DoUpdates: clause.AssignmentColumns([]string{"data", "version", "checksum", "updated_at"}),}).Create(data).Errorif err != nil {return err}return nil
}func main() {var (datas            []*BusinessDatatoUpsertDataList []*BusinessDatauuids            []string)for _, data := range datas {uuids = append(uuids, data.UUID)}dbDatas, err := GetDbData(uuids)if err != nil {fmt.Println(err)return}existDataMap := make(map[string]*BusinessData)for _, dbData := range dbDatas {existDataMap[dbData.UUID] = dbData}for _, data := range datas {if _, ok := existDataMap[data.UUID]; !ok {toUpsertDataList = append(toUpsertDataList, data)continue}if data.Checksum == existDataMap[data.UUID].Checksum {continue}if data.Version != 0 && data.Version < existDataMap[data.UUID].Version {continue}toUpsertDataList = append(toUpsertDataList, data)}err = UpsertData(toUpsertDataList)if err != nil {fmt.Println(err)}
}

代码很简单，读取db中的数据到内存，根据checksum和version的要求进行过滤，符合条件的data便进行更新/插入

2.2 进阶实现

2.2.1 思考1

将db中的数据读到内存后进行uuid匹配，是不是跟upsert时利用唯一键冲突更新数据重复了呢？？？

2.2.2 思考2

难道说利用唯一键冲突就是满足更新的所有条件吗？

在db进行update时，我们使用where来过滤数据，那么在upsert时是不是也可以通过什么手段过滤数据呢？？？

将两个简单思考进行实现（这里只展示UpsertData方法）

err := db.Table("business").Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "uuid"}},DoUpdates: clause.Assignments(map[string]interface{}{"checksum":   gorm.Expr("IF((version <= VALUES(version) OR VALUES(version) = 0) AND checksum != VALUES(checksum), VALUES(checksum), checksum)"),"updated_at": gorm.Expr("IF((version <= VALUES(version) OR VALUES(version) = 0) AND checksum != VALUES(checksum), VALUES(updated_at), updated_at)"),"version":    gorm.Expr("IF((version <= VALUES(version) OR VALUES(version) = 0) AND checksum != VALUES(checksum), VALUES(version), version)"),}),}).Create(data).Error

是不是大功告成！准备完结撒花🎉了

ON DUPLICATE KEY UPDATE 语序会有影响结果

也就是说，按顺序执行时，先更新checksum（正常），再更新updated_at和version（异常）

异常原因： checksum更新后， IF条件中checksum == VALUES(checksum) 成立

2.2.3 思考3

那这样的话，我改变语序不就好了吗，直接就是完结撒花🎉！

那假如，我有很多个字段，我一个一个调试看看该字段会不会影响其他字段太浪费时间了， 那有没有什么办法能够记录快照状态呢

对，快照！

我们把当前的条件字段变成快照状态，用快照进行条件判断，就不用担心语序问题了

开始实现：

	condition := "(version <= VALUES(version) OR VALUES(version) = 0) AND checksum != VALUES(checksum)"err := db.Table("business").Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "uuid"},},DoUpdates: []clause.Assignment{{Column: clause.Column{Name: "version"},Value: gorm.Expr(fmt.Sprintf("IF(@should_update := %s, VALUES(%s), %s)", condition, "version", "version")),},{Column: clause.Column{Name: "checksum"},Value: gorm.Expr(fmt.Sprintf("IF(@should_update := %s, VALUES(%s), %s)", condition, "checksum", "checksum")),},{Column: clause.Column{Name: "updated_at"},Value: gorm.Expr(fmt.Sprintf("IF(@should_update := %s, VALUES(%s), %s)", condition,  "updated_at",  "updated_at")),},},}).Create(data).Error

美化一下

// UpsertExprs _
func UpsertExprs(condition string, columns []string) clause.Set {if len(columns) < 1 {return clause.Set{}}var assignments []clause.Assignmentfor i, column := range columns {if i == 0 {assignments = append(assignments, clause.Assignment{Column: clause.Column{Name: column},Value: gorm.Expr(fmt.Sprintf("IF(@should_update := %s, VALUES(%s), %s)", condition, column, column)),})continue}assignments = append(assignments, clause.Assignment{Column: clause.Column{Name: column},Value:  gorm.Expr(fmt.Sprintf("IF(@should_update, VALUES(%s), %s)", column, column)),})}return assignments
}func UpsertData(data []*BusinessData) error {dsn := "user:password@tcp(localhost:3306)/database_name?charset=utf8mb4&parseTime=True&loc=Local"db, oErr := gorm.Open(mysql.Open(dsn), &gorm.Config{})if oErr != nil {fmt.Println(oErr)return oErr}condition := "(version <= VALUES(version) OR VALUES(version) = 0) AND checksum != VALUES(checksum)"err := db.Table("business").Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "uuid"},},DoUpdates: UpsertExprs(condition, []string{"version", "checksum", "updated_at"}),}).Create(data).Errorif err != nil {return err}return nil
}