micro学习笔记：server

1、启动
启动例子：

    // 创建的时候会加载默认配置 
    service := micro.NewService()

    // 初始化配置，这里可以添加配置，或从flags中覆盖默认配置
    service.Init()

    // 注册服务
    hello.RegisterSayHandler(service.Server(), new(Say))

    // 启动
    if err := service.Run(); err != nil {
        log.Fatal(err)
    }

1.1、默认配置

func newOptions(opts ...Option) Options {
    opt := Options{
        Broker:    broker.DefaultBroker,
        Cmd:       cmd.DefaultCmd,
        Client:    client.DefaultClient,
        Server:    server.DefaultServer,
        Registry:  registry.DefaultRegistry,
        Transport: transport.DefaultTransport,
        Context:   context.Background(),
    }

    for _, o := range opts {
        o(&opt)
    }

    return opt
}

所有默认服务

DefaultBroker Broker = newHttpBroker()
DefaultCmd = newCmd()
DefaultClient Client = newRpcClient()
DefaultRegistry = newConsulRegistry()
DefaultTransport Transport = newHTTPTransport()
DefaultSelector = newDefaultSelector() // Strategy: Random,
DefaultServer  Server = newRpcServer()

服务器端启动流程

1、opts.BeforeStart 事件触发
2、opts.Server.Start() 开启server服务，go ts.Accept监听端口请求，会注册一个debugHandler提供Health，Stats方法，config.Broker.Connect()
3、opts.Server.Register() 注册服务到服务发现
4、opts.AfterStart 事件触发
5、go opts.Server.Register() 定时注册服务到服务发现
6、signal.Notify 监听signal退出
7、opts.BeforeStop 事件触发
8、opts.Server.Deregister() 从服务发现中剔除
9、opts.Server.Stop() 触发服务端退出ch, 等待请求结束，关闭Transport，Broker.Disconnect()
10、opts.AfterStop 事件触发

连接处理函数

func (s *rpcServer) accept(sock transport.Socket) {
    defer func() {
        // close socket
        sock.Close()

        if r := recover(); r != nil {
            log.Log("panic recovered: ", r)
            log.Log(string(debug.Stack()))
        }
    }()

    for {
        var msg transport.Message  // 取message
        if err := sock.Recv(&msg); err != nil {
            return
        }

        // we use this Timeout header to set a server deadline
        to := msg.Header["Timeout"]
        // we use this Content-Type header to identify the codec needed
        ct := msg.Header["Content-Type"]

        cf, err := s.newCodec(ct) // 初始化decoder
        // TODO: needs better error handling
        if err != nil {
            sock.Send(&transport.Message{
                Header: map[string]string{
                    "Content-Type": "text/plain",
                },
                Body: []byte(err.Error()),
            })
            return
        }

        codec := newRpcPlusCodec(&msg, sock, cf) // sock用于写错误

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

        ctx := metadata.NewContext(context.Background(), hdr)

        // set the timeout if we have it
        if len(to) > 0 {
            if n, err := strconv.ParseUint(to, 10, 64); err == nil {
                ctx, _ = context.WithTimeout(ctx, time.Duration(n))
            }
        }

        // add to wait group
        s.wg.Add(1)
        defer s.wg.Done()

        // TODO: needs better error handling  调用对应service方法
        if err := s.rpc.serveRequest(ctx, codec, ct); err != nil {
            log.Logf("Unexpected error serving request, closing socket: %v", err)
            return
        }
    }
}