golang微服務框架go-micro 入門筆記2.4 go-micro service解讀

時間  2019-11-16
標籤 golang 微服 框架 micro 入門 筆記 2.4 service 解讀 欄目 Go 简体版
原文   原文鏈接

本章節闡述go-micro 服務發現原理golang

go-micro架構

下圖來自go-micro官方 web

go-micro系統架構

閱讀本文前你可能須要進行以下知識儲備

Service

實現以下接口的,都是Serviceapi

type Service interface {
	Init(...Option)
	Options() Options
	Client() client.Client
	Server() server.Server
	Run() error
	String() string
}
複製代碼

咱們根據常見的service初始化流程來理解這個接口bash

func main() {
	// New Service
	service := micro.NewService(
		micro.Name("go.micro.srv.broker"),
		micro.Version("latest"),
	)

	// Initialise service
	service.Init()

	// Register Struct as Subscriber
	micro.RegisterSubscriber("go.micro.srv.broker", service.Server(), new(subscriber.Broker))

	// Register Function as Subscriber
	micro.RegisterSubscriber("go.micro.srv.broker", service.Server(), subscriber.Handler)

	// Run service
	if err := service.Run(); err != nil {
		log.Fatal(err)
	}
}
複製代碼

micro.NewService

micro.NewService的函數調用關係以下 微信

micro.NewService的函數調用關係以下
該函數主要是初始化一個service並設置service的經常使用屬性屬性,代碼以下: 

opt := micro.Options{
	Broker: broker.DefaultBroker,
	Cmd: cmd.DefaultCmd,
	Client: client.DefaultClient,
	Server: server.DefaultServer,
	Registry: registry.DefaultRegistry
	}
複製代碼

micro.Options所有屬性包含以下幾個方面網絡

type micro.Options struct {
	Broker    broker.Broker  //訂閱事件用的
	Cmd       cmd.Cmd       //命令行用的
	Client    client.Client  //做爲服務消費者,用來發送請求
	Server    server.Server  //做爲服務提供者用來監聽請求
	Registry  registry.Registry //這個是服務發現支持
	Transport transport.Transport //這是網絡傳遞用的

	// Before and After funcs 
	BeforeStart []func() error //服務start錢的回調函數 BeforeStop []func() error //服務stop後的回調函數 AfterStart []func() error //服務start後的回調函數 AfterStop []func() error //服務stop後的回調函數 // Other options for implementations of the interface // can be stored in a context Context context.Context //有時候服務須要攜帶一些參數,須要規定時間內返回，能夠用這個來承載 } 複製代碼

咱們能夠經過micro.Options.Server.Options得到server的屬性,實際上server還有其餘的屬性,server的所有參數屬性以下架構

type server.Options struct {
	Codecs       map[string]codec.NewCodec //訂閱事件用的
	Broker       broker.Broker             //訂閱事件用的
	Registry     registry.Registry        //這個是服務發現支持
	Transport    transport.Transport     //節點之間網絡傳遞用的
	Metadata     map[string]string       //傳遞一些k,v數據用
	Name         string                  //名稱
	Address      string                 //服務地址
	Advertise    string                  //廣播地址
	Id           string                //服務ID
	Version      string                 //處理程序wrapper
	HdlrWrappers []HandlerWrapper            //網絡處理程序封裝
	SubWrappers  []SubscriberWrapper        //訂閱程序封裝

	// RegisterCheck runs a check function before registering the service
	RegisterCheck func(context.Context) error //註冊服務前的校驗函數 // The register expiry time RegisterTTL time.Duration //註冊服務生命週期,過了這麼長事件須要從新註冊 // The interval on which to register RegisterInterval time.Duration //多長時間註冊一次 // The router for requests Router Router //請求的路由表 // Other options for implementations of the interface // can be stored in a context Context context.Context //有時候服務須要攜帶一些參數,須要規定時間內返回，能夠用這個來承載 } 複製代碼

若是server須要配置額外的屬性,則調用屬性配置函數來刷新配置。如Name()運行結果是返回一個函數,所以最後經過server.Name(name)來刷新這個函數,app

micro.Name("go.micro.srv.broker"),
// Name of the service
func Name(n string) Option {
	return func(o *Options) {
		o.Server.Init(server.Name(n))
	}
}
複製代碼

函數關係以下 框架

micro.NewService的函數調用關係以下

注意最後的o.Name即設置service.Options.Name = name micro.Version("latest")也是同樣的分佈式

service.Init

init 函數原型以下

func (s *service) Init(opts ...Option) {
	// process options
	for _, o := range opts {
		o(&s.opts)
	}

	s.once.Do(func() {
		// Initialise the command flags, overriding new service
		_ = s.opts.Cmd.Init(
			cmd.Broker(&s.opts.Broker),
			cmd.Registry(&s.opts.Registry),
			cmd.Transport(&s.opts.Transport),
			cmd.Client(&s.opts.Client),
			cmd.Server(&s.opts.Server),
		)
	})
}
複製代碼

for _, o := range opts代碼片斷的做用是設置service的屬性,底層也是經過 相似o.Server.Init(server.NaME(n)來實現的。

s.once.Do(func) 這裏利用了sync.Once特性,這裏預示着func只會運行1次。

micro.NewService的函數調用關係以下

兩行傳入的參數經過c := a.Command(name)來得到須要運行的命令行

service.Run

源代碼以下

func (s *service) Run() error {
	// 這一段代碼的做用其實只開啓調試支持
	s.opts.Server.Handle(
		s.opts.Server.NewHandler(
			handler.DefaultHandler,//默認處理腳本handler就是調試
			server.InternalHandler(true),//把這個標記爲成內置的handler
		),
	)

	if err := s.Start(); err != nil {
		return err
	}

	ch := make(chan os.Signal, 1)
	signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT, syscall.SIGQUIT)

	select {
	// wait on kill signal
	case <-ch:
	// wait on context cancel
	case <-s.opts.Context.Done():
	}

	return s.Stop()
}
複製代碼

s.Start()做用是啓動服務,目前micro支持的rpcservce/httpservice/apiservice/websercice等,他們都實現了這個Start()方法。以RPCService爲例,函數流程以下

micro.NewService的函數調用關係以下

具體代碼解讀以下

func (s *rpcServer) Start() error {
	config := s.Options()

	//啓動監聽
	ts, err := config.Transport.Listen(config.Address)
	if err != nil {
		return err
	}


	// 鏈接消息代理
	if err := config.Broker.Connect(); err != nil {
		return err
	}


	// 註冊發現,嚴格校驗註冊服務,
	if err = s.opts.RegisterCheck(s.opts.Context); err != nil {
		log.Logf("Server %s-%s register check error: %s", config.Name, config.Id, err)
	} else {
		// 註冊發現,匿名註冊
		if err = s.Register(); err != nil {
			log.Logf("Server %s-%s register error: %s", config.Name, config.Id, err)
		}
	}

	exit := make(chan bool)

	//啓動監聽服務
	go func() {
		for {
			// 監聽
			err := ts.Accept(s.ServeConn)

			// TODO: listen for messages
			// msg := broker.Exchange(service).Consume()

			select {
			// check if we're supposed to exit
			case <-exit:
				return
			// check the error and backoff
			default:
				if err != nil {
					log.Logf("Accept error: %v", err)
					time.Sleep(time.Second)
					continue
				}
			}

			// no error just exit
			return
		}
	}()

	//開啓定時註冊攜程
	go func() {
		t := new(time.Ticker)

		// only process if it exists
		if s.opts.RegisterInterval > time.Duration(0) {
			// new ticker
			t = time.NewTicker(s.opts.RegisterInterval)
		}

		// return error chan
		var ch chan error

	Loop:
		for {
			select {
			// register self on interval
			case <-t.C:
				s.RLock()
				registered := s.registered
				s.RUnlock()
				//校驗註冊
				if err = s.opts.RegisterCheck(s.opts.Context); err != nil && registered {
					log.Logf("Server %s-%s register check error: %s, deregister it", config.Name, config.Id, err)
					// deregister self in case of error
					if err := s.Deregister(); err != nil {
						log.Logf("Server %s-%s deregister error: %s", config.Name, config.Id, err)
					}
				} else {
					//匿名註冊
					if err := s.Register(); err != nil {
						log.Logf("Server %s-%s register error: %s", config.Name, config.Id, err)
					}
				}
			// wait for exit
			case ch = <-s.exit:
				t.Stop()
				close(exit)
				break Loop
			}
		}

		// 若是退出來了,那麼註銷服務
		if err := s.Deregister(); err != nil {
			log.Logf("Server %s-%s deregister error: %s", config.Name, config.Id, err)
		}

		// 等到請求完成
		if s.wg != nil {
			s.wg.Wait()
		}

		// 關閉
		ch <- ts.Close()

		// 關閉消息代理鏈接
		config.Broker.Disconnect()

		// swap back address
		s.Lock()
		s.opts.Address = addr
		s.Unlock()
	}()

	return nil
}
複製代碼

相關代碼得到

關注公衆號回覆micro-broker便可得到

公衆號betaidea

推薦閱讀

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程