面试题：微服务架构里gRPC双向流式通信的流量控制实现

实现思路

令牌桶算法：令牌桶算法是常用的流量控制算法，它以固定速率生成令牌放入桶中，请求需要从桶中获取令牌才能继续处理，如果桶中没有令牌则请求被限流。
gRPC拦截器：利用gRPC的拦截器机制，在请求处理前后加入流量控制逻辑。
共享令牌桶：在服务端多个gRPC服务实例间共享令牌桶，确保整个服务的流量控制统一。

关键代码片段

引入必要的包

package main

import (
    "context"
    "fmt"
    "google.golang.org/grpc"
    "google.golang.org/grpc/metadata"
    "golang.org/x/time/rate"
    "log"
)

定义令牌桶

var limiter *rate.Limiter

func init() {
    // 每秒生成2个令牌，桶容量为10
    limiter = rate.NewLimiter(rate.Every(1000*1000*1000/2), 10)
}

创建gRPC拦截器

func rateLimitInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
    if!limiter.Allow() {
        return nil, fmt.Errorf("rate limit exceeded")
    }
    return handler(ctx, req)
}

注册gRPC服务并应用拦截器

func main() {
    s := grpc.NewServer(grpc.UnaryInterceptor(rateLimitInterceptor))
    // 注册你的服务
    // pb.RegisterYourServiceServer(s, &yourServiceImpl{})
    lis, err := net.Listen("tcp", ":50051")
    if err!= nil {
        log.Fatalf("failed to listen: %v", err)
    }
    if err := s.Serve(lis); err!= nil {
        log.Fatalf("failed to serve: %v", err)
    }
}

对于双向流式通信，在流处理函数内同样在处理请求数据前进行令牌获取检查：

func (s *YourServer) YourBidirectionalStream(ctx context.Context, stream grpc.ServerStream) error {
    for {
        if!limiter.Allow() {
            return fmt.Errorf("rate limit exceeded")
        }
        req := &YourRequest{}
        err := stream.RecvMsg(req)
        if err!= nil {
            // 处理接收错误
            break
        }
        // 处理请求
        res := &YourResponse{}
        // 填充响应
        err = stream.SendMsg(res)
        if err!= nil {
            // 处理发送错误
            break
        }
    }
    return nil
}

面试题：微服务架构里gRPC双向流式通信的流量控制实现

知识考点

面试题答案

实现思路

关键代码片段