通过Consul Raft库打造自己的分布式系统

来源：cnblogs　　作者：egmkang　　时间：2020/12/8 9:09:39　　对本文有异议

通用的CP系统有etcd和consul, 通用的对立面就是专用系统. 所以在某些场合是有这种需求的.

然而etcd embed的可用性极差, Windows上面跑会出现各种问题, 而且不能定制协议, 你必须得用etcd定义好的协议和客户端来和etcd集群通讯. 所以这时候的选择:

1. 忍着

2. 自己实现一个raft算法库, 在这上面做应用

　有一定的可能性, 起码MIT 6.824可以做出来, 但是和工业应用还是有很大的差距

3. 找一个工业级raft库, 然后在这上面做应用

这时候到Raft Consensus Algorithm上面看看就能找到几个可选的Raft算法库, 例如braft, hashicorp/raft, lni/dragonboat.

但是呢, C++代码比较难写的, 所以就pass掉了braft. 就剩下consul raft和dragonboat.

本文就用consul raft做一个简单的KeyValue服务.

首先前端用的gin, 提供put/get/inc/delete几个接口, 三个接口都走raft状态机, 因为要支持多节点, 所以内部非leader节点就需要把请求转发给leader节点.

前端的代码类似于这样:

func (this *ApiService) Start() error {
        //转发请求给leader节点
	this.router.Use(this.proxyHandler())
 
	this.router.POST("/get", this.Get)
	this.router.POST("/put", this.Put)
	this.router.POST("/delete", this.Delete)
	this.router.POST("/inc", this.Inc)
 
	address := fmt.Sprintf(":%d", this.port)
	return this.router.Run(address)
}

请求都很简单, 就是直接把命令, 或者叫服务提供的原语塞到Raft状态机里面等候Raft状态Apply, 然后才能拿到结果(future/promise模式), 例如put命令:

func (this *ApiService) Put(ctx *gin.Context) {
	req := &Request{}
	if err := ctx.ShouldBindJSON(req); err != nil {
		ctx.JSON(http.StatusBadRequest, Response{
			Error: err.Error(),
		})
		return
	}
	result, err := this.raft.ApplyCommand(raft.CommandPut, req.Key, req.Value)
	if err != nil {
		ctx.JSON(http.StatusInternalServerError, Response{
			Error: err.Error(),
		})
		return
	}
	ctx.JSON(http.StatusOK, Response{
		Value: result.Value,
	})
}

前端还有一个转发请求到leader节点的拦截器(? 应该叫这个名字, 实际上是pipeline模式的一种)

func (this *ApiService) proxyHandler() gin.HandlerFunc {
	return func(context *gin.Context) {
		if this.raft.IsLeader() {
			context.Next()
		} else {
			leaderServiceAddress := this.raft.GetLeaderServiceAddress()
			if this.leaderServiceAddress != leaderServiceAddress {
				Director := func(req *http.Request) {
					req.URL.Scheme = "http"
					req.URL.Host = leaderServiceAddress
				}
				this.leaderProxy = &httputil.ReverseProxy{
					Director: Director,
				}
				this.leaderServiceAddress = leaderServiceAddress
			}
			this.leaderProxy.ServeHTTP(context.Writer, context.Request)
			context.Abort()
		}
	}
}

下面是对协议的处理:

func (this *FSM) Apply(log *raft.Log) interface{} {
	result := &FSMApplyResult{
		Success: false,
	}
	t, cmd, err := raftLogToCommand(log)
	if err != nil {
		result.Error = err
		return result
	}
	binary.LittleEndian.PutUint64(keyCache, uint64(cmd.Key))
	binary.LittleEndian.PutUint64(valueCache, uint64(cmd.Value))
	switch t {
	case CommandPut:
		result.Success, result.Error = this.add(keyCache, valueCache)
	case CommandDelete:
		result.Success, result.Error = this.delete(keyCache)
	case CommandGet:
		result.Value, result.Error = this.get(keyCache)
	case CommandInc:
		result.Value, result.Error = this.inc(keyCache, cmd.Value)
	}
	return result
}

输入给Raft状态的命令实际上都是序列化好的, Raft状态机会自己把命令保存到Storage里面(可以是内存, 也可以是磁盘/DB等). 所以Apply命令的时候, 先对raft log进行解码, 然后switch去处理.

这边再看看例如inc的处理:

func (this *FSM) inc(key []byte, add int64) (int64, error) {
	var value int64 = 0
	err := this.db.Update(func(tx *bbolt.Tx) error {
		b, err := tx.CreateBucketIfNotExists(BBoltBucket)
		if err != nil {
			return err
		}
		valueBytes := b.Get(key)
		if len(valueBytes) != 8 {
			logging.Errorf("FSM.inc, key:%d, value length:%d, Reset",
				int64(binary.LittleEndian.Uint64(key)), len(valueBytes))
			valueBytes = make([]byte, 8)
		}
		value = int64(binary.LittleEndian.Uint64(valueBytes))
		value += add
		binary.LittleEndian.PutUint64(valueBytes, uint64(value))
		err = b.Put(key, valueBytes)
		return err
	})
	if err != nil {
		return -1, err
	}
	return value, err
}