Browse Source

update lock on trans

alpha
yedf2 3 years ago
parent
commit
0f981e8572
  1. 23
      dtmsvr/storage/sql/sql.go

23
dtmsvr/storage/sql/sql.go

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

Loading…
Cancel
Save