如何用工作池控制Go gRPC服务器的请求处理goroutine数量？

当然可以，你可以通过信号量模式实现工作池，限制同时处理请求的goroutine数量，确保无论请求量多大，最多只用20个goroutine处理业务逻辑。下面提供两种可行的实现方式：

方法一：在服务方法中直接添加信号量控制 #

这种方式适合针对单个服务方法做限制，步骤清晰直观：

1. 修改服务结构体，加入一个带缓冲的通道作为信号量（通道容量即为最大并发数）
2. 在请求处理方法中，先申请信号量令牌，处理完成后必须释放令牌

修改后的代码示例：

import (
    "context"
    "log"
    "net"
    "time"

    "google.golang.org/grpc"
    pb "your/path/to/proto" // 替换为你的proto实际包路径
)

type server struct {
    pb.UnimplementedMyServerServer
    workerSem chan struct{} // 信号量通道，控制并发处理数
}

func (s *server) HandleRequest(ctx context.Context, in *pb.IncomingRequest) (*pb.ServerResponse, error) {
    // 申请工作池令牌：若通道已满则阻塞等待；若请求被取消则直接返回错误
    select {
    case s.workerSem <- struct{}{}:
        // 成功获取令牌，继续处理业务
    case <-ctx.Done():
        return nil, ctx.Err()
    }
    // 确保处理完成后释放令牌，避免资源泄漏
    defer func() { <-s.workerSem }()

    // 这里放置你的实际业务处理逻辑
    time.Sleep(1 * time.Second) // 模拟耗时操作
    log.Printf("处理请求内容: %v", in.GetMessage())

    return &pb.ServerResponse{Message: "received request"}, nil
}

func main() {
    port := ":50051"
    lis, err := net.Listen("tcp", port)
    if err != nil {
        log.Fatalf("failed to listen: %v", err)
    }

    // 初始化服务，设置工作池最大并发数为20
    s := &server{
        workerSem: make(chan struct{}, 20),
    }

    grpcServer := grpc.NewServer()
    pb.RegisterMyServerServer(grpcServer, s)

    log.Printf("Server listening at %v", lis.Addr())
    if err := grpcServer.Serve(lis); err != nil {
        log.Fatalf("failed to serve: %v", err)
    }
}

方法二：用gRPC拦截器实现全局控制 #

这种方式更通用，能一次性限制所有Unary请求的并发处理数，无需修改每个服务方法：

1. 实现一个Unary拦截器，在拦截器中统一处理信号量的申请与释放
2. 创建gRPC服务器时注册该拦截器

修改后的代码示例：

import (
    "context"
    "log"
    "net"
    "time"

    "google.golang.org/grpc"
    pb "your/path/to/proto" // 替换为你的proto实际包路径
)

type server struct {
    pb.UnimplementedMyServerServer
}

func (s *server) HandleRequest(ctx context.Context, in *pb.IncomingRequest) (*pb.ServerResponse, error) {
    // 你的实际业务处理逻辑
    time.Sleep(1 * time.Second) // 模拟耗时操作
    log.Printf("处理请求内容: %v", in.GetMessage())

    return &pb.ServerResponse{Message: "received request"}, nil
}

// workerPoolInterceptor 创建限制并发数的Unary拦截器
func workerPoolInterceptor(maxWorkers int) grpc.UnaryServerInterceptor {
    sem := make(chan struct{}, maxWorkers)
    return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
        // 申请令牌
        select {
        case sem <- struct{}{}:
            defer func() { <-sem }() // 处理完成后释放令牌
            return handler(ctx, req) // 调用实际服务方法
        case <-ctx.Done():
            return nil, ctx.Err() // 请求已取消，返回对应错误
        }
    }
}

func main() {
    port := ":50051"
    lis, err := net.Listen("tcp", port)
    if err != nil {
        log.Fatalf("failed to listen: %v", err)
    }

    // 创建gRPC服务器时注册拦截器，设置最大并发数为20
    grpcServer := grpc.NewServer(
        grpc.UnaryInterceptor(workerPoolInterceptor(20)),
    )

    pb.RegisterMyServerServer(grpcServer, &server{})

    log.Printf("Server listening at %v", lis.Addr())
    if err := grpcServer.Serve(lis); err != nil {
        log.Fatalf("failed to serve: %v", err)
    }
}

关键注意事项 #

• 两种方式均通过带缓冲通道实现信号量，通道容量就是允许同时处理请求的goroutine数量
• 必须通过defer确保令牌被释放，避免因panic或业务错误导致的资源泄漏
• 业务逻辑中要监听ctx.Done()，及时响应请求取消事件，避免无效的资源占用

内容的提问来源于stack exchange，提问作者flankers33