微服务-08-服务管理

2021-08-03 Microservices Microservices 0 Comments

问题背景

之前云平台是基于MQ实现的通信机制，后来需要将通信方式切换为Grpc但又不想修改老的接口和自定义逻辑。所以就需要修改原本的库。因为内部包含了配置加载、MQ适配、Grpc功能、服务注册发现、链路追踪、自定义功能(限速、大包处理、调试、过滤、统计、存储…)。这些子服务是相互独立且可选(不启用也可以)，部分存在依赖关系。所以就想着使用服务管理的方式管理这些小功能，并按照期望的方式进行启动运行

实现方案

package main

import "context"

type ServerNum uint8

const (
	DemoA ServerNum = iota
	DemoB
    MaxServerNum
)


var serverList = []Server{
	DemoA: &Server1{},
	DemoB: &Server2{},
}

type Server interface {
	Start(ctx context.Context) (Server, error)	//服务启动
	ID() int								//返回服务标志，用于循序启动
	Close() error	//关闭服务
	IsOpen() bool	//是否开启
}

//开启服务
func Start(ctx context.Context) {
	for _, v := range serverList {
		v.Start(ctx)
	}
}

//服务管理
type Manager struct {

}

//服务启动入口
func (m *Manager)Start(ctx context.Context, s Server) error {
	if s == nil || !s.IsOpen() {	//未设置对象或未开启则直接退出
		return nil
	}
	v, err := s.Start(ctx)
	if err != nil {
		panic(err)
	}

	key := v.ID()	//构建k/v进行服务句柄存储
	//TODO 这里使用 封装的context
	return nil
}

func (m *Manager)Stop(ctx context.Context) {
	for i := MaxServerNum - 1; i >=0; i++ {
		v := serverList[i]
		if v != nil {
			v.Close()
		}
	}
}

单个服务内的子模块是否相互依赖的，所以这里使用切片进行存储，关闭时按照启动顺序的反方向
构建 IsOpen 方便根据配置加载服务需要的模块
服务句柄的存储暂时保存，需要使用服务句柄也许直接使用大写暴露会更好，或者使用不需要句柄的外部函数
注意服务的常驻等待(例如HTTP服务就是Listen之后不做后续处理，这个时候就需要使用协程拉起)，而所有服务加载完成之后，需要有一个常驻等待逻辑

func Loop() {
	exit := make(chan os.Signal, 1)
	signal.Notify(exit, syscall.SIGINT, syscall.SIGTERM)
	select {
		case sig := <- exit:
			log.Infof(context.Background(), "recv signal %s", sig.String())
	}
}

技术内幕

代码： core/service/servicegroup.go

import "github.com/zeromicro/go-zero/core/service"

// more code

func main() {
  group := service.NewServiceGroup()	//初始化管理对象
  defer group.Stop()									//停止
  group.Add(Morning{})								//添加服务morning
  group.Add(Evening{})								//添加服务evening
  group.Start()												//服务开启
}

数据结构

type (
	Starter interface {		//开启接口
		Start()
	}
  
	Stopper interface {		//停止接口
		Stop()
	}

	//服务接口，开启与停止
	Service interface {
		Starter
		Stopper
	}

	//服务批量管理接口
	ServiceGroup struct {
		services []Service	//切片保存服务
		stopOnce func()			//停止一次接口
	}
)

初始化对象

// NewServiceGroup returns a ServiceGroup.
func NewServiceGroup() *ServiceGroup {
	sg := new(ServiceGroup)
	sg.stopOnce = syncx.Once(sg.doStop)	//停止逻辑使用sync.Once 处理
	return sg
}

增加对象

//Add 服务管理中添加自定义服务
func (sg *ServiceGroup) Add(service Service) {
	// push front, stop with reverse order.
	sg.services = append([]Service{service}, sg.services...)
}

开启服务

//Start 服务管理启动
func (sg *ServiceGroup) Start() {
	proc.AddShutdownListener(func() {		//将服务添加到退出监听中
		log.Println("Shutting down...")
		sg.stopOnce()
	})

	sg.doStart()
}

//doStart 开始
func (sg *ServiceGroup) doStart() {
	routineGroup := threading.NewRoutineGroup()

	for i := range sg.services {
		service := sg.services[i]
    routineGroup.RunSafe(func() {	//使用每个服务使用协程开启(包含panic处理)
			service.Start()
		})
	}

	routineGroup.Wait()	//等待退出
}

使用协程池进行每个服务的启动，所以每个服务启动的顺序是不一定的
使用RunSafe进行服务启动，所以一个服务panic，另外的服务也能启动

服务关闭

// Stop stops the ServiceGroup.
func (sg *ServiceGroup) Stop() {NewCache
   sg.stopOnce()
}

//停止服务管理(服务初始化的时候已经设置好了)
func (sg *ServiceGroup) doStop() {
	for _, service := range sg.services {
		service.Stop()
	}
}

仅有开启服务

// WithStart wraps a start func as a Service.
func WithStart(start func()) Service {
	return startOnlyService{
		start: start,
	}
}

// WithStarter wraps a Starter as a Service.
func WithStarter(start Starter) Service {
	return starterOnlyService{
		Starter: start,
	}
}

type (
	stopper struct{}

	startOnlyService struct {
		start func()
		stopper
	}

	starterOnlyService struct {
		Starter
		stopper
	}
)

func (s stopper) Stop() {
}

func (s startOnlyService) Start() {
	s.start()
}

总结

由于子服务无法保证启动顺序，所以各服务之间不能有相互依赖或关联
使用 stopOnce 防止出现多次调用stop的情况
使用 proc.AddShutdownListener做退出监听

优化：

如果能控制服务启动和调用顺序是否会更好(一个服务内的多个子模块必然存在先后启动关系)，那么停止服务的时候也注意先后关系
如果使用协程池拉起服务，那么单个服务的异常无法让所有子服务都退出。缺少一个同步退出机制
使用 startOnlyService是否有点多余，在服务停止中不做任何处理，那么退出时也能正常退出

参考文档

https://mp.weixin.qq.com/s/G6WG_-C6d-raoRmH4hBjoQ

Yuan Kang

Learn and live