Golang中如何将HTTP请求参数传递给gRPC双向流RPC调用

核心问题分析 #

你的代码中，HTTP服务（8080端口）和gRPC服务（9000端口）是独立运行的，gRPC的ConnectAndStream方法由客户端主动触发，无法直接获取HTTP请求中的test_id参数。要实现两者的关联，需要在gRPC服务端维护已连接的流式客户端实例，并通过通道将HTTP参数传递到流式调用上下文里。

具体实现步骤 #

1. 重构gRPC服务结构体，添加客户端连接管理

修改你的Server结构体，新增用于保存已连接流式客户端的集合，以及同步锁保证并发安全：

import (
    "sync"
    "github.com/google/uuid"
    // 其他必要包...
)

type StreamClient struct {
    stream TestApiService_ConnectAndStreamServer
    idChan chan int32 // 用于接收HTTP传递的test_id
}

type Server struct {
    testApi.UnimplementedTestApiServiceServer
    clients map[string]*StreamClient
    mu      sync.Mutex
}

2. 修改gRPC流式方法，等待HTTP参数

更新ConnectAndStream方法，创建参数接收通道，将当前流式连接保存到服务实例中，等待HTTP请求传递test_id后再执行收发逻辑：

func (s *Server) ConnectAndStream(channelStream TestApiService_ConnectAndStreamServer) error {
    // 生成唯一客户端ID，用于标识不同的连接
    clientID := uuid.NewString()
    idChan := make(chan int32)

    // 注册客户端连接
    s.mu.Lock()
    s.clients[clientID] = &StreamClient{stream: channelStream, idChan: idChan}
    s.mu.Unlock()

    // 函数退出时清理连接
    defer func() {
        s.mu.Lock()
        delete(s.clients, clientID)
        s.mu.Unlock()
        close(idChan)
    }()

    // 等待HTTP请求传递test_id
    testID, ok := <-idChan
    if !ok {
        return fmt.Errorf("client connection closed")
    }
    log.Println("Id from HTTP request: ", testID)

    // 使用test_id替代硬编码值执行流式逻辑
    id := testID
    for i := 1; i <= 2; i++ {
        id += 1
        log.Println("Speed Server is sending data : ", id)
        if err := channelStream.Send(&Input{Id: id}); err != nil {
            return err
        }
    }

    for i := 1; i <= 2; i++ {
        log.Println("now time to receive")
        client_response, err := channelStream.Recv()
        if err != nil {
            log.Println("Error while receiving from client : ", err)
            return err
        }
        log.Println("Response from client : ", client_response.Id)
    }

    return nil
}

3. 修改HTTP处理函数，传递test_id到流式连接

更新ConnectAndExchange函数，获取HTTP请求中的test_id，找到已连接的gRPC客户端并传递参数，最后返回JSON响应：

import (
    "encoding/json"
    "net/http"
    "strconv"
    "github.com/gorilla/mux"
    // 其他必要包...
)

func ConnectAndExchange(w http.ResponseWriter, r *http.Request, s *Server) {
    vars := mux.Vars(r)
    testIDStr := vars["test_id"]
    testID, err := strconv.Atoi(testIDStr)
    if err != nil {
        http.Error(w, "invalid test_id parameter", http.StatusBadRequest)
        return
    }
    log.Println("Test id request from user : ", testID)

    // 获取已连接的客户端（这里假设只有一个客户端，若多客户端需扩展标识逻辑）
    s.mu.Lock()
    var targetClient *StreamClient
    for _, client := range s.clients {
        targetClient = client
        break
    }
    s.mu.Unlock()

    if targetClient == nil {
        http.Error(w, "no active gRPC client connected", http.StatusServiceUnavailable)
        return
    }

    // 将test_id发送到流式连接的通道
    targetClient.idChan <- int32(testID)

    // 返回JSON响应
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(map[string]interface{}{
        "status":  "success",
        "message": "test_id has been sent to gRPC stream",
        "test_id": testID,
    })
}

4. 初始化服务并关联HTTP与gRPC实例

修改main和runGrpc函数，初始化Server实例并传递给两个服务：

func main() {
    // 初始化gRPC服务实例
    grpcServer := &Server{
        clients: make(map[string]*StreamClient),
    }

    // 启动gRPC服务
    go runGrpc(grpcServer)

    log.Printf("*------ Waiting for requests from users ------*")
    router := mux.NewRouter().StrictSlash(true)
    // 将gRPC服务实例传递给HTTP处理函数
    router.HandleFunc("/exchangeId/{test_id}", func(w http.ResponseWriter, r *http.Request) {
        ConnectAndExchange(w, r, grpcServer)
    }).Methods("GET")

    log.Fatal(http.ListenAndServe(":8080", router))
}

func runGrpc(s *Server) {
    log.Println("Listening for connections from client ........")
    lis, err := net.Listen("tcp", ":9000")
    if err != nil {
        log.Fatalf("failed to listen: %v", err)
    }

    grpcSrv := grpc.NewServer()
    testApi.RegisterTestApiServiceServer(grpcSrv, s)

    if err := grpcSrv.Serve(lis); err != nil {
        log.Fatalf("failed to serve: %s", err)
    }
}

关键说明 #

• 若你的场景存在多个gRPC客户端，需要在HTTP请求中额外传递客户端标识（比如client_id），并在ConnectAndStream方法中从gRPC上下文获取客户端标识，实现精准参数传递。
• 代码中使用了UUID生成客户端ID，你可以根据实际需求替换为其他标识方式（比如客户端传递的自定义ID）。
• 所有并发操作都加了同步锁，避免多请求下的竞态问题。

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