This commit is contained in:
bryanqiu 2022-11-23 16:02:30 +08:00
parent ea81f5a3a6
commit 6f7e10aecb
2 changed files with 99 additions and 58 deletions

2
go.sum Normal file
View File

@ -0,0 +1,2 @@
github.com/gammazero/deque v0.2.1 h1:qSdsbG6pgp6nL7A0+K/B7s12mcCY/5l5SIUpMOl+dC0=
github.com/gammazero/deque v0.2.1/go.mod h1:LFroj8x4cMYCukHJDbxFCkT+r9AndaJnFMuZDV34tuU=

155
uid.go
View File

@ -8,12 +8,14 @@
package uid
import (
"errors"
"fmt"
"sync"
"sync/atomic"
"time"
"github.com/gammazero/deque"
"qoobing.com/gomod/log"
)
// IdCreator
@ -24,35 +26,42 @@ type IdCreator interface {
// IdCreatorHelper
type IdCreatorHelper interface {
// CreatePrefix create id prefix
CreatePrefix(parts ...interface{})
CreatePrefix(parts ...interface{}) string
// CreateRange create id range from database table t_id.
CreateRange(prefix string) (start, length int64, err error)
CreateRange(prefix string) (start int64, length int, err error)
// CreateMixedId create finnal transaction id
CreateMixedId(prefix string, id int64, parts ...interface{}) string
}
// idCreater
type idCreater struct {
lock sync.Mutex
Type string
Ranges map[string]*deque.Deque
Helper IdCreateHelper
type rangeQueue struct {
deque.Deque[*Range]
}
// NewIdCreater to new a id creator by helper.
func NewIdCreater(typ string, helper IdCreatorHelper) IdCreater {
// idCreator
type idCreator struct {
lock sync.Mutex
Type string
Ranges map[string]*rangeQueue
RangesLocks map[string]*sync.Mutex
Helper IdCreatorHelper
}
// NewIdCreator to new a id creator by helper.
func NewIdCreator(typ string, helper IdCreatorHelper) IdCreator {
// Step 1. basic value
var idg = &idCreater{
lock: sync.Mutex,
Type: typ,
Ranges: map[string]*deque.Deque{},
Helper: helper,
var idg = &idCreator{
lock: sync.Mutex{},
Type: typ,
Ranges: map[string]*rangeQueue{},
RangesLocks: map[string]*sync.Mutex{},
Helper: helper,
}
// Step 2. generate the first range.
if prefix, err = idg.createDefaultPrefix(); err == nil {
if prefix, err := idg.createDefaultPrefix(); err == nil {
if err := idg.createNewRange(prefix); err != nil {
log.Errorf("generate the first range failed: %s", err)
panic("NewIdCreator error")
}
}
@ -61,20 +70,20 @@ func NewIdCreater(typ string, helper IdCreatorHelper) IdCreater {
for {
for prefix, queue := range idg.Ranges {
var qlen = queue.Len()
var timeoutMs = 1
var timeoutMs = 1 * time.Millisecond
if queueLen <= 0 {
timeoutMs = 1
} else if queueLen <= 2 {
timeoutMs = 50
if qlen <= 0 {
timeoutMs = 1 * time.Millisecond
} else if qlen <= 2 {
timeoutMs = 50 * time.Millisecond
} else {
timeoutMs = -1000
timeoutMs = -1000 * time.Millisecond
}
if timeoutMs > 0 {
if int(timeoutMs) > 0 {
idg.createNewRange(prefix)
time.Sleep(timeoutMs * time.Millisecond)
} else if timeoutMs < 0 {
} else if int(timeoutMs) < 0 {
time.Sleep((-timeoutMs) * time.Millisecond)
}
}
@ -85,47 +94,50 @@ func NewIdCreater(typ string, helper IdCreatorHelper) IdCreater {
return idg
}
func (idg *idCreater) GetId(idparts ...interface{}) string {
var prefix = idg.Creator.CreatePrefix(idparts...)
func (idg *idCreator) GetId(idparts ...interface{}) string {
var prefix = idg.Helper.CreatePrefix(idparts...)
for {
var id int64 = 0
var cur = idg.getCurrentRange(prefix)
var currange = idg.getCurrentRange(prefix)
log.Infof("success get current range:(%s, %d)", prefix, currange.Start)
// Step 1. Optimistic[no lock] to add id in current range.
for {
id = atomic.LoadInt64(&cur.Next)
if id > cur.End {
id = atomic.LoadInt64(&currange.Next)
if id > currange.End {
//next range
break
}
if atomic.CompareAndSwapInt64(&cur.Next, id, id+1) {
if atomic.CompareAndSwapInt64(&currange.Next, id, id+1) {
//success
break
}
}
// Step 2. No lock add failed, try add id in next range.
if id > cur.End {
if id > currange.End {
//next range
idg.turnToNextRange(cur)
log.Infof("failed get id from current range, try next")
idg.turnToNextRange(currange)
continue
}
// Step 3. Success, return mixed id.
return idg.MixId(cur, id)
log.Infof("success get id(not miexed):(%s, %d)", prefix, id)
return idg.Helper.CreateMixedId(prefix, id, idparts...)
}
}
// Range Range define the range [start, end] of ids.
type Range struct {
Id int64 // id
Next int64 // next id
Status int // 0:invlid; 1:ok; 2:used
Prefix string // prefix
Start int64 // begin id of this range(include)
End int64 // end id of this range(include)
Length int64 // lenght of this range
Length int // lenght of this range
}
func (idg *idCreater) createDefaultPrefix() (prefix string, err error) {
func (idg *idCreator) createDefaultPrefix() (prefix string, err error) {
defer func() {
if r := recover(); r != nil {
err = fmt.Errorf("default prefix not surport by user")
@ -137,76 +149,103 @@ func (idg *idCreater) createDefaultPrefix() (prefix string, err error) {
return prefix, err
}
func (idg *idCreater) createNewRange(prefix string) error {
func (idg *idCreator) removeOldRange(prefix string) error {
if lock, ok := idg.RangesLocks[prefix]; ok {
lock.Lock()
defer lock.Unlock()
queue := idg.Ranges[prefix]
for {
r := queue.Front()
if r.Status == 1 {
break
} else if r.Status == 2 {
queue.PopFront()
} else {
panic("invalid status, maybe somethong bug")
}
}
}
return nil
}
func (idg *idCreator) createNewRange(prefix string) error {
// Step 1. create range from db or something else.
start, length, err := idg.Creator.CreateRange(idg.Type, prefix)
start, length, err := idg.Helper.CreateRange(prefix)
if err != nil {
return fmt.Errorf(
"idCreater '%s' generate range err:'%s'",
str := fmt.Sprintf(
"idCreator '%s' generate range err:'%s'",
idg.Type, err.Error())
log.Errorf("%s", str)
return errors.New(str)
}
// Step 2. lock generator.
idg.lock.Lock()
defer idg.lock.Unlock()
lock, ok := idg.RangesLocks[prefix]
if !ok {
lock = &sync.Mutex{}
queue = deque.New[*Range]()
idg.Ranges[prefix] = queue
idg.RangesLocks[prefix] = lock
if _, ok := idg.RangesLocks[prefix]; !ok {
idg.Ranges[prefix] = &rangeQueue{Deque: *deque.New[*Range]()}
idg.RangesLocks[prefix] = &sync.Mutex{}
}
var lock = idg.RangesLocks[prefix]
var queue = idg.Ranges[prefix]
lock.Lock()
defer lock.Unlock()
// Step 3. push to ranges queue.
queue.PushBack(
&Range{
Id: 0,
Next: start,
Status: 1,
Prefix: prefix,
Start: start,
End: start + length - 1,
End: start + int64(length) - 1,
Length: length,
})
return nil
}
func (idg *idCreater) getCurrentRange(prefix string) (r *Range) {
func (idg *idCreator) getCurrentRange(prefix string) (r *Range) {
// Step 0. get ranges queue by prefix.
var queue, ok = idg.Ranges[prefix]
if !ok {
idg.createNewRange(prefix)
}
var lock, ok = idg.RangesLocks[prefix]
if !ok {
var lock, okk = idg.RangesLocks[prefix]
if !okk {
panic("unreachable code: lock is not exist")
}
lock.Lock()
defer lock.Unlock()
// Step 1. try read from queue directly.
lock.Lock()
if queue.Len() >= 1 {
return queue.Front()
r = queue.Front()
lock.Unlock()
return r
}
lock.Unlock()
// Step 2. try create a new range & try again.
idg.createNewRange(prefix)
lock.Lock()
defer lock.Unlock()
if queue.Len() >= 1 {
return queue.Front()
r = queue.Front()
return r
}
panic("get current range failed, too many pressure???")
}
func (idg *idCreater) turnToNextRange(r *Range) {
func (idg *idCreator) turnToNextRange(r *Range) {
// We donot lock the range, because there is no too much
// side effect even if 'createNewRange' execute twice.
switch r.Status {
case 1:
r.Status = 2
idg.removeOldRange(r.Prefix)
idg.createNewRange(r.Prefix)
case 2:
//have turn by others, we do nothing.
log.Debugf("have turn by others, we do nothing.")
default:
panic("invalid status, maybe somethong bug")
}
@ -218,13 +257,13 @@ func (idg *idCreater) turnToNextRange(r *Range) {
// **DONOT** use in product enviroment.
type demo struct{}
func (d *demo) GetRange(typ, prefix string) (start, length int64, err error) {
func (d *demo) CreateRange(typ, prefix string) (start int64, length int, err error) {
var curdate string = time.Now().Format("20060102")
if prefix != curdate {
return 0, 0, fmt.Errorf("demo just support prefix is current date")
}
start = uint64((time.Now().Unix() % 86400) * 10000)
start = int64((time.Now().Unix() % 86400) * 10000)
length = 10000
return start, length, nil
}