雪花算法 | 晨露微凉

来源

雪花算法（Snowflake）是一种生成分布式全局唯一ID的算法，生成的ID称为Snowflake IDs或snowflakes。这种算法由Twitter创建，并用于推文的ID。Discord和Instagram等其他公司采用了修改后的版本。

格式

一个Snowflake ID有64位。前41位是时间戳，表示了自选定的时期以来的毫秒数。接下来的10位代表计算机ID，防止冲突。其余12位代表每台机器上生成ID的序列号，这允许在同一毫秒内创建多个Snowflake ID。最后以十进制将数字序列化。

SnowflakeID基于时间生成，故可以按时间排序。此外，一个ID的生成时间可以由其自身推断出来，反之亦然。该特性可以用于按时间筛选ID，以及与之联系的对象。

范例

2022年六月由@Wikipedia所发的一条推文的雪花ID是1541815603606036480。这个数字被转换成二进制就是0 0101 0101 1001 0110 1000 0100 0111 1101 1000 1000|01 0111 1010|0000 0000 0000，其中以竖线分隔成三个部分。

64位的二进制所示

最高位表示符号位
后面的41bit是产生该ID的unix毫秒时间戳
10bit是机器编号,最多可以部署在 2^8 = 1024 机器上
12bit 序列号,同一毫秒最多可以产生 2^12 = 4096 个序列号

算法实现思路

时间戳左移22位,机器编号左移12位,序列号不动,三者按位或运算,得到一个64位二进制,再转成10进制,就是雪花ID

反之,根据雪花ID可以反推导出机器ID,时间戳,序列号

41位时间戳2^41/(1000360024*365) ≈ 69.73057年,引入 epoch基准时间,是为了能是雪花ID能使用的时间更长一点,41位时间戳按照unix时间,最多只能到2039-09-07 23:47:35

省流不看系列

github jokechat/guid

1 2	docker run --rm -p 8080:80 jokechat/guid:v1.0.3 curl 127.0.0.1:8080

{
  "code": 0,
  "msg": "success",
  "data": {
    "id": 645124828453408800,
    "workId": 0,
    "base32": "13pv378bo0800",
    "time": "2023-11-16T04:55:44.943Z"
  }
}

代码实现

type.go

package snowflake

import "time"

type Option interface {
	apply(*Worker)
}

type OptionFunc func(*Worker)

func (o OptionFunc) apply(worker *Worker) {
	o(worker)
}

func WithEpoch(epoch time.Time) OptionFunc {
	return func(worker *Worker) {
		worker.epoch = epoch
	}
}

func WithWorkerId(workerId int64) OptionFunc {
	return func(worker *Worker) {
		worker.workerId = workerId
	}
}

snowflake.go

package snowflake

import (
	"errors"
	"fmt"
	"sync"
	"time"
)

const (
	workerBits uint8 = 10
	workerMax  int64 = -1 ^ (-1 << workerBits)

	stepBits uint8 = 12
	stepMax  int64 = -1 ^ (-1 << stepBits)

	workerShift uint8 = stepBits
	timeShift   uint8 = workerBits + stepBits
)

type Worker struct {
	mu sync.Mutex

	timestamp int64     // 记录时间戳
	workerId  int64     // 当前工作节点ID
	step      int64     // 当前毫秒已经生成的id序列号(从0开始累加) 1毫秒内最多生成4096个ID
	epoch     time.Time // 开始运行时间

}

func NewWorkerWithOpts(opts ...Option) (*Worker, error) {
	w := &Worker{}
	for _, opt := range opts {
		opt.apply(w)
	}

	if w.epoch.IsZero() {
		return nil, errors.New("epoch is required")
	}

	if w.workerId < 0 || w.workerId > workerMax {
		return nil, errors.New(fmt.Sprintf("worker ID must be in [0,%d]", workerMax))
	}

	return w, nil
}

func (w *Worker) Next() ID {
	w.mu.Lock()
	defer w.mu.Unlock()
	now := time.Now().UnixMilli()

	// 处理时钟回拨问题或处于同一毫秒 ntp 时间变化
	if now < w.timestamp {
		now = w.timestamp + 1
	}

	if now == w.timestamp {
		w.step = (w.step + 1) & stepMax
		if w.step == 0 { // 当前毫秒生成的ID已超上限,等待下一毫秒
			for now <= w.timestamp {
				now = time.Now().UnixMilli()
			}
		}
	} else {
		w.step = 0
		// todo 持久化存储当前时间,文件或者redis
	}
	w.timestamp = now
	id := (now-w.epoch.UnixMilli())<<timeShift | (w.workerId << workerShift) | w.step
	return ID(id)
}

id.go

package snowflake

import (
	"bytes"
	"encoding/base32"
	"encoding/binary"
	"strconv"
	"strings"
	"time"
)

type ID uint64

func (i ID) Uint64() uint64 {
	return uint64(i)
}

func (i ID) String() string {
	return strconv.FormatUint(uint64(i), 10)
}

func (i ID) Base32() string {
	bytesBuffer := bytes.NewBuffer([]byte{})
	binary.Write(bytesBuffer, binary.BigEndian, i)
	return base32.HexEncoding.WithPadding(base32.NoPadding).EncodeToString(bytesBuffer.Bytes())
}

func (i ID) Base32Lower() string {
	return strings.ToLower(i.Base32())
}

func (i ID) UnixMilli(epoch time.Time) int64 {
	return epoch.UnixMilli() + int64(i.Uint64()>>timeShift)
}

func (i ID) WorkId() uint64 {
	d := i.Uint64() >> workerShift & uint64(workerMax)
	return d
}

func (i ID) Step() uint64 {
	d := i.Uint64() & uint64(stepMax)
	return d
}

func (i ID) Time(epoch time.Time) time.Time {
	return time.UnixMilli(i.UnixMilli(epoch))
}

func main() {
    epoch := time.Date(2019, time.January, 1, 0, 0, 0, 0, time.Local)
    w, _ := NewWorkerWithOpts(
        WithEpoch(epoch),
        WithWorkerId(1),
    )
    id := w.Next()
    spew.Dump(id.Uint64())
}

参考文献

维基百科雪花算法