Codis源码分析之环境篇

一、Codis介绍

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

https://github.com/CodisLabs/...

以下是官方介绍:

Codis 是一个分布式 Redis 解决方案, 对于上层的应用来说, 连接到 Codis Proxy 和连接原生的 Redis Server 没有显著区别 (不支持的命令列表), 上层应用可以像使用单机的 Redis 一样使用, Codis 底层会处理请求的转发, 不停机的数据迁移等工作, 所有后边的一切事情, 对于前面的客户端来说是透明的, 可以简单地认为后边连接的是一个内存无限大的 Redis 服务。

https://github.com/CodisLabs/...

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

二、Codis安装及配置

1、整体架构

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

我们看下Codis的整体架构,基于3.2:

Codis源码分析之环境篇

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整体代码结构:

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 failedn%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()

这个函数内部会启动一个apiServer,以下是具体的路由:

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主界面就会调用的,用来展示当前有多少集群的;

Codis源码分析之环境篇

还有一个路由是转发的,即如果参数中有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,再根据客户端具体输入的命令进行处理,具体细节后续再聊。

image.png

你可能感兴趣的