codis-ha源码解析

原创
2016/03/21 22:13
阅读数 1.8K

本文将从源码角度解析codis的主从切换服务codis-ha。读者需要对codis的架构有所了解,并且熟悉基本的golang语法。 另外,本文假定读者熟悉golang一些常用包,比如flag,net/http,error等。

####获取源码 codis-ha的github地址为https://github.com/ngaut/codis-ha,执行

go get github.com/ngaut/codis-ha

可以将codis-ha下载到本地$GOPATH/src/github.com/ngaut/codis-ha目录下。

结构为:

-rw-r--r-- 1 root 197608   71 3月  19 15:16 aliverchecker.go
-rw-r--r-- 1 root 197608 1358 3月  19 15:16 main.go
-rw-r--r-- 1 root 197608  260 3月  19 15:16 README.md
-rw-r--r-- 1 root 197608  679 3月  19 15:16 redischecker.go
-rw-r--r-- 1 root 197608  461 3月  19 15:16 redischecker_test.go
-rw-r--r-- 1 root 197608  924 3月  19 15:16 rest.go
-rw-r--r-- 1 root 197608 3495 3月  19 15:16 servergroup.go
-rw-r--r-- 1 root 197608 3602 3月  19 15:16 servergroup_test.go

核心源码共五个go文件。其中main.go为入口函数(我用函数来概称一个方法/go文件/函数/程序片段),另外四个:

  • aliverchecker.go
  • redischecker.go
  • rest.go
  • servergroup.go

除此之外,redischecker_test.go和servergroup_test.go分别为测试文件。 README.md为markdown的说明文件(这是一句废话,而且显得我很二)。

解析的顺序为从局部到整体,称之为倒叙。

####倒叙之aliverchecker.go 倒叙之aliverchecker.go只定义了一个接口AliveChecker,用来抽象实现了CheckAlive() error方法的类型

type AliveChecker interface {
	CheckAlive() error
}

####倒叙之redischecker.go redischecker.go定义了一个redisChecker的类型,它实现了ping() error和CheckAlive() error两个方法,所以它也一个AliveChecker类型。

var (
	_ AliveChecker = &redisChecker{}
)

type redisChecker struct {
	addr           string
	defaultTimeout time.Duration
}

// 使用go的redis客户端"github.com/garyburd/redigo/redis"连接一下redis server,相当于一个心跳
func (r *redisChecker) ping() error {
	c, err := redis.DialTimeout("tcp", r.addr, r.defaultTimeout, r.defaultTimeout, r.defaultTimeout)
	if err != nil {
		return err
	}

	defer c.Close()
	_, err = c.Do("ping")
	return err
}

// 检查redis server的存活状态,尝试ping()三次 间隔3s/次,任意一次ping通即redis server状态良好,全部失败为redis server宕机
func (r *redisChecker) CheckAlive() error {
	var err error
	for i := 0; i < 2; i++ { //try a few times
		err = r.ping()
		if err != nil {
			time.Sleep(3 * time.Second)
			continue
		}

		return nil
	}

	return err
}

####倒叙之rest.go

rest.go封装了一个发送http请求到codis-config(dashboard)的方法

//call http url and get json, then decode to objptr
func httpCall(objPtr interface{}, url string, method string, arg interface{}) error {
    //返回一个http.Client对象,使用http默认协议
	client := &http.Client{Transport: http.DefaultTransport}
    //用一个bytes.Buffer来传输请求参数
	rw := &bytes.Buffer{}
	if arg != nil {
		buf, err := json.Marshal(arg)
		if err != nil {
			return errors.Trace(err)
		}
		rw.Write(buf)
	}
    
    //返回一个http的request对象,传入method url和保证的请求参数rw
	req, err := http.NewRequest(method, url, rw)
	if err != nil {
		return errors.Trace(err)
	}

	resp, err := client.Do(req)
	if err != nil {
		return errors.Trace(err)
	}
	defer resp.Body.Close()
    
    //确认返回2**的http response status code
	if resp.StatusCode/100 != 2 {
		msg, _ := ioutil.ReadAll(resp.Body)
		return errors.Errorf("error: %d, message: %s", resp.StatusCode, string(msg))
	}

    //用传入的objPtr类型将返回的json类型解析返回
	if objPtr != nil {
		return json.NewDecoder(resp.Body).Decode(objPtr)
	}

	return nil
}

####倒叙之servergroup.go

codis-ha的核心代码都在这个文件中

GetServerGroups通过调用callHttp()获取codis在zookeeper上注册的server groups信息

func GetServerGroups() ([]models.ServerGroup, error) {
	var groups []models.ServerGroup
	err := callHttp(&groups, genUrl(*apiServer, "/api/server_groups"), "GET", nil)
	return groups, err
}

PingServer用来检测codis server也就是redis服务的状态,并将CheckAlive的结果放进一个channel

func PingServer(checker AliveChecker, errCtx interface{}, errCh chan<- interface{}) {
	err := checker.CheckAlive()
	log.Debugf("check %+v, result:%v, errCtx:%+v", checker, err, errCtx)
	if err != nil {
		errCh <- errCtx
		return
	}
	errCh <- nil
}

verifyAndUpServer来确定一个offline的redis实例是存活的并将其设置为online状态

func verifyAndUpServer(checker AliveChecker, errCtx interface{}) {
	errCh := make(chan interface{}, 100)

	go PingServer(checker, errCtx, errCh)

	s := <-errCh

	if s == nil { //alive
		handleAddServer(errCtx.(*models.Server))
	}

}

getSlave: 当一个redis master挂掉的时候,用来获取同一个group下的slave

func getSlave(master *models.Server) (*models.Server, error) {
	var group models.ServerGroup
	err := callHttp(&group, genUrl(*apiServer, "/api/server_group/", master.GroupId), "GET", nil)
	if err != nil {
		return nil, errors.Trace(err)
	}

	for _, s := range group.Servers {
		if s.Type == models.SERVER_TYPE_SLAVE {
			return s, nil
		}
	}

	return nil, errors.Errorf("can not find any slave in this group: %v", group)
}

handleCrashedServer用来处理挂掉的redis,如果挂的redis是master则获取该codis group对应点slave并将其promote到master,其他则不作处理

func handleCrashedServer(s *models.Server) error {
	switch s.Type {
	case models.SERVER_TYPE_MASTER:
		//get slave and do promote
		slave, err := getSlave(s)
		if err != nil {
			log.Warning(errors.ErrorStack(err))
			return err
		}

		log.Infof("try promote %+v", slave)
		err = callHttp(nil, genUrl(*apiServer, "/api/server_group/", slave.GroupId, "/promote"), "POST", slave)
		if err != nil {
			log.Errorf("do promote %v failed %v", slave, errors.ErrorStack(err))
			return err
		}
	case models.SERVER_TYPE_SLAVE:
		log.Errorf("slave is down: %+v", s)
	case models.SERVER_TYPE_OFFLINE:
		//no need to handle it
	default:
		log.Fatalf("unkonwn type %+v", s)
	}

	return nil
}

handleAddServer将存活的redis添加其对应给group下作为slave节点

func handleAddServer(s *models.Server) {
	s.Type = models.SERVER_TYPE_SLAVE
	log.Infof("try reusing slave %+v", s)
	err := callHttp(nil, genUrl(*apiServer, "/api/server_group/", s.GroupId, "/addServer"), "PUT", s)
	log.Errorf("do reusing slave %v failed %v", s, errors.ErrorStack(err))
}

CheckAliveAndPromote是codis-ha的核心方法,解析见代码注释

//ping codis-server find crashed codis-server
func CheckAliveAndPromote(groups []models.ServerGroup) ([]models.Server, error) {
	errCh := make(chan interface{}, 100)    //建一个缓冲容量为100的channel
	var serverCnt int                       //serverCnt用来对所有group中的redis节点计数
	for _, group := range groups { //each group
		for _, s := range group.Servers { //each server
			serverCnt++                         //发现一个redis server
			rc := acf(s.Addr, 5*time.Second)    //返回一个redisChecker类型对象地址,其拥有CheckAlive()方法
			news := s                           //复制为news
			go PingServer(rc, news, errCh)      //检测该redis server状态,如果正常会将nil写入errCh,如果该节点挂了会将news写入errCh
		}
	}

	//get result
	var crashedServer []models.Server           //crashedServer用来记录挂掉的redis server
	for i := 0; i < serverCnt; i++ {            //按照所有redis server的数量进行遍历,不然s := <-errCh会一直阻塞
		s := <-errCh                            //阻塞知道errCh有内容写入
		if s == nil { //alive                   
			continue
		}

		log.Warningf("server maybe crashed %+v", s)
		crashedServer = append(crashedServer, *s.(*models.Server))  //将挂的节点加入crashedServer   

		err := handleCrashedServer(s.(*models.Server))  //调用handleCrashedServer出来挂掉的redis server
		if err != nil {
			return crashedServer, err
		}
	}

	return crashedServer, nil
}

//ping codis-server find node up with type offine(勉为其难翻译一下:ping codis-server找到一个offline的节点并将其online)
func CheckOfflineAndPromoteSlave(groups []models.ServerGroup) ([]models.Server, error) {
	for _, group := range groups { //each group
		for _, s := range group.Servers { //each server
			rc := acf(s.Addr, 5*time.Second)
			news := s
			if (s.Type == models.SERVER_TYPE_OFFLINE) {
				verifyAndUpServer(rc, news)
			}
		}
	}
	return nil, nil
}

####最后是main.go main.go除了程序入口外还定义了一些全局变量和一个包函数。

// 定义一个fnHttpCall的函数类型,接收四个参数,返回一个error
type fnHttpCall func(objPtr interface{}, api string, method string, arg interface{}) error

// 定义一个aliveCheckerFactory的函数类型,接收两个参数,返回一个AliveChecker类型。AliveChecker是一个接口类型,在aliverchecker.go中定义
type aliveCheckerFactory func(addr string, defaultTimeout time.Duration) AliveChecker

// 下面是一组全局变量
var (
    //调用flag的String()方法分别解析codis-ha启动时的命令行参数,并设置一个default值
    
    //codis-config参数,用来调用codis-cofig的api来配置codis的主从关系
	apiServer   = flag.String("codis-config", "localhost:18087", "api server address")
	//productName参数,用来连接zookeeper获取codis的各种注册信息
	productName = flag.String("productName", "test", "product name, can be found in codis-proxy's config")
	//logLevel参数,设置日志级别
	logLevel    = flag.String("log-level", "info", "log level")

    //定义一个fnHttpCall类型的函数callHttp,详细实现在rest.go中
	callHttp fnHttpCall          = httpCall
	
	//定义一个aliveCheckerFactory类型的函数acf,返回一个redisChecker类型实例地址,redisChecker定义在redischecker.go中
	acf      aliveCheckerFactory = func(addr string, timeout time.Duration) AliveChecker {
		return &redisChecker{
			addr:           addr,
			defaultTimeout: timeout,
		}
	}
)

总的来说,codis-ha做了下面三件事: - 获取codis server groups列表 - 检查codis server(redis实例)的状态 - 检查offline状态的codis server并将其上线(online)

func main() {
	flag.Parse()                        //解析命令行参数
	log.SetLevelByString(*logLevel)     //设置日志级别

	for {                                       //别停
		groups, err := GetServerGroups()        //获取codis server groups信息
		if err != nil {
			log.Error(errors.ErrorStack(err))
			return
		}

		CheckAliveAndPromote(groups)            //见CheckAliveAndPromote方法解析
		CheckOfflineAndPromoteSlave(groups)     //见CheckOfflineAndPromoteSlave方法解析
		time.Sleep(3 * time.Second)
	}
}

####后记 这篇文是我简略看过codis-ha之后写的,目的主要是学习golang的程序结构和组织。codis-ha本身也未作讨论,有问题可以一起交流。写的也匆忙,如有错误也请不吝指正。

展开阅读全文
打赏
2
9 收藏
分享
加载中
更多评论
打赏
0 评论
9 收藏
2
分享
在线直播报名
返回顶部
顶部