详解如何解决golang定时器引发的id重复问题

目录

• 问题描述
• 简述根因
• 问题代码
• 根因分析
• 验证

问题描述

线上服务日志中突然出现很多主键冲突的异常,而这个主键是一个int64的id,这个id的生成依赖了秒级时间戳和机器码.那么下面先把问题代码贴出来,由于具体分析较长,这里先简述下根因,后面不感兴趣可以不看

简述根因

本质上是golang运行时的单调时钟和物理世界的墙上时钟不一致导致的。

当golang的单调时钟跑过1s后，会获取墙上时钟并塞到ticker.C中，上述两个操作并不同时，也不是同一个时间。

【单调时钟没有单位】

• 在单调时钟1000的时候，隔了0，调用了time.Now()，得到了XX:XX:11:9999；
• 在单调时钟2000的时候，隔了2，调用了time.Now()，得到了XX:XX:12:0001；
• 在单调时钟3000的时候，隔了0，调用了time.Now()，得到了XX:XX:12:9999；
• 在单调时钟4000的时候，隔了3，调用了time.Now()，得到了XX:XX:14:0002；

我们对比1,2，发现间隔超过了1s。对比2,3，发现间隔小于1s。

问题代码

func init() {
    // 先设置最初的时间，保证基本的正确性
    now := time.Now()
    updateUnixTimestamp(uint64(now.Unix()))

    var err error
    g, err = NewGenerator()
    if err != nil {
       panic(fmt.Sprintf("init default generator failed. err=%v", err))
    }
    g2, err = NewGenerator()
    if err != nil {
       panic(fmt.Sprintf("init second default generator failed. err=%v", err))
    }
    go func() {
       // sleep到下一秒开始，再创建一个ticker，尽量从某一秒的开始
       time.Sleep(time.Until(time.Now().Truncate(time.Second).Add(time.Second)))
       tk := time.NewTicker(time.Second)
       // ticker的更新会从下一秒开始，当前的这一秒还是需要立刻更新
       now = time.Now()
       updateUnixTimestamp(uint64(now.Unix()))
       for {
          // 拿到这个ticker chan返回的时间
          now = <-tk.C
          updateUnixTimestamp(uint64(now.Unix()))
       }
    }()
}

// 为所有的generator设置时间counter
func updateUnixTimestamp(timestamp uint64) {
    atomic.StoreUint64(&gUnixTimestamp, timestamp)
    tsHigh := timestamp << 32
    gGeneratorsMutex.Lock()
    for i := range gGenerators {
       atomic.StoreUint64(&gGenerators[i].timestampCounter, tsHigh)
    }
    gGeneratorsMutex.Unlock()
}

func (i *Generator) NextUint64() uint64 {
    c := atomic.AddUint64(&i.timestampCounter, 1)
    return (c & high32) | ((c & low16) << 16) | i.Config.workerID16
}

根因分析

从上面的方法中可以看出,这个id生成依赖了秒级时间戳和机器码,机器码我们已经排查了不会重复,那么最可能得原因就是时间戳重复导致,起初我们怀疑是ntp服务问题导致的时间回退,但是排查后发现ntp并没有问题,我们把怀疑的方向转向go的timer实现,下面我们来看go的timer实现

timer实现

目前线上的服务使用的是Go 1.20版本,我们看下go 1.20版本的go ticker如何触发运行的,这里不会展示完整的timer实现链,如果想了解timer整体实现可以参考 深入解析go Timer 和Ticker实现原理

NewTicker

我们先看下ticker初始化,重点关注sendTime(也就是后续的f)

可以看到ticker是触发sendTime时才去获得的最新时间,并尝试塞给了channel,如果channel满了则丢弃

startTimer这个实现不再展示(使用的是runtime包的startTimer),大致逻辑是把这个timer绑定到proccesser上,并放到这个processer的timer堆中相应的位置上

python

func NewTicker(d Duration) *Ticker {
    if d <= 0 {
       panic(errors.New("non-positive interval for NewTicker"))
    }
    // Give the channel a 1-element time buffer.
    // If th编程客栈e client falls behind while reading, we drop ticks
    // on the floor until the client catches up.
    c := make(chan Time, 1)
    t := &Ticker{
       C: c,
       r: runtimeTimer{
          when:   when(d),
          period: int64(d),
          f:      sendTime,
          arg:    c,
       },
    }
    startTimer(&t.r)
    return t
}

// sendTime does a non-blocking send of the current time on c.
func sendTime(c any, seq uintptr) {
    select {
    case c.(chan Time) <- Now():
    default:
    }
}

runtimer

我们看下1.20的go如何运行的timer

func runtimer(pp *p, now int64) int64 {
    for {
       t := pp.timers[0]
       if t.pp.ptr() != pp {
          throw("runtimer: bad p")
       }
       switch s := t.status.Load(); s {
       case timerWaiting:
          if t.when > now {
             // Not ready to run.
             return t.when
          }

          if !t.status.CompareAndSwap(s, timerRunning) {
             continue
          }
          // 重点就是这个方法
          runOneTimer(pp, t, now)
          return 0

       case timerDeleted:
       // 下面的逻辑对这个问题没有影响 忽略
          .....
    }
}

func runOneTimer(pp *p, t *timer, now int64) {
    f := t.f
    arg := t.arg
    seq := t.seq

    if t.period > 0 {
       // 对于ticker 会先设置下次运行的时间,然后重新触发堆排序
       delta := t.when - now // t.when 一定小于等于 now，所以delta是个负数
       // 整数除整数，得到的还是整数。
       // delta一般会比t.period小特别多 (在1s的ticker下，t.period也已经是10^6了）
       // 所以这个除法的结果大概率是0，所以这里的加减不太影响 t.when 
       t.when += t.period * (1 + -delta/t.period) 
       if t.when < 0 { // check for overflow.
          t.when = maxWhen
}
       siftdownTimer(pp.timers, 0)
       if !t.status.CompareAndSwap(timerRunning, timerWaiting) {
          badTimer()
       }
       updateTimer0When(pp)
    } else {
       // Remove from heap.
       dodeltimer0(pp)
       if !t.status.CompareAndSwap(timerRunning, timerNoStatus) {
          badTimer()
       }
    }

    unlock(&pp.timersLock)
    // 编程触发sendTimer
    f(arg, seq)

    lock(&pp.timersLock)
}

从上面的代码其实就可以看到问题了,下次触发的时间和sendTime拿到的时间不是一致的,也就是说如果unlock或者其他操作执行的较慢,那很可能sendTime这次拿到的时间是比预期晚,而下次拿到的时间比预期早,正好这个id生成器尽量从整秒开始,当出现上面描述的情况就会出现两次在同一秒的情况,导致id重复,同时当go调度器较忙时,可能触发runtimer的时间比预期晚,这个时候相当于返回的时间大于1s了,很可能又把之前小于1s的误差追平了,这个时候如果再出现小于1s的情况,可能又会触发id重复。所以日志中会看到多次出现id重复问题

验证

我们写一个很简单的ticker

package main

import (
    "fmt"
    "time"
)

func main() {
    // 为了更容易复现问题,这里尽量从接近整秒但不足整秒开始
    time.Sleep(time.Until(time.Now().Truncate(time.Second).Add(999099999 * time.Nanosecond)))
    tick := time.NewTicker(1 * time.Second)
    for i := 0; i < 5; i++ {
       c := <-tick.C
       fmt.Println("tick", i, ":", c.Format(time编程.StampNano))
    }
}

然后修改sendTime方法,我们记录下上次触发的时间戳,然后和这次的时间戳比较

var prephp int64

// sendTime does a non-blocking send of the current time on c.
func sendTime(c any, seq uintptr) {
    var n = runtimeNano()
    println("send", n-pre)
    pre = n
    select {
    case c.(chan Time) <- Now():
    default:
    }
}

测试结果

可以看到这个sendTime的间隔先是不足1s后又超过1s,5次ticker中出现了2次落到同一秒的情况

观察上述的输出，和【简述根因】中的推演结果一致。结论成立。

到此这篇关于详解如何解决golang定时器引发的id重复问题的文章就介绍到这了,更多相关golang定时器引发id重复内容请搜索编程客栈(www.devze.com)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程客栈(www.devze.com)！