micro微服務 基礎元件的組織方式

簡介

micro是go語言實現的一個微服務框架，該框架自身實現了為服務常見的幾大要素，閘道器，代理，註冊中心，訊息傳遞，也支援可插拔擴充套件。本本通過micro中的一個核心物件展開去探討這個專案是如何實現這些元件並將其組織在一起工作的。

Notice: go程式碼有時候比較繁瑣，擷取原始碼的時候會刪除部分不影響思想的程式碼會標記為...

核心服務

micro通過micro.NewService建立一個服務例項，所有的微服務例項（包括閘道器，代理等等）需要通過這個例項來與其他服務打交道。

    type service struct {
        opts Options

        once sync.Once
    }


 

程式碼很少，但是micro中所有微服務要素最後都會匯聚在這個型別上了，因為option包羅永珍（^^）.

    type 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
        BeforeStop  []func() error
        AfterStart  []func() error
        AfterStop   []func() error

        // Other options for implementations of the interface
        // can be stored in a context
        Context context.Context
    }


 

這裡可以看到微服務的常見元件了，當micro.Server初始化的時候會給這個物件設定上對應的元件，元件的設定方式包括預設值，cli指定，env讀取。

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),
		)
	})
}


 

Server是最終的行為實體，監聽埠並提供業務服務，註冊本地服務到註冊中心。 Broker非同步訊息，mq等方法可以替換這個型別 Client服務呼叫客戶端 Registry 註冊中心 Cmd cli客戶端 Transport 類似與socket，訊息同步通訊，服務監聽等

服務型別

目前支援的服務型別有rpc,grpc。服務中如果存在兩種不同的rpc協議，訊息投遞的時候會進行協議轉換。這裡詳細說下預設的rpc服務。 rpc服務基於HTTP POST協議，服務啟動的時候會嘗試連線Broker，然後註冊本服務到註冊中心，最後監聽服務埠.簡單提一句這裡是如何做到協議轉換，如果http過來的訊息要投遞到一個grpc協議服務上，需要在Content-Type設定對應目的服務協議型別application/grpc，SerConn這裡讀取在Content-Type在獲取對應的Codec進行協議轉換，最後投遞到對應服務，服務的返回內容也同樣要處理下再返回。

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

rpc server

func (s *rpcServer) Start() error {
...
	ts, err := config.Transport.Listen(config.Address)
	// swap address
...
	// connect to the broker
	if err := config.Broker.Connect(); err != nil {
		return err
	}
...
	// use RegisterCheck func before register
	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 {
		// announce self to the world
		if err = s.Register(); err != nil {
			log.Logf("Server %s-%s register error: %s", config.Name, config.Id, err)
		}
	}
...
	go func() {
		for {
			// listen for connections
			err := ts.Accept(s.ServeConn)
...
		}
	}()
..
	return nil
}

協議轉換

    func (s *rpcServer) ServeConn(sock transport.Socket) {
...
        for {
            var msg transport.Message
            if err := sock.Recv(&msg); err != nil {
                return
            }
...
            // we use this Content-Type header to identify the codec needed
            ct := msg.Header["Content-Type"]

            // strip our headers
            hdr := make(map[string]string)
            for k, v := range msg.Header {
                hdr[k] = v
            }

            // set local/remote ips
            hdr["Local"] = sock.Local()
            hdr["Remote"] = sock.Remote()
...
            // TODO: needs better error handling
            var err error
            if cf, err = s.newCodec(ct); err != nil {  //請求協議轉換器
...返回錯誤
                return
            }
...
            rcodec := newRpcCodec(&msg, sock, cf)   //返回轉換器

            // internal request
            request := &rpcRequest{
                service:     getHeader("Micro-Service", msg.Header),
                method:      getHeader("Micro-Method", msg.Header),
                endpoint:    getHeader("Micro-Endpoint", msg.Header),
                contentType: ct,
                codec:       rcodec,
                header:      msg.Header,
                body:        msg.Body,
                socket:      sock,
                stream:      true,
            }

            // internal response
            response := &rpcResponse{
                header: make(map[string]string),
                socket: sock,
                codec:  rcodec,
            }

            // set router
            r := Router(s.router)
...
            // serve the actual request using the request router
            if err := r.ServeRequest(ctx, request, response); err != nil {
                // write an error response
                err = rcodec.Write(&codec.Message{
                    Header: msg.Header,
                    Error:  err.Error(),
                    Type:   codec.Error,
                }, nil)
                // could not write the error response
                if err != nil {
                    log.Logf("rpc: unable to write error response: %v", err)
                }
                if s.wg != nil {
                    s.wg.Done()
                }
                return
            }
...
        }
    }

註冊中心

註冊中心包括服務發現和服務註冊。micro中每個註冊中心型別都要實現註冊中心介面

    type Registry interface {
        Init(...Option) error
        Options() Options
        Register(*Service, ...RegisterOption) error
        Deregister(*Service) error
        GetService(string) ([]*Service, error)
        ListServices() ([]*Service, error)
        Watch(...WatchOption) (Watcher, error)
        String() string
    }

預設的註冊中心是mdns，該程式會在本地監聽一個組播地址接收網路中所有及其廣播的資訊，同時傳送的資訊也能被所有其他機器發現。每當程式啟動時都會廣播自己的服務資訊，其他節點收到該資訊後新增到自己的服務列表裡面，服務關閉時會發出關閉資訊。mdns自身不具備健康檢查，熔斷等功能，其出發點也僅是便於測試使用，因而不推薦在生產環境中使用。

Resolve

查詢服務需要根據url或者content資訊來獲取服務名稱，在通過服務名稱到註冊中心查詢，獲取到服務後，隨機一個節點投遞

    type Resolver interface {
        Resolve(r *http.Request) (*Endpoint, error)
        String() string
    }
    //預設的api resolve例項，除此之外還有host，path，grpc 三種resolve，可以根據需求在啟動程式的時候指定或者設定在環境變數裡面
    //
    func (r *Resolver) Resolve(req *http.Request) (*resolver.Endpoint, error) {
        var name, method string

        switch r.Options.Handler {
        // internal handlers
        case "meta", "api", "rpc", "micro":
        ///foo/bar/zool => go.micro.api.foo  方法:Bar.Zool
            ///foo/bar => go.micro.api.foo 方法:Foo.Bar
            name, method = apiRoute(req.URL.Path)
        default:
            //如果handler是web會走到這裡   
            ///foo/bar/zool => go.micro.api.foo  method bar/zool 
            // 1/foo/bar/ => go.micro.api.1.foo  method bar 
            method = req.Method
            name = proxyRoute(req.URL.Path)
        }

        return &resolver.Endpoint{
            Name:   name,
            Method: method,
        }, nil
    }

外掛

程式碼中定義了一個plugin，然而這個plugin並非是引入元件的作用，其作用在於在請求的管道中加一層，這樣比較容易新增新的邏輯進去。然而真正需要著重提到的是micro引入新元件的方式，micro service的幾個重要成員都有其對應的介面規約，只要正確實現了介面就可以輕鬆的接入新的元件。這裡用一個引入kubernetes作為註冊中心的例子。

kubernetes元件位於github.com\micro\go-plugins中

    go get -u github.com\micro\go-plugins\kubernetes

由於在go中無法實現java/c#中包動態載入功能，語言本身不提供全域性型別，函式掃描。往往只能通過一些曲折的辦法來實現。而micro是通過在入口檔案中匯入包，利用init函式在啟動時將需要的功能元件寫入到一個map裡面。 在import中加入外掛

    _ "github.com/micro/go-plugins/registry/kubernetes"

   micro api --registry=kubernetes  --registry_address=yourAddress

工作原理： 在github.com\micro\go-micro\config\cmd\cmd.go中有一個map用於儲存所有的註冊中心建立函式

	DefaultRegistries = map[string]func(...registry.Option) registry.Registry{
		"consul": consul.NewRegistry,
		"gossip": gossip.NewRegistry,
		"mdns":   mdns.NewRegistry,
		"memory": rmem.NewRegistry,
	}

kubernetes會在包的init中寫入對應的建立函式

    func init() {
        cmd.DefaultRegistries["kubernetes"] = NewRegistry
    }

生效的位置

    if name := ctx.String("registry"); len(name) > 0 && (*c.opts.Registry).String() != name {
		r, ok := c.opts.Registries[name]
		if !ok {
			return fmt.Errorf("Registry %s not found", name)
		}

		*c.opts.Registry = r()
		serverOpts = append(serverOpts, server.Registry(*c.opts.Registry))
		clientOpts = append(clientOpts, client.Registry(*c.opts.Registry))

		if err := (*c.opts.Selector).Init(selector.Registry(*c.opts.Registry)); err != nil {
			log.Fatalf("Error configuring registry: %v", err)
		}

		clientOpts = append(clientOpts, client.Selector(*c.opts.Selector))

		if err := (*c.opts.Broker).Init(broker.Registry(*c.opts.Registry)); err != nil {
			log.Fatalf("Error configuring broker: %v", err)
		}
	}

最後附上一張圖說明下這個核心的物件的引用關係,元件引用那一塊只是畫了註冊中心的，Broker，Server也都是類似的原理。

相關推薦