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

问题描述

线上服务日志中突然出现很多主键冲突的异常,而这个主键是一个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堆中相应的位置上

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 the 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 pre 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次落到同一秒的情况