发表更新Golang随记23 分钟读完 (大约3389个字)
 0

跟着B站学Golang(admin-ep-saga)

本文完全出于学习的目的，如有异议，请联系删除。

之前XL事件流出的优秀代码太多了，这次选择的是一个好像与具体业务无关的模块(admin-ep-saga)来进行学习。

首先看看目录结构：

目录结构

这么多先看哪一个呢？在不知道具体每个包是干什么的情况下，只好一个一个的看了。

先看看api下有些什么：

api

不得不说这个目录的结构相当规范呀，虽然我没有点开具体文件，但是仅仅从目录名和文件名就能猜出这个目录下是干什么的：

应该是使用了grpc框架和protobuf协议定义的接口。这个暂时先放一边，我需要先找到程序入口，这样才能一步一步的学习优秀代码是如何编写的。

接下来打开cmd:

cmd

这个目录下有三个文件：

  1. BUILD看着应该是用来做构建用的，这不是我这次学习的重点，先跳过。
  2. saga-admin-test.toml 我打开看了一眼，是一个配置文件，从名字可以看出应该是测试用的配置项，后面还会碰到。
  3. main.go 如果不出意外，这个应该就是程序入口了，运气还不错，第二个目录就找到了入口。

接下来详细的看看main.go做了些什么事情：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
/*
这里我忽略了一些包导入，以及一些常量
因为如果每个导入的包都要看的话，会越陷越深。
*/
func main() {
    //解析命令行参数
	flag.Parse()
    //初始化配置
	if err := conf.Init(); err != nil {
		log.Error("conf.Init() error(%v)", err)
		panic(err)
	}
	//初始化Log
	log.Init(conf.Conf.Log)
	defer log.Close()
	log.Info("saga-admin start")
	//启动一个服务
	s := service.New()
	http.Init(s)
    //启动一个grpc服务
	grpcsvr, err := grpc.New(nil, s.Wechat())
	if err != nil {
		panic(err)
	}
    //创建一个长度为1的os.Signal类型的channel
	c := make(chan os.Signal, 1)
    //通知
	signal.Notify(c, syscall.SIGHUP, syscall.SIGQUIT, syscall.SIGTERM, syscall.SIGINT)
	for {
        //从前面创建的channel中读取signal
		si := <-c
		log.Info("saga-admin get a signal %s", si.String())
		switch si {
        //如果是SIGQUIT、SIGTERM、SIGINT则关闭相关服务，然后退出
		case syscall.SIGQUIT, syscall.SIGTERM, syscall.SIGINT:
			grpcsvr.Shutdown(context.Background())
			log.Info("saga-admin exit")
			s.Close()
			time.Sleep(_durationForClosingServer * time.Second)
			return
		case syscall.SIGHUP:
		default:
			return
		}
	}
}

粗略的看了一下代码后，带着疑问，一行一行的来分析，首先第一行：

1
flag.Parse()

作为Golang小白，我知道这个应该是使用在flag.StringVar这样的代码后面，定义需要获取的命令行参数。

但是flag.Parse()作为第一行代码前面并没有flag.StringVar类似这样的代码呀，然后我想到了Golang中init函数的作用。于是我开始找main.go中导入的其他包中有没有定义init函数，果不其然，在saga/conf/conf.go中我找到了：

1
2
3
4
5
6
func init() {
    //定义一个命令行参数，用来接收配置文件路径
	flag.StringVar(&confPath, "conf", "", "config path")
    //这个reload后面再讲
	reload = make(chan bool, 10)
}

回到main.go来看下面几行代码：

1
2
3
4
if err := conf.Init(); err != nil {
    log.Error("conf.Init() error(%v)", err)
    panic(err)
}

忽略错误判断以及日志打印，我们可以看到这几行中最关键的代码就是conf.Init()，这个代码做了些什么事情呢？接下来进入saga/conf/conf.go：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
func Init() (err error) {
    //判断如果配置文件的路径为空，则执行configCenter()方法
	if confPath == "" {
		return configCenter()
	}
    //如果配置文件路径不为空通过toml.DecodeFile(confPath, &Conf)解析配置到&Conf中
	if _, err = toml.DecodeFile(confPath, &Conf); err != nil {
		log.Error("toml.DecodeFile(%s) err(%+v)", confPath, err)
		return
	}
    //单独解析TeamInfo相关配置
	Conf = parseTeamInfo(Conf)
	return
}

首先看：

1
2
3
if confPath == "" {
	return configCenter()
}

从函数名可以看出，当没有手动指定配置文件路径是，走配置中心解析配置。configCenter我们先放一放，我们接着往下看：

1
2
3
4
if _, err = toml.DecodeFile(confPath, &Conf); err != nil {
	log.Error("toml.DecodeFile(%s) err(%+v)", confPath, err)
	return
}

跟之前一样，我们忽略错误和日志处理，可以看到这几行关键代码是toml.DecodeFile(confPath, &Conf)，

toml这个看着是不是很眼熟，之前在cmd包下我们看到过一个这个格式的文件saga-admin-test.toml，这是一个由GitHub联合创始人Tom Preston-Werner 搞出的极简配置文件格式。各个语言都有相关实现，BZ这里使用的是github.com/BurntSushi/toml这个库。

总而言之，这几行代码无非就是解析配置文件。

继续往下看:

1
Conf = parseTeamInfo(Conf)

单独用了一个方法来解析TeamInfo说明toml标准的DecodeFile解析不了，我们来看看这个方法做了什么事情：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
/*
这个方法做的事情比较简单，直接采用注释的方法讲解
*/
func parseTeamInfo(c *Config) *Config {
	/*
	strings.Fields，我们都知道这个是根据字符串中的空格或者一些个特殊符号来拆分字符串为Array的方法。
	我们来看看之前提到的saga-admin-test.toml中c.Property.Department 定义的是什么：
	[property.department]
        label = "主站 直播 bplus 开放平台 创作中心 商业产品 数据中心 视频云 游戏 火鸟"
        value = "mainsite live bplus openplatform creative advertising datacenter videocloud game firebird"
	*/
	DeLabel := strings.Fields(c.Property.Department.Label)
	DeValue := strings.Fields(c.Property.Department.Value)
	for i := 0; i < len(DeLabel); i++ {
		/*
		所以这几行代码，很显而易见了，就是将上述的label 和value组合成key-value的形式然后append到另外一个(DeInfo)Array中
		*/
		info := &model.PairKey{
			Label: DeLabel[i],
			Value: DeValue[i],
		}
		c.Property.DeInfo = append(c.Property.DeInfo, info)
	}
	//下面几行代码同上，就不在赘述
	buLabel := strings.Fields(c.Property.Business.Label)
	buValue := strings.Fields(c.Property.Business.Value)
	for i := 0; i < len(buLabel); i++ {

		info := &model.PairKey{
			Label: buLabel[i],
			Value: buValue[i],
		}
		c.Property.BuInfo = append(c.Property.BuInfo, info)
	}
	return c
}

话说，上面的Business和Department处理过程一样，为啥不将这个过程提取成一个函数呢？来自小白的疑问。

还记得我们之前跳过一个函数configCenter()吗？接下来我们一起来看看：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
func configCenter() (err error) {
    //这里的conf应该是BZ一个公共组件，这里做的就是创建一个配置中心的client
	if client, err = conf.New(); err != nil {
		panic(err)
	}
    //这里调用了load函数
	if err = load(); err != nil {
		return
	}
    //这里应该是添加了一个配置中心的监听
	client.WatchAll()
    //起一个goroute
	go func() {
        //获取事件，如果配置中心的配置存在修改重新调用load函数
		for range client.Event() {
			log.Info("config reload")
			if load() != nil {
				log.Error("config reload error (%v)", err)
			} else {
                //load成功往reload chan写入一个数据，这里有个疑问，等后面再说
				reload <- true
			}
		}
	}()
	return
}

忽略其它代码，可以看到上面实现配置中心配置加载的应该是load函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
func load() (err error) {
    //定义一组局部变量
	var (
		s       string
		ok      bool
		tmpConf *Config
	)
    //通过配置中心的client获取配置，这里的_configkey是常量："saga-admin.toml"
	if s, ok = client.Value(_configKey); !ok {
		err = errors.Errorf("load config center error [%s]", _configKey)
		return
	}
    //跟之前一样通过toml解析配置
	if _, err = toml.Decode(s, &tmpConf); err != nil {
		err = errors.Wrapf(err, "could not decode config err(%+v)", err)
		return
	}
    //跟之前一样单独解析TeamInfo
	Conf = parseTeamInfo(tmpConf)
	return
}

到这里我们差不多刚刚看完main.go中conf.Init()的调用，接下来回到main.go，继续往下看：

我们跳过Log的初始化，直接看：

1
2
3
4
//调用/saga/service中的New函数
s := service.New()
//调用/saga/http中的Init函数
http.Init(s)

跳转到New函数中，我们看看做了些什么：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
func New() (s *Service) {
	var (
		err error
	)
	s = &Service{
		dao:  dao.New(),
		cron: cron.New(),
	}
	if err = s.cron.AddFunc(conf.Conf.Property.SyncProject.CheckCron, s.collectprojectproc); err != nil {
		panic(err)
	}
	if err = s.cron.AddFunc(conf.Conf.Property.Git.CheckCron, s.alertProjectPipelineProc); err != nil {
		panic(err)
	}

	if err = s.cron.AddFunc(conf.Conf.Property.SyncData.CheckCron, s.syncdataproc); err != nil {
		panic(err)
	}
	if err = s.cron.AddFunc(conf.Conf.Property.SyncData.CheckCronAll, s.syncalldataproc); err != nil {
		panic(err)
	}
	if err = s.cron.AddFunc(conf.Conf.Property.SyncData.CheckCronWeek, s.syncweekdataproc); err != nil {
		panic(err)
	}
	s.cron.Start()

	// init gitlab client
	s.gitlab = gitlab.New(conf.Conf.Property.Gitlab.API, conf.Conf.Property.Gitlab.Token)
	// init online gitlab client
	s.git = gitlab.New(conf.Conf.Property.Git.API, conf.Conf.Property.Git.Token)
	// init wechat client
	s.wechat = wechat.New(s.dao)

	return
}

上面代码大部分都是在做定时任务的创建，cron使用的是"github.com/robfig/cron"这个库，我们挑一个看看：

1
2
3
if err = s.cron.AddFunc(conf.Conf.Property.SyncProject.CheckCron, s.collectprojectproc); err != nil {
		panic(err)
}

conf.Conf.Property.SyncProject.CheckCron 是配置中的cron表达式，在saga-admin-test.toml中看到是

* */15 * * * ? 也就是说这个任务每15分钟执行一次。

s.collectprojectproc 是要执行的任务，接下来看看这个任务做了什么事情：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
func (s *Service) collectprojectproc() {
	var err error
    //可以看到实际调用的是CollectProject，这里的context.TODO()表示context还未实现，这里仅仅用作占位，没有实际意义
	if err = s.CollectProject(context.TODO()); err != nil {
		log.Error("s.CollectProject err (%+v)", err)
	}
}

func (s *Service) CollectProject(c context.Context) (err error) {
    //这是一组局部变量
	var (
		projects []*gitlab.Project
		total    = 0
		page     = 1
	)

	log.Info("Collect Project start")
    //这里出现了一个magic number，1000
	for page <= 1000 {
		//调用gitlab接口获取指定页码的项目列表，这里的s.gitlab是在service.New()中实例化的。这里有一个疑问。
		if projects, err = s.gitlab.ListProjects(page); err != nil {
			return
		}

		num := len(projects)
		if num <= 0 {
			break
		}
		total = total + num
		//将获取到的项目保存到db中，这个insertDB就不展开讲了，这里面做的大概就是saveOrUpdate的事情。
		for _, p := range projects {
			if err = s.insertDB(p); err != nil {
				return
			}
		}

		page = page + 1
	}
	log.Info("Collect Project end, find %d projects", total)

	return
}

在service.New()中其他的cron也差不多是做着类似的事情，由于太多，就不在这里一一展开。刚刚说到

s.gitlab.ListProjects(page);我有一个疑问，是什么呢？我们看这里：

1
2
3
4
5
6
7
8
s.cron.Start()
// init gitlab client
s.gitlab = gitlab.New(conf.Conf.Property.Gitlab.API, conf.Conf.Property.Gitlab.Token)
// init online gitlab client
s.git = gitlab.New(conf.Conf.Property.Git.API, conf.Conf.Property.Git.Token)
// init wechat client
s.wechat = wechat.New(s.dao)

可以发现cron的start是在gitlab、git、wechat实例化之前，而cron相关的任务中又依赖了这些client，那有没有这么一种可能：这个程序启动的时候正好碰上cron触发，而gitlab，wechat这些client还没有实例化，所以有没有可能出现panic？当然了，这个可能性很小。

让我们再次回到main.go中：

1
2
//初始化一个http服务
http.Init(s)

进入Init:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// Init init
func Init(s *service.Service) {
    //这个srv很重要，这是在上面service.New()最后返回的实例，在后面经常用到
	srv = s
    //这个permit是go-common/library/net/http/blademaster 中的组件，应该是用来做接口认证的
	authSvc = permit.New2(nil)
	//下面就是启动http engine了，这个engine就是上面提到的这个blademaster
	engine := bm.DefaultServer(conf.Conf.BM)
	engine.Ping(ping)
	initRouter(engine)
	if err := engine.Start(); err != nil {
		log.Error("engine.Start error(%v)", err)
		panic(err)
	}
}

由于这个http服务是依赖的BZ公共的组件，就不继续深入了，我怕出不来了。我们看看initRouter中定义的Router，由于太长，我只选择开始一段：

1
2
3
4
5
6
7
8
9
version := e.Group("/ep/admin/saga/v1", authSvc.Permit2(""))
	{
		project := version.Group("/projects")
		{
			project.GET("/favorite", favoriteProjects)
			project.POST("/favorite/edit", editFavorite)
			project.GET("/common", queryCommonProjects)
		}
...

这就是很常见的url-mapping了，这里的favoriteProjects、editFavorite、queryCommonProjects 都是定义在当前http包下的函数，我们选择favoriteProjects看下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
func favoriteProjects(ctx *bm.Context) {
	var (
		req      = &model.Pagination{}
		err      error
		userName string
	)
    //这里应该是解析请求参数到req变量中
	if err = ctx.Bind(req); err != nil {
		ctx.JSON(nil, err)
		return
	}
    //这里是调用函数获取当前用户名
	if userName, err = getUsername(ctx); err != nil {
		ctx.JSON(nil, err)
		return
	}
    //我们看到这里最终回到了srv上，调用了实际的处理方法。这里的FavoriteProjects实际就是通过db查询当前用户收藏的项目，我们就不继续深入了。
	ctx.JSON(srv.FavoriteProjects(ctx, req, userName))
}

上面的流程大概是这样的，首先在http包中将参数等一些信息进行解析，最后调用到了service中的方法。这个就很像Java中流行的写法：从Controller到Service，符合MVC分层的思想。

最后还有一个grpc，我大概看了下，应该是用来企业微信发消息的。

总结：

这应该是用来做gitlab ci告警之类的project，可以看到这个project层次很分明，代码看过去一目了然。

最后，之前还有一个疑问，就是reload这个变量，在初始化时是有长度的：

1
reload = make(chan bool, 10)

长度为10，在每次配置文件修改后，goroute watch到event之后会往这个channel写入一个true，但是看完整个代码之后，并没有看到有地方从这个channel取出数据（也有可能是我漏看了），也就是说当修改10次之后，这个地方：

1
reload <- true

就会阻塞，从而导致这个goroute无响应？

码字不易，且转且珍惜。

跟着B站学Golang(admin-ep-saga)

https://jingzhouzhao.github.io/archives/f11de5dd.html

作者

太阳当空赵先生

发布于

2019-09-05

更新于

2022-02-22

许可协议

#

喜欢这篇文章？打赏一下作者吧

评论

关注我

链接

分类

归档

最新文章

标签

ASM1
Activiti1
BUG2
Bug1
CAP1
CPU1
Chrome1
DeadLock1
Deadlock1
Docker3
Dubbo3
EPUB1
Golang8
IDEA1
IRIW1
Java27
Lombok1
Maven1
MessageId1
MongodDB2
MySQL5
Mybatis1
OCR1
PDF1
Paxos1
Python3
RPC1
RocketMQ2
Spotlight1
Spring2
UniqKey1
VSCode1
dubbo1
elasticsearch1
flume1
kafka2
launchctl1
linux1
uTools1
volatile1
一致性1
博客1
双亲委派1
可用内存1
图床1
定时任务1
并发1
开源1
效率1
死锁1
爬虫1
电子书1

follow.it

×