diff --git a/dtmsvr/storage/sql/sql.go b/dtmsvr/storage/sql/sql.go index cf8583c..8e444bc 100644 --- a/dtmsvr/storage/sql/sql.go +++ b/dtmsvr/storage/sql/sql.go @@ -8,6 +8,8 @@ package sql import ( + "database/sql" + "errors" "fmt" "math" "time" @@ -159,23 +161,26 @@ func (s *Store) LockOneGlobalTrans(expireIn time.Duration) *storage.TransGlobalS db := dbGet() owner := shortuuid.New() nextCronTime := getTimeStr(int64(expireIn / time.Second)) - where := map[string]string{ - dtmimp.DBTypeMysql: fmt.Sprintf(`next_cron_time < '%s' and status in ('prepared', 'aborting', 'submitted') limit 1`, nextCronTime), - dtmimp.DBTypePostgres: fmt.Sprintf(`id in (select id from trans_global where next_cron_time < '%s' and status in ('prepared', 'aborting', 'submitted') limit 1 )`, nextCronTime), + where := fmt.Sprintf(`next_cron_time < '%s' and status in ('prepared', 'aborting', 'submitted')`, nextCronTime) + + order := map[string]string{ + dtmimp.DBTypeMysql: `order by rand()`, + dtmimp.DBTypePostgres: `order by random()`, }[conf.Store.Driver] - ssql := fmt.Sprintf(`select count(1) from trans_global where %s`, where) - var cnt int64 - err := db.ToSQLDB().QueryRow(ssql).Scan(&cnt) - dtmimp.PanicIf(err != nil, err) - if cnt == 0 { + ssql := fmt.Sprintf(`select id from trans_global where %s %s limit 1`, where, order) + var id int64 + err := db.ToSQLDB().QueryRow(ssql).Scan(&id) + if errors.Is(err, sql.ErrNoRows) { return nil } + dtmimp.PanicIf(err != nil, err) - sql := fmt.Sprintf(`UPDATE trans_global SET update_time='%s',next_cron_time='%s', owner='%s' WHERE %s`, + sql := fmt.Sprintf(`UPDATE trans_global SET update_time='%s',next_cron_time='%s', owner='%s' WHERE id=%d and %s`, getTimeStr(0), getTimeStr(conf.RetryInterval), owner, + id, where) affected, err := dtmimp.DBExec(conf.Store.Driver, db.ToSQLDB(), sql)