使用 etcd 集群生成序列号

本篇文章给大家分享《使用 etcd 集群生成序列号》，覆盖了Golang的常见基础知识，其实一个语言的全部知识点一篇文章是不可能说完的，但希望通过这些问题，让读者对自己的掌握程度有一定的认识(B 数)，从而弥补自己的不足，更好的掌握它。

我正在探索 etcd 来实现分布式环境的序号生成器。我的要求是生成要在同一应用程序的多个实例的每个请求中使用的非重复序列号。并且符合要求的这样的应用程序可以有n个。我使用 golang 客户端包中提供的 stm 和互斥锁以多种方式对此进行了 poc

在本地机器设置中使用单节点 etcd 服务器（将是至少 3 个节点集群，以便 raft 在生产中工作），我编写了一个简单的程序来在 500 个 goroutine 中生成 ids（数字）。每个例程各有 10 个 id，因此总共有 5000 个 id。根据时间统计，具有重试尝试的 stm 比互斥锁表现更好。除了这些方法之外，是否有更好的选择来实现序列号生成？首先，etcd 可以用于此目的吗？

ps：我附上代码示例仅供参考。我不指望它会被审查。我关心的是使用 etcd 生成序列号的正确方法

package main

import (
    "context"
    "errors"
    "strconv"
    "sync"
    "sync/atomic"
    "time"

    CONC "go.etcd.io/etcd/clientv3/concurrency"

    "github.com/golang/glog"
    ETCD "go.etcd.io/etcd/clientv3"
)

var client *ETCD.Client
var deadline = 200 * time.Second

func main() {
    var err error
    client, err = ETCD.New(ETCD.Config{
        Endpoints: []string{"127.0.0.1:2379"},
    })
    if err != nil {
        glog.Errorln("err:", err)
        return
    }
    idGen := &SeqIDGenerator{key: "_id"}

    err = func() error {
        ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
        defer cancel()
        _, err = client.Put(ctx, idGen.key, strconv.FormatInt(0, 10))
        return err
    }()
    if err != nil {
        glog.Errorln("err:", err)
        return
    }
    id, err := idGen.nextWithMutex()
    if err != nil {
        glog.Errorln("err:", err)
        return
    }
    glog.Errorln("done", id)
    id, err = idGen.nextWithSTMSerialiazable()
    if err != nil {
        glog.Errorln("err:", err)
        return
    }
    glog.Errorln("done", id)
    // st := time.Now()
    // stressSTMSerialiazableSeq(idGen)
    // glog.Errorln(time.Since(st))

}

type SeqIDGenerator struct {
    key string
}

func (idGen *SeqIDGenerator) nextWithSTMSerialiazable() (int64, error) {

    var retrived int64
    ctx, cancel := context.WithTimeout(context.Background(), deadline)
    defer cancel()
    var err error
    retry := retry
    for retry > 0 {
        retry--
        stmresp, err := CONC.NewSTMSerializable(ctx, client, func(s CONC.STM) error {
            v := s.Get(idGen.key)
            retrived, err = strconv.ParseInt(v, 10, 64)
            if err != nil {
                return err
            }
            retrived++
            s.Put(idGen.key, strconv.FormatInt(retrived, 10))
            return nil
        })
        if err != nil {
            continue
        } else if stmresp.Succeeded {
            return retrived, nil
        }

    }
    return 0, errors.New("ID gen failed. Retry exceeded")
}

func (idGen *SeqIDGenerator) nextWithMutex() (int64, error) {
    s, err := CONC.NewSession(client) // explore options to pass
    if err != nil {
        return 0, err
    }
    m := CONC.NewMutex(s, idGen.key)
    ctx, cancel := context.WithTimeout(context.Background(), deadline)
    defer cancel()
    m.Lock(ctx)
    defer m.Unlock(ctx)
    resp, err := client.Get(ctx, idGen.key)
    if err != nil {
        return 0, err
    }

    retrived, err := strconv.ParseInt(string(resp.OpResponse().Get().Kvs[0].Value), 10, 64)
    if err != nil {
        return 0, err
    }
    retrived++
    _, err = client.Put(ctx, idGen.key, strconv.FormatInt(retrived, 10))
    if err != nil {
        return 0, err
    }
    return retrived, nil
}

func (idGen *SeqIDGenerator) nextWithSTMReapeatable() (int64, error) {

    var retrived int64
    ctx, cancel := context.WithTimeout(context.Background(), deadline)
    defer cancel()
    var err error
    retry := retry
    for retry > 0 {
        retry--
        stmresp, err := CONC.NewSTMRepeatable(ctx, client, func(s CONC.STM) error {
            v := s.Get(idGen.key)
            retrived, err = strconv.ParseInt(v, 10, 64)
            if err != nil {
                return err
            }
            retrived++
            s.Put(idGen.key, strconv.FormatInt(retrived, 10))
            return nil
        })
        if err != nil {
            continue
        } else if stmresp.Succeeded {
            return retrived, nil
        }

    }
    return 0, errors.New("ID gen failed. Retry exceeded")
}

var n int = 500
var retry int = 40 // move as conf

func stressMutex(idGen *SeqIDGenerator) {
    wg := &sync.WaitGroup{}
    wg.Add(n)
    for i := 0; i < n; i++ {
        go func(i int) {
            defer wg.Done()
            for j := 0; j < 10; j++ {
                _, err := idGen.nextWithMutex()
                if err != nil {
                    glog.Errorln("err:", err)
                    return
                }
            }
        }(i)
    }
    wg.Wait()

}

func stressMutexSeq(idGen *SeqIDGenerator) {
    for i := 0; i < n; i++ {
        for j := 0; j < 10; j++ {
            _, err := idGen.nextWithMutex()
            if err != nil {
                glog.Errorln("err:", err)
            }
        }
    }

}

func stressSTMSerialiazableSeq(idGen *SeqIDGenerator) {
    for i := 0; i < n; i++ {
        for j := 0; j < 10; j++ {
            _, err := idGen.nextWithSTMSerialiazable()
            if err != nil {
                glog.Errorln("err:", err)
            }
        }
    }

}

func stressSTMReapeatableSeq(idGen *SeqIDGenerator) {
    for i := 0; i < n; i++ {
        for j := 0; j < 10; j++ {
            _, err := idGen.nextWithSTMReapeatable()
            if err != nil {
                glog.Errorln("err:", err)
            }
        }
    }

}

func stressSTMSerialiazable(idGen *SeqIDGenerator) {
    wg := &sync.WaitGroup{}
    wg.Add(n)
    var success int64
    for i := 0; i < n; i++ {
        go func(i int) {
            defer wg.Done()
            for j := 0; j < 10; j++ {
                _, err := idGen.nextWithSTMSerialiazable()
                if err != nil {
                    glog.Errorln("err:", err)
                } else {
                    atomic.AddInt64(&success, 1)
                }

            }
        }(i)
    }
    wg.Wait()
    glog.Errorln("success:", success)

}

func stressSTMReapeatable(idGen *SeqIDGenerator) {
    wg := &sync.WaitGroup{}
    wg.Add(n)
    var success int64
    for i := 0; i < n; i++ {
        go func(i int) {
            defer wg.Done()
            for j := 0; j < 10; j++ {
                _, err := idGen.nextWithSTMReapeatable()
                if err != nil {
                    glog.Errorln("err:", err)
                } else {
                    atomic.AddInt64(&success, 1)
                }

            }
        }(i)
    }
    wg.Wait()
    glog.Errorln("success:", success)

}

解决方案

我自己也在看这个。以下是两种可能的方法： （我是 etcd 的新手，所以买者自负）

1. 如果您可以接受独特的单调值和大间隙，您可以使用etcd集群“修订版”，每次修改存储时它都会增加：放置密钥等。 （该值是从成熟的 etcd 客户端中的 put 操作返回的）

$ etcdctl put k v -w json
{"header":{"cluster_id":14841639068965178418,"member_id":10276657743932975437,"revision":8,"raft_term":3}}

使用客户端中“修订”返回的值。这实际上应该是一个 为调用“put”的客户端生成唯一值（全局）的原子操作。

1. 如果您需要没有间隙的唯一顺序值，看起来您可以使用与每个键关联的“mod_revision”值。 它是每个键的，并且在每次放置后都会增加。 当您删除密钥时，它会重置为零。

试试这个：

$ etcdctl put k v --prev-kv -w json
{"header":{"cluster_id":14841639068965178418,"member_id":10276657743932975437,"revision":10,"raft_term":3},"prev_kv":{"key":"aw==","create_revision":8,"mod_revision":8,"version":1,"value":"dg=="}}

同样，客户端中为“mod_revision”返回的值。 这取决于 'prev-kv' 是原子 wrt 'put'，这样任何 'put'+'prev-key' 形成原子操作。我相信确实如此，但没有引用。

我的规则是如何在两者之间做出决定的：

1. 如果我可以接受有间隙的唯一值，请使用全局“修订版”， 除非您从新的 kv 存储开始，否则它无法重置为零。 这是我更喜欢的方法，因为这不会被意外重置。

2. 如果您必须具有唯一的顺序值，请使用每个键“mod_revision”。 但请注意，如果您不小心删除了用于跟踪 mod_revision 的 k/v， 序列号将从零开始！这可能会给您带来大问题。但这里的优点是您可以拥有多个连续的唯一序列号。

到这里，我们也就讲完了《使用 etcd 集群生成序列号》的内容了。个人认为，基础知识的学习和巩固，是为了更好的将其运用到项目中，欢迎关注golang学习网公众号，带你了解更多关于的知识点！