关于golang:LeafSegment分布式ID生成系统Golang实现版本

Leaf-Segment

IDjavagobugbugStarleaf-segment

疾速应用

创立数据库

CREATE TABLE `leaf_alloc` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `biz_tag` varchar(128)  NOT NULL DEFAULT '',
  `max_id` bigint(20) NOT NULL DEFAULT '1',
  `step` int(11) NOT NULL,
  `description` varchar(256)  DEFAULT NULL,
  `update_time` bigint(20) NOT NULL DEFAULT '0',
  PRIMARY KEY (`id`),
  UNIQUE KEY (`biz_tag`)
) ENGINE=InnoDB;

SQL

获取并运行我的项目

// 1 新建文件目录
$ mdkir asong.cloud
$ cd asong.cloud
// 2 获取我的项目
$ git clone git@github.com:asong2020/go-algorithm.git
// 3 进入我的项目目录
$ cd go-go-algorithm/leaf
// 4 运行
$ go run main.go //运行

测试

创立业务号段

URI: POST http://localhost:8080/api/leaf
Param(json):
{
    "biz_tag": "test_create_one",
    "max_id":  1, // 能够不传 默认为1
    "step": 2000, // 能够不传 默认为200
    "descprition": "test api one"
}

• 示例：

curl --location --request POST 'http://localhost:8080/api/leaf' \
--header 'Content-Type: application/json' \
--data-raw '{
    "biz_tag": "test_create_one",
    "descprition": "test api one"
}'

初始化DB中的号段到内存中

URI: PUT http://localhost:8080/api/leaf/init/cache
Param(json):
{
    "biz_tag": "test_create"
}

• 示例

curl --location --request PUT 'http://localhost:8080/api/leaf/init/cache' \
--header 'Content-Type: application/json' \
--data-raw '{
    "biz_tag": "test_create"
}'

获取ID

URI: GET http://localhost:8080/api/leaf
Param: 
?biz_tag=test_create

• 示例

curl --location --request GET 'http://localhost:8080/api/leaf?biz_tag=test_create'

step

URI: PUT http://localhost:8080/api/leaf/step
Param(json):
{
    "step":   10000,
    "biz_tag": "test_create"
}

• 示例

curl --location --request PUT 'http://localhost:8080/api/leaf/step' \
--header 'Content-Type: application/json' \
--data-raw '{
    "step": 10000,
    "biz_tag": "test_create"
}'

Leaf-Segment计划实现

背景

在简单分布式系统中，往往须要对大量的数据和音讯进行惟一标识。一个可能生成全局惟一ID的零碎是十分必要的。比方某宝，业务散布宽泛，这么多业务对数据分库分表后须要有一个惟一ID来标识一条数据或音讯，数据库的自增ID显然不能满足需要；所以，咱们能够总结一下业务系统对ID号的要求有哪些呢？

1. 全局唯一性：不能呈现反复的ID号，既然是惟一标识，这是最根本的要求。
2. 趋势递增：在MySQL InnoDB引擎中应用的是汇集索引，因为少数RDBMS应用B-tree的数据结构来存储索引数据，在主键的抉择下面咱们应该尽量应用有序的主键保障写入性能。
3. 枯燥递增：保障下一个ID肯定大于上一个ID，例如事务版本号、IM增量音讯、排序等非凡需要。
4. 信息安全：如果ID是间断的，歹意用户的扒取工作就非常容易做了，间接依照程序下载指定URL即可；如果是订单号就更危险了，竞对能够间接晓得咱们一天的单量。所以在一些利用场景下，会须要ID无规则、不规则。

上述123对应三类不同的场景，3和4需要还是互斥的，无奈应用同一个计划满足。

leaf-segment雪花算法

数据库生成

leaf-sement

auto_increment_incrementauto_increment_offset

begin;
REPLACE INTO Tickets64 (stub) VALUES ('a');
SELECT LAST_INSERT_ID();
commit;

这种计划的优缺点如下：

长处：

• 非常简单，利用现有数据库系统的性能实现，老本小，有DBA业余保护。
• ID号枯燥自增，能够实现一些对ID有特殊要求的业务。

毛病：

• 强依赖DB，当DB异样时整个零碎不可用，属于致命问题。配置主从复制能够尽可能的减少可用性，然而数据一致性在非凡状况下难以保障。主从切换时的不统一可能会导致反复发号。
• ID发号性能瓶颈限度在单台MySQL的读写性能。

对于MySQL性能问题，可用如下计划解决：在分布式系统中咱们能够多部署几台机器，每台机器设置不同的初始值，且步长和机器数相等。比方有两台机器。设置步长step为2，TicketServer1的初始值为1（1，3，5，7，9，11…）、TicketServer2的初始值为2（2，4，6，8，10…）。这是Flickr团队在2010年撰文介绍的一种主键生成策略（Ticket Servers: Distributed Unique Primary Keys on the Cheap ）。如下所示，为了实现上述计划别离设置两台机器对应的参数，TicketServer1从1开始发号，TicketServer2从2开始发号，两台机器每次发号之后都递增2。

TicketServer1:
auto-increment-increment = 2
auto-increment-offset = 1

TicketServer2:
auto-increment-increment = 2
auto-increment-offset = 2

假如咱们要部署N台机器，步长需设置为N，每台的初始值顺次为0,1,2…N-1那么整个架构就变成了如下图所示：

这种架构貌似可能满足性能的需要，但有以下几个毛病：

• 零碎程度扩大比拟艰难，比方定义好了步长和机器台数之后，如果要增加机器该怎么做？假如当初只有一台机器发号是1,2,3,4,5（步长是1），这个时候须要扩容机器一台。能够这样做：把第二台机器的初始值设置得比第一台超过很多，比方14（假如在扩容工夫之内第一台不可能发到14），同时设置步长为2，那么这台机器下发的号码都是14当前的偶数。而后摘掉第一台，把ID值保留为奇数，比方7，而后批改第一台的步长为2。让它合乎咱们定义的号段规范，对于这个例子来说就是让第一台当前只能产生奇数。扩容计划看起来简单吗？貌似还好，当初设想一下如果咱们线上有100台机器，这个时候要扩容该怎么做？几乎是噩梦。所以零碎程度扩大计划简单难以实现。
• ID没有了枯燥递增的个性，只能趋势递增，这个毛病对于个别业务需要不是很重要，能够容忍。
• 数据库压力还是很大，每次获取ID都得读写一次数据库，只能靠堆机器来进步性能

Leaf-Segment

Leaf-Segment

做了如下扭转： – 原计划每次获取ID都得读写一次数据库，造成数据库压力大。改为利用proxy server批量获取，每次获取一个segment(step决定大小)号段的值。用完之后再去数据库获取新的号段，能够大大的加重数据库的压力。 – 各个业务不同的发号需要用biz_tag字段来辨别，每个biz-tag的ID获取互相隔离，互不影响。如果当前有性能需求须要对数据库扩容，不须要上述形容的简单的扩容操作，只须要对biz_tag分库分表就行。

数据库表设计如下：

CREATE TABLE `leaf_alloc` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `biz_tag` varchar(128)  NOT NULL DEFAULT '',
  `max_id` bigint(20) NOT NULL DEFAULT '1',
  `step` int(11) NOT NULL,
  `description` varchar(256)  DEFAULT NULL,
  `update_time` bigint(20) NOT NULL DEFAULT '0',
  PRIMARY KEY (`id`),
  UNIQUE KEY (`biz_tag`)
) ENGINE=InnoDB;

这里我仍旧应用了一个自增主键，不过没什么用，能够疏忽。biz_tag用来辨别业务(所以我把它设置成了惟一索引)，max_id示意该biz_tag目前所被调配的ID号段的最大值，step示意每次调配的号段长度。原来获取ID每次都须要写数据库，当初只须要把step设置得足够大，比方1000。那么只有当1000个号被耗费完了之后才会去从新读写一次数据库。读写数据库的频率从1减小到了1/step，大抵架构如下图所示：

test_tag在第一台Leaf机器上是1~1000的号段，当这个号段用完时，会去加载另一个长度为step=1000的号段，假如另外两台号段都没有更新，这个时候第一台机器新加载的号段就应该是3001~4000。同时数据库对应的biz_tag这条数据的max_id会从3000被更新成4000，更新号段的SQL语句如下：

Begin
UPDATE table SET max_id=max_id+step WHERE biz_tag=xxx
SELECT tag, max_id, step FROM table WHERE biz_tag=xxx
Commit

这种模式有以下优缺点：

长处：

• Leaf服务能够很不便的线性扩大，性能齐全可能撑持大多数业务场景。
• ID号码是趋势递增的8byte的64位数字，满足上述数据库存储的主键要求。
• 容灾性高：Leaf服务外部有号段缓存，即便DB宕机，短时间内Leaf仍能失常对外提供服务。
• 能够自定义max_id的大小，十分不便业务从原有的ID形式上迁徙过去。

毛病：

• ID号码不够随机，可能泄露发号数量的信息，不太平安。
• TP999数据稳定大，当号段应用完之后还是会hang在更新数据库的I/O上，tg999数据会呈现偶然的尖刺。
• DB宕机会造成整个零碎不可用。

双buffer优化

对于第二个毛病，Leaf-segment做了一些优化，简略的说就是：

Leaf 取号段的机会是在号段耗费完的时候进行的，也就意味着号段临界点的ID下发工夫取决于下一次从DB取回号段的工夫，并且在这期间进来的申请也会因为DB号段没有取回来，导致线程阻塞。如果申请DB的网络和DB的性能稳固，这种状况对系统的影响是不大的，然而如果取DB的时候网络产生抖动，或者DB产生慢查问就会导致整个零碎的响应工夫变慢。

为此，咱们心愿DB取号段的过程可能做到无阻塞，不须要在DB取号段的时候阻塞申请线程，即当号段生产到某个点时就异步的把下一个号段加载到内存中。而不须要等到号段用尽的时候才去更新号段。这样做就能够很大水平上的升高零碎的TP999指标。具体实现如下图所示：

采纳双buffer的形式，Leaf服务外部有两个号段缓存区segment。以后号段已下发10%时，如果下一个号段未更新，则另启一个更新线程去更新下一个号段。以后号段全副下发完后，如果下个号段筹备好了则切换到下个号段为以后segment接着下发，周而复始。

• 每个biz-tag都有生产速度监控，通常举荐segment长度设置为服务高峰期发号QPS的600倍（10分钟），这样即便DB宕机，Leaf仍能继续发号10-20分钟不受影响。
• 每次申请来长期都会判断下个号段的状态，从而更新此号段，所以偶然的网络抖动不会影响下个号段的更新。

代码实现

终于到本文的重点了，上面就给大家解说一下我是怎么实现的。

代码架构

这里先贴一下我的代码架构，具体如下：

leaf
├── common -- common包，放的是一些client初始的代码
├── conf   -- conf包，配置文件
├── config -- config包，读取配置文件代码局部
├── dao    -- dao包，DB操作局部
├── handler -- hanler包，路由注册即API代码实现局部
├── images -- 本文的图片文件
├── model -- model包，db模型或其余模型
├── service -- service包，逻辑实现局部
├── wire    -- wire包，依赖绑定
├── leaf_svr.go -- main运行先置条件
└── main.go -- main函数

实现剖析

在咱们实现之前，必定要剖析一波，咱们要做什么，怎么做？我老大常常跟我说的一句话：”先把需要剖析明确了，再入手，返工反而是浪费时间”。

mapbiz_tagkey

// 全局分配器
// key: biz_tag value: SegmentBuffer
type LeafSeq struct {
    cache sync.Map
}

sync.map

IDstruct

// 号段
type LeafSegment struct {
    Cursor uint64 // 以后发放地位
    Max    uint64 // 最大值
    Min    uint64 // 开始值即最小值
    InitOk bool   // 是否初始化胜利
}

CursorminmaxbufferbufferinitOK

id

type LeafAlloc struct {
    Key        string                 // 也就是`biz_tag`用来辨别业务
    Step       int32                  // 记录步长
    CurrentPos int32                  // 以后应用的 segment buffer光标; 总共两个buffer缓存区，循环应用
    Buffer     []*LeafSegment         // 双buffer 一个作为预缓存作用
    UpdateTime time.Time              // 记录更新工夫 不便长时间不必进行清理，避免占用内存
    mutex      sync.Mutex             // 互斥锁
    IsPreload  bool                   // 是否正在预加载
    Waiting    map[string][]chan byte // 挂起期待
}

keybiz_tagmapStepcurrentPosbufferbufferbufferUpdate_timeIsPreloadwaitingmap+chan

根本思维就是这样啦，上面咱们就分块看一下代码。

先从DB层走起

DBCRUD

func (l *LeafDB) Create(ctx context.Context, leaf *model.Leaf) error {}
func (l *LeafDB) Get(ctx context.Context, bizTag string, tx *sql.Tx) (*model.Leaf, error) {}
func (l *LeafDB) UpdateMaxID(ctx context.Context, bizTag string, tx *sql.Tx) error {}
func (l *LeafDB) UpdateMaxIdByCustomStep(ctx context.Context, step int32, bizTag string, tx *sql.Tx) error {}
func (l *LeafDB) GetAll(ctx context.Context) ([]*model.Leaf, error) {}
func (l *LeafDB) UpdateStep(ctx context.Context, step int32, bizTag string) error {

leafstepstep

实现获取新号段局部的代码

先贴出我的代码：

func (l *LeafDao) NextSegment(ctx context.Context, bizTag string) (*model.Leaf, error) {
    // 开启事务
    tx, err := l.sql.Begin()
    defer func() {
        if err != nil {
            l.rollback(tx)
        }
    }()
    if err = l.checkError(err); err != nil {
        return nil, err
    }
    err = l.db.UpdateMaxID(ctx, bizTag, tx)
    if err = l.checkError(err); err != nil {
        return nil, err
    }
    leaf, err := l.db.Get(ctx, bizTag, tx)
    if err = l.checkError(err); err != nil {
        return nil, err
    }
    // 提交事务
    err = tx.Commit()
    if err = l.checkError(err); err != nil {
        return nil, err
    }
    return leaf, nil
}

func (l *LeafDao) checkError(err error) error {
    if err == nil {
        return nil
    }
    if message, ok := err.(*mysql.MySQLError); ok {
        fmt.Printf("it's sql error; str:%v", message.Message)
    }
    return errors.New("db error")
}

func (l *LeafDao) rollback(tx *sql.Tx) {
    err := tx.Rollback()
    if err != sql.ErrTxDone && err != nil {
        fmt.Println("rollback error")
    }
}

实现其实很简略，也就是先更新一下数据库中的号段，而后再取出来就能够了，这里为了保证数据的一致性和避免屡次更新DB导致号段失落，所以应用了事务，没有什么特地的点，看一下代码就能懂了。

初始化及获取ID

ID

// 第一次应用要初始化也就是把DB中的数据存到内存中,非必须操作，间接应用的话有兜底策略
func (l *LeafService) InitCache(ctx context.Context, bizTag string) (*model.LeafAlloc, error) {
    leaf, err := l.dao.NextSegment(ctx, bizTag)
    if err != nil {
        fmt.Printf("initCache failed; err:%v\n", err)
        return nil, err
    }
    alloc := model.NewLeafAlloc(leaf)
    alloc.Buffer = append(alloc.Buffer, model.NewLeafSegment(leaf))

    _ = l.leafSeq.Add(alloc)
    return alloc, nil
}

这里步骤次要分两步：

map

id

先看主流程：

func (l *LeafService) GetID(ctx context.Context, bizTag string) (uint64, error) {
    // 先去内存中看一下是否曾经初始了，未初始化则开启兜底策略初始化一下。
    l.mutex.Lock()
    var err error
    seqs := l.leafSeq.Get(bizTag)
    if seqs == nil {
        // 不存在初始化一下
        seqs, err = l.InitCache(ctx, bizTag)
        if err != nil {
            return 0, err
        }
    }
    l.mutex.Unlock()

    var id uint64
    id, err = l.NextID(seqs)
    if err != nil {
        return 0, err
    }
    l.leafSeq.Update(bizTag, seqs)

    return id, nil
}

次要分为三步：

id

id

func (l *LeafService) NextID(current *model.LeafAlloc) (uint64, error) {
    current.Lock()
    defer current.Unlock()
    var id uint64
    currentBuffer := current.Buffer[current.CurrentPos]
    // 判断以后buffer是否是可用的
    if current.HasSeq() {
        id = atomic.AddUint64(&current.Buffer[current.CurrentPos].Cursor, 1)
        current.UpdateTime = time.Now()
    }

    // 以后号段已下发10%时，如果下一个号段未更新加载，则另启一个更新线程去更新下一个号段
    if currentBuffer.Max-id < uint64(0.9*float32(current.Step)) && len(current.Buffer) <= 1 && !current.IsPreload {
        current.IsPreload = true
        cancel, _ := context.WithTimeout(context.Background(), 3*time.Second)
        go l.PreloadBuffer(cancel, current.Key, current)
    }

    // 第一个buffer的segment应用实现 切换到下一个buffer 并移除当初的buffer
    if id == currentBuffer.Max {
        // 判断第二个buffer是否筹备好了(因为下面开启协程去更新下一个号段会呈现失败)，筹备好了切换  currentPos 永远是0 不管怎么切换
        if len(current.Buffer) > 1 && current.Buffer[current.CurrentPos+1].InitOk {
            current.Buffer = append(current.Buffer[:0], current.Buffer[1:]...)
        }
        // 如果没筹备好，间接返回就好了，因为当初曾经调配id了, 前面会进行弥补
    }
    // 有id间接返回就能够了
    if current.HasID(id) {
        return id, nil
    }

    // 以后buffer曾经没有id可用了，此时弥补线程肯定正在运行，咱们期待一会
    waitChan := make(chan byte, 1)
    current.Waiting[current.Key] = append(current.Waiting[current.Key], waitChan)
    // 开释锁 期待让其余客户端进行走后面的步骤
    current.Unlock()

    timer := time.NewTimer(500 * time.Millisecond) // 期待500ms最多
    select {
    case <-waitChan:
    case <-timer.C:
    }

    current.Lock()
    // 第二个缓冲区仍未初始化好
    if len(current.Buffer) <= 1 {
        return 0, errors.New("get id failed")
    }
    // 切换buffer
    current.Buffer = append(current.Buffer[:0], current.Buffer[1:]...)
    if current.HasSeq() {
        id = atomic.AddUint64(&current.Buffer[current.CurrentPos].Cursor, 1)
        current.UpdateTime = time.Now()
    }
    return id, nil

}

这里我感觉用文字描述不分明，所以我画了个图，不晓得你们能不能看懂，能够对照着代码来看，这样是最清晰的，有问题欢送留言探讨：

预加载的流程也被我画下来了，预加载的步骤次要有三个：

buffer

代码实现如下：

func (l *LeafService) PreloadBuffer(ctx context.Context, bizTag string, current *model.LeafAlloc) error {
    for i := 0; i < MAXRETRY; i++ {
        leaf, err := l.dao.NextSegment(ctx, bizTag)
        if err != nil {
            fmt.Printf("preloadBuffer failed; bizTag:%s;err:%v", bizTag, err)
            continue
        }
        segment := model.NewLeafSegment(leaf)
        current.Buffer = append(current.Buffer, segment) // 追加
        l.leafSeq.Update(bizTag, current)
        current.Wakeup()
        break
    }
    current.IsPreload = false
    return nil
}
func (l *LeafAlloc) Wakeup() {
    l.mutex.Lock()
    defer l.mutex.Unlock()
    for _, waitChan := range l.Waiting[l.Key] {
        close(waitChan)
    }
    l.Waiting[l.Key] = l.Waiting[l.Key][:0]
}

缓存清理

timer

// 清理超过15min没用过的内存
func (l *LeafSeq) clear() {
    for {
        now := time.Now()
        // 15分钟后
        mm, _ := time.ParseDuration("15m")
        next := now.Add(mm)
        next = time.Date(next.Year(), next.Month(), next.Day(), next.Hour(), next.Minute(), 0, 0, next.Location())
        t := time.NewTimer(next.Sub(now))
        <-t.C
        fmt.Println("start clear goroutine")
        l.cache.Range(func(key, value interface{}) bool {
            alloc := value.(*LeafAlloc)
            if next.Sub(alloc.UpdateTime) > ExpiredTime {
                fmt.Printf("clear biz_tag: %s cache", key)
                l.cache.Delete(key)
            }
            return true
        })
    }
}

总结

好啦，到这里就是靠近序幕了，下面就是我实现的整个过程，目前本人测试没有什么问题，前期还会在缝缝补补，大家也能够帮我找找问题，欢送提出你们贵重的倡议～～～。

代码已收录到我的集体仓库——[go-算法系列(go-algorithm)](https://github.com/asong2020/…。

Star

好啦，这一篇文章到这就完结了，咱们下期见～～。心愿对你们有用，又不对的中央欢送指出，可增加我的golang交换群，咱们一起学习交换。

结尾给大家发一个小福利吧，最近我在看[微服务架构设计模式]这一本书，讲的很好，本人也收集了一本PDF，有须要的小伙能够到自行下载。获取形式：关注公众号：[Golang梦工厂]，后盾回复：[微服务]，即可获取。

我翻译了一份GIN中文文档，会定期进行保护，有须要的小伙伴后盾回复[gin]即可下载。

翻译了一份Machinery中文文档，会定期进行保护，有须要的小伙伴们后盾回复[machinery]即可获取。

golangvx

举荐往期文章：

• machinery-go异步工作队列
• 十张动图带你搞懂排序算法(附go实现代码)
• Go语言相干书籍举荐（从入门到放弃）
• go参数传递类型
• 手把手教姐姐写音讯队列
• 常见面试题之缓存雪崩、缓存穿透、缓存击穿
• 详解Context包，看这一篇就够了！！！
• go-ElasticSearch入门看这一篇就够了(一)
• 面试官：go中for-range应用过吗？这几个问题你能解释一下起因吗
• 学会wire依赖注入、cron定时工作其实就这么简略！
• 据说你还不会jwt和swagger-饭我都不吃了带着实际我的项目我就来了
• 把握这些Go语言个性，你的程度将进步N个品位(二)