Go Micro Server 源码分析

概述

在Go Micro框架中，Server是对Broker、Register、Codec、Transort等服务的一个封装，从下图中就可以看到。

再看一下Server定义的接口

• Init：初始化
• Handler：注册rpchandler
• NewHandler：封装rpchandler
• NewSubscriber：封装Subscriber给Subscribe方法
• Subscribe：//注入订阅事件
• Start：启动服务、监听端口、处理请求
• Stop：停止服务，broker等资源关闭
• String

type Server interface {
    Options() Options
    Init(...Option) error 
    Handle(Handler) error
    NewHandler(interface{}, ...HandlerOption) Handler
    NewSubscriber(string, interface{}, ...SubscriberOption) Subscriber
    Subscribe(Subscriber) error
    Start() error
    Stop() error
    String() string

主要方法 源码 以rpcServer为例

Init

server的初始化方法，初始化options之后 利用once函数初始化了Cmd

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

Handle

调用 router的handler方法把方法注册到server中

func (s *rpcServer) Handle(h Handler) error {
    s.Lock()
    defer s.Unlock()

    if err := s.router.Handle(h); err != nil {
        return err
    }

    s.handlers[h.Name()] = h

    return nil
}

NewSubscriber

订阅某个主题，调用subscriber中的newSubscriber

newSubscriber函数我们在 Go Micro Broker源码分析
文章中已经分析过这里就不展开了

func (s *rpcServer) NewSubscriber(topic string, sb interface{}, opts ...SubscriberOption) Subscriber {
    return newSubscriber(topic, sb, opts...)
}

Subscribe

Subscribe函数接受一个Subscriber的接口，下面可以看到接口的定义
Subscribe函数接受到接口之后保存到rpcServer中的subscribers map中
在Deregister、Register、Subscribe方法中调用注册的Subscriber() Unsubscribe()函数

type Subscriber interface {
    Topic() string
    Subscriber() interface{}
    Endpoints() []*registry.Endpoint
    Options() SubscriberOptions
}
// broker.go 
type Subscriber interface {
    Options() SubscribeOptions
    Topic() string
    Unsubscribe() error
}

func (s *rpcServer) Subscribe(sb Subscriber) error {
    sub, ok := sb.(*subscriber)
    if !ok {
        return fmt.Errorf("invalid subscriber: expected *subscriber")
    }
    if len(sub.handlers) == 0 {
        return fmt.Errorf("invalid subscriber: no handler functions")
    }

    if err := validateSubscriber(sb); err != nil {
        return err
    }

    s.Lock()
    defer s.Unlock()
    _, ok = s.subscribers[sub]
    if ok {
        return fmt.Errorf("subscriber %v already exists", s)
    }
    s.subscribers[sub] = nil
    return nil
}

Start

start函数是server中最重要的函数，在service run的时候回调用server的start方法来正式开启这个服务。
代码如下

在 Go Micro Register 源码分析
这篇文章中已经分析过Start函数的中对于服务发现注册的项管代码这边就不重复了。主要来看一下的是 监听端口、处理清酒、和初始化broker
，总体来说就是实现了监听端口/处理请求/初始化Broker。

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

    // 开始监听transport 处理客户端接收到的请求
    ts, err := config.Transport.Listen(config.Address)
    if err != nil {
        return err
    }

    log.Logf("Transport [%s] Listening on %s", config.Transport.String(), ts.Addr())

    // swap address
    s.Lock()
    addr := s.opts.Address
    s.opts.Address = ts.Addr()
    s.Unlock()

    // 连接broker 用于订阅服务
    if err := config.Broker.Connect(); err != nil {
        return err
    }

    bname := config.Broker.String()

    log.Logf("Broker [%s] Connected to %s", bname, config.Broker.Address())

    // 调用RegisterCheck检查注册服务是否可用 可从外部注入函数
    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)
    // 处理上面监听的Transport.Listen listerer
    go func() {
        for {
            // 监听到请求处理请求代码 在Transport会深入详细说明
            err := ts.Accept(s.ServeConn)

            select {
            // check if we're supposed to exit
            case  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()
                // 调用Register组件中的RegisterCheck方法 测试服务是否正常
                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
            }
        }

        // deregister self
        if err := s.Deregister(); err != nil {
            log.Logf("Server %s-%s deregister error: %s", config.Name, config.Id, err)
        }

        // wait for requests to finish
        if s.wg != nil {
            s.wg.Wait()
        }

        // close transport listener
        ch <- ts.Close()

        // disconnect the broker
        config.Broker.Disconnect()

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

    return nil
}

总结

server是对于底层一些方法的封装，比如实现服务的开启、关闭、节点的注册和订阅的注册。同样级别的封装在最上面图可以看到还有CLient，然而在Client和Server上层还有service，这些会在另外的地方再分析。