grpc Server初始化注册监听源码

古城微笑少年丶 2022-12-09 03:12 230阅读 0赞

func main() {
        
    
    	server := grpc.NewServer()
    	pb.RegisterSearchServiceServer(server, &SearchService{
        })
    
    	lis, err := net.Listen("tcp", ":"+PORT)
    
    	server.Serve(lis)
    }

为什么四行代码，就能够起一个 gRPC Server

## 初始化 ##

// grpc.NewServer()
    func NewServer(opt ...ServerOption) *Server {
        
    	opts := defaultServerOptions
    	for _, o := range opt {
        
    		o(&opts)
    	}
    	s := &Server{
        
    		lis:    make(map[net.Listener]bool),
    		opts:   opts,
    		conns:  make(map[io.Closer]bool),
    		m:      make(map[string]*service),
    		quit:   make(chan struct{
        }),
    		done:   make(chan struct{
        }),
    		czData: new(channelzData),
    	}
    	s.cv = sync.NewCond(&s.mu)
    	...
    
    	return s
    }

1.  lis：监听地址列表。
2.  opts：服务选项，这块包含 Credentials、Interceptor 以及一些基础配置。
3.  conns：客户端连接句柄列表。
4.  m：服务信息映射。
5.  quit：退出信号。
6.  done：完成信号。
7.  czData：用于存储 ClientConn，addrConn 和 Server 的channelz 相关数据。
8.  cv：当优雅退出时，会等待这个信号量，直到所有 RPC 请求都处理并断开才会继续处理。

## 注册 ##

pb.RegisterSearchServiceServer(server, &SearchService{
        })

### 步骤一：Service API interface ###

// search.pb.go
    type SearchServiceServer interface {
        
    	Search(context.Context, *SearchRequest) (*SearchResponse, error)
    }
    
    func RegisterSearchServiceServer(s *grpc.Server, srv SearchServiceServer) {
        
    	s.RegisterService(&_SearchService_serviceDesc, srv)
    }

还记得我们平时编写的 Protobuf 吗？在生成出来的 .pb.go 文件中，会定义出 Service APIs interface 的具体实现约束。而我们在 gRPC Server 进行注册时，会传入应用 Service 的功能接口实现，此时生成的 RegisterServer 方法就会保证两者之间的一致性

### Service API IDL ###

你想乱传糊弄一下？不可能的，请乖乖定义与 Protobuf 一致的接口方法。但是那个 &\_SearchService\_serviceDesc 又有什么作用呢？代码如下：

var _SearchService_serviceDesc = grpc.ServiceDesc{
        
    	ServiceName: "proto.SearchService",
    	HandlerType: (*SearchServiceServer)(nil),
    	Methods: []grpc.MethodDesc{
        
    		{
        
    			MethodName: "Search",
    			Handler:    _SearchService_Search_Handler,
    		},
    	},
    	Streams:  []grpc.StreamDesc{
        },
    	Metadata: "search.proto",
    }

1.  ServiceName：服务名称
2.  HandlerType：服务接口，用于检查用户提供的实现是否满足接口要求
3.  Methods：一元方法集，注意结构内的 Handler 方法，其对应最终的 RPC 处理方法，在执行 RPC 方法的阶段会使用。
4.  Streams：流式方法集
5.  Metadata：元数据，是一个描述数据属性的东西。在这里主要是描述 SearchServiceServer 服务

### 步骤三：Register Service ###

func (s *Server) register(sd *ServiceDesc, ss interface{
        }) {
        
        ...
    	srv := &service{
        
    		server: ss,
    		md:     make(map[string]*MethodDesc),
    		sd:     make(map[string]*StreamDesc),
    		mdata:  sd.Metadata,
    	}
    	for i := range sd.Methods {
        
    		d := &sd.Methods[i]
    		srv.md[d.MethodName] = d
    	}
    	for i := range sd.Streams {
        
    		...
    	}
    	s.m[sd.ServiceName] = srv
    }

我们会将先前的服务接口信息、服务描述信息给注册到内部 service 去。

1.  server：服务的接口信息
2.  md：一元服务的 RPC 方法集
3.  sd：流式服务的 RPC 方法集
4.  mdata：metadata，元数据

### 小结 ###

主要介绍的是 gRPC Server 在启动前的整理和注册行为，看上去很简单，但其实一切都是为了后续的实际运行的预先准备。因此我们整理一下思路，将其串联起来看看，如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMyNDY4OTg_size_16_color_FFFFFF_t_70_pic_center]

## 监听 ##

func (s *Server) Serve(lis net.Listener) error {
        
    	...
    	var tempDelay time.Duration 
    	for {
        
    		rawConn, err := lis.Accept()
    		if err != nil {
        
    			if ne, ok := err.(interface {
        
    				Temporary() bool
    			}); ok && ne.Temporary() {
        
    				if tempDelay == 0 {
        
    					tempDelay = 5 * time.Millisecond
    				} else {
        
    					tempDelay *= 2
    				}
    				if max := 1 * time.Second; tempDelay > max {
        
    					tempDelay = max
    				}
    				...
    				timer := time.NewTimer(tempDelay)
    				select {
        
    				case <-timer.C:
    				case <-s.quit:
    					timer.Stop()
    					return nil
    				}
    				continue
    			}
    			...
    			return err
    		}
    		tempDelay = 0
    
    		s.serveWG.Add(1)
    		go func() {
        
    			s.handleRawConn(rawConn)
    			s.serveWG.Done()
    		}()
    	}
    }

Serve 会根据外部传入的 Listener 不同而调用不同的监听模式，这也是 net.Listener 的魅力，灵活性和扩展性会比较高。而在 gRPC Server 中最常用的就是 TCPConn，基于 TCP Listener 去做。接下来我们一起看看具体的处理逻辑，如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMyNDY4OTg_size_16_color_FFFFFF_t_70_pic_center 1]

1.  循环处理连接，通过 lis.Accept 取出连接，如果队列中没有需处理的连接时，会形成阻塞等待
2.  若 lis.Accept 失败，则触发休眠机制，若为第一次失败那么休眠 5ms，否则翻倍，再次失败则不断翻倍直至上限休眠时间 1s，而休眠完毕后就会尝试去取下一个 “它”。
3.  若 lis.Accept 成功，则重置休眠的时间计数和启动一个新的 goroutine 调用 handleRawConn 方法去执行/处理新的请求，也就是大家很喜欢说的 “每一个请求都是不同的 goroutine 在处理”。
4.  在循环过程中，包含了 “退出” 服务的场景，主要是硬关闭和优雅重启服务两种情况。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMyNDY4OTg_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/58089b1803f146b8a311f9903e7249f5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMyNDY4OTg_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/ac566f111ac8430287f7efa9af8d55c6.png