七叶笔记 » golang编程 » Codis源码分析之环境篇

Codis源码分析之环境篇

分类: golang编程 | 浏览: 272

一、Codis介绍

Codis是豌豆荚开源的 Redis 集群方案，github地址：

以下是官方介绍：

Codis主要解决的是redis的扩展和运维问题，因为redis官方以前没有集群方案，自从3.0才有，并且刚开始做的比较弱，特别是运维这块不是很友好，很多都是命令行操作的。我们可以认为Codis是一个可以支持容量可以无限扩大的Redis集群就行。

二、Codis安装及配置

1、整体架构

Codis是由golang写的，因此需要先安装go语言，官方给的文档已经很详细了，这里不详述。

我们看下Codis的整体架构，基于3.2：

2、相关概念

Codis Server：可以理解为原生的Redis，在整个集群中充当存储用，即最终数据是存在Redis中的，不过Codis在上面改了些东西，加了些命令，便于集群迁移和分片；

Codis Proxy：客户端连接的 Redis 代理服务, 实现了 Redis 协议。即Proxy只是一个代理，把命令做解析，然后根据路由规则转到不同的Codis Server中，Proxy是无状态的，可以无限扩容，Proxy信息保存在Zookeeper等Storage（后面再介绍）中，客户端需要自己刷新最新的Proxy列表。

Codis Dashboard：这个是Codis的核心，对集群的绝大部分修改都是它来完成的。官方的介绍如下：集群管理工具，支持 codis-proxy、codis-server 的添加、删除，以及据迁移等操作；说的比较笼统，下面我们会通过代码来说明其作用。

Codis FE：集群管理界面，可以理解为UI。

Codis Admin：集群管理的命令行工具，可用于控制 codis-proxy、codis-dashboard 状态以及访问外部存储。为什么有了Dashboard，还要Codis Admin，这个工具的主要定位是通过命令行的方式操作，其实后面的逻辑是一样的，即Codis Admin和Codis Fe是两种不同的界面，最后的逻辑是一样的。

Storage：主要存储元数据，如集群有多少个 Proxy，当前的分片是怎么样的。目前有 Zookeeper、Etcd、Fs等方式的实现。

3、集群启动

整个集群的启动方式官方也有说明，大体顺序如下：

1）、启动codis-dashboard 

 /admin/codis-dashboard-admin.sh start

最终启动的是codis-dashboard这个二进制文件： 

 ./ bin /codis-dashboard --ncpu=4 -- config =dashboard.toml \    --log=dashboard.log --log-level=WARN

主要的参数是–config，即配置文件，核心几个配置如下： 

 coordinator_name = "zookeeper"
coordinator_addr = "127.0.0.1:2181"

# Set Codis Product {Name/Auth}.
product_name = "codis-demo"
product_auth = ""

# Set bind address for admin( rpc ), tcp only.
admin_addr = "0.0.0.0:18080"

coordinator_name和coordinator_addr配置的是Storage的类型及地址；

product_name配置的是集群的名称，Codis以Product作为一个集群，每个集群有不同的Codis Server和Codis Proxy及Dashboard；product_auth为密码，如果开发环境建议留空，生产上最好是不为空。

2）、启动codis-proxy 

 ./admin/codis-proxy-admin.sh start

最终启动的是codis-proxy这个二进制文件 

 ./bin/codis-proxy --ncpu=4 --config=proxy.toml \    --log=proxy.log --log-level=WARN &

最主要的是–config参数，核心参数有： 

 product_name = "codis-demo"
product_auth = ""

# Set bind address for admin(rpc), tcp only.
admin_addr = "0.0.0.0:11080"

product_name和product_auth要和上面的Codis Dashboard对应上

3）启动codis-server 

 ./admin/codis-server-admin.sh start

4）启动codis-fe 

 ./admin/codis-fe-admin.sh star

最终启动的是codis-fe这个二进制文件 

 ./bin/codis-fe --ncpu=4 --log=fe.log --log-level=WARN \    --zookeeper=127.0.0.1:2181 --listen=127.0.0.1:8080 &

listen参数比较重要，就是我们要在浏览器访问的地址了。

Codis Admin是命令行不用启动，每次执行一个命令就退出了。

三、代码分析各组件作用

前面讲到了Codis的和组件，总体还是比较多的，我刚开始也是一头雾水，通过看代码就比较清晰了。

我们先看下Codis整体代码结构：

代码大概在以下目录：

cmd：上面说的二进制文件的入口，即main函数所在文件，这个下面不同命令有不同的文件夹，都叫main.go。

pkg：核心的业务逻辑

vendor：主要是存放第三方代码库

1)、Codis Dashboard 

 
s, err := topom.New(client, config)
  if err != nil {
    log.PanicErrorf(err, "create topom with config file failed\n%s", config)
  }
  
  //省略一些代码
  for i := 0; !s.IsClosed() && !s.IsOnline(); i++ {
    //启动Topom
    if err := s.Start(true); err != nil {
      if i <= 15 {
        log.Warnf("[%p] dashboard online failed [%d]", s, i)
      } else {
        log.Panicf("dashboard online failed, give up & abort :'(")
      }
      time.Sleep(time.Second * 2)
    }
  }

dashboard的初始化逻辑都在上面了，几个关键代码是topom.New以及s.Start(true)，Start的逻辑以后再细讲，今天关注 topom的初始化。

topom.new会启动一个协程执行： 

 s.serveAdmin()

这个函数内部会启动一个 api Server，以下是具体的路由： 

 r.Get("/", func(r render.Render) {
    r.Redirect("/topom")
  })
  r.Any("/debug/**", func(w http.ResponseWriter, req *http.Request) {
    http.DefaultServeMux.ServeHTTP(w, req)
  })

  r.Group("/topom", func(r martini.Router) {
    r.Get("", api.Overview)
    r.Get("/model", api.Model)
    r.Get("/stats", api.StatsNoXAuth)
    r.Get("/slots", api.SlotsNoXAuth)
  })
  r.Group("/api/topom", func(r martini.Router) {
    r.Get("/model", api.Model)
    r.Get("/xping/:xauth", api.XPing)
    r.Get("/stats/:xauth", api.Stats)
    r.Get("/slots/:xauth", api.Slots)
    r. Put ("/reload/:xauth", api.Reload)
    r.Put("/shutdown/:xauth", api.Shutdown)
    r.Put("/loglevel/:xauth/:value", api.LogLevel)
    
    r.Group("/proxy", func(r martini.Router) {
      r.Put("/create/:xauth/:addr", api.CreateProxy)
      r.Put("/online/:xauth/:addr", api.OnlineProxy)
      r.Put("/reinit/:xauth/:token", api.ReinitProxy)
      r.Put("/remove/:xauth/:token/:force", api.RemoveProxy)
    })
    r.Group("/group", func(r martini.Router) {
      r.Put("/create/:xauth/:gid", api.CreateGroup)
      r.Put("/remove/:xauth/:gid", api.RemoveGroup)
      r.Put("/resync/:xauth/:gid", api.ResyncGroup)
      r.Put("/resync-all/:xauth", api.ResyncGroupAll)
      r.Put("/add/:xauth/:gid/:addr", api.GroupAddServer)
      r.Put("/add/:xauth/:gid/:addr/:datacenter", api.GroupAddServer)
      r.Put("/del/:xauth/:gid/:addr", api.GroupDelServer)
      r.Put("/promote/:xauth/:gid/:addr", api.GroupPromoteServer)
      r.Put("/replica-groups/:xauth/:gid/:addr/:value", api.EnableReplicaGroups)
      r.Put("/replica-groups-all/:xauth/:value", api.EnableReplicaGroupsAll)
      r.Group("/action", func(r martini.Router) {
        r.Put("/create/:xauth/:addr", api.SyncCreateAction)
        r.Put("/remove/:xauth/:addr", api.SyncRemoveAction)
      })
      r.Get("/info/:addr", api.InfoServer)
    })
    r.Group("/slots", func(r martini.Router) {
      r.Group("/action", func(r martini.Router) {
        r.Put("/create/:xauth/:sid/:gid", api.SlotCreateAction)
        r.Put("/create-some/:xauth/:src/:dst/:num", api.SlotCreateActionSome)
        r.Put("/create-range/:xauth/:beg/:end/:gid", api.SlotCreateActionRange)
        r.Put("/remove/:xauth/:sid", api.SlotRemoveAction)
        r.Put("/interval/:xauth/:value", api.SetSlotActionInterval)
        r.Put("/disabled/:xauth/:value", api.SetSlotActionDisabled)
      })
      r.Put("/assign/:xauth", binding.Json([]*models.SlotMapping{}), api.SlotsAssignGroup)
      r.Put("/assign/:xauth/offline", binding.Json([]*models.SlotMapping{}), api.SlotsAssignOffline)
      r.Put("/rebalance/:xauth/:confirm", api.SlotsRebalance)
    })
    r.Group("/sentinels", func(r martini.Router) {
      r.Put("/add/:xauth/:addr", api.AddSentinel)
      r.Put("/del/:xauth/:addr/:force", api.DelSentinel)
      r.Put("/resync-all/:xauth", api.ResyncSentinels)
      r.Get("/info/:addr", api.InfoSentinel)
      r.Get("/info/:addr/monitored", api.InfoSentinelMonitored)
    })
  })

可以看到dashboard有以下几块功能：

proxy管理：即/proxy开头的请求

group管理及slots管理，还有一些统计信息。

2)、Codis Fe

入口为cmd/fe下面的main.go，核心就启动2个路由处理： 

 r.Get("/list", func() (int, string) {
    names := router.GetNames()
    sort.Sort(sort.StringSlice(names))
    return rpc.ApiResponseJson(names)
  })

  r.Any("/**", func(w http.ResponseWriter, req *http.Request) {
    name := req.URL.Query().Get("forward")
    if p := router.GetProxy(name); p != nil {
      p.ServeHTTP(w, req)
    } else {
      w.WriteHeader(http.StatusForbidden)
    }
  })

其中/list请求是进入Codis Fe主界面就会调用的，用来展示当前有多少集群的；

还有一个路由是转发的，即如果参数中有forward参数，会把请求转到相应的Proxy来处理。

可以看到Proxy就两功能：一是显示所有Product Name，即当前有多少集群，第二就是请求转发，即将请求转到具体的Proxy。

3）、Codis Proxy

入口文件在 cmd/proxy/main.go， 

 s, err := proxy.New(config)

proxy.New会启动几个协程来处理任务： 

 go s.serveAdmin()
  go s.serveProxy()
  //下面是一些统计信息
  s.startMetricsJson()
  s.startMetricsInfluxdb()
  s.startMetricsStatsd()

重点关注前面2个，特别是第2个，s.serveProxy: 

 
go func(l net.Listener) (err error) {
    defer func() {
      eh <- err
    }()
    for {
      c, err := s.acceptConn(l)
      if err != nil {
        return err
      }
      NewSession(c, s.config).Start(s.router)
    }
  }(s.lproxy)

可以看到，Proxy就是不断的accept连接，然后启动一个Session。

总结下：Proxy的作用是处理客户端的连接，然后为每个连接建立一个Session，再根据客户端具体输入的命令进行处理，具体细节后续再聊。

相关文章