C++双向流gRPC服务端设置超时阈值并主动中断连接的实现方法

gRPC C++ 服务端没有提供直接的全局流超时配置接口，需要结合异步调度和ServerContext的取消能力自行实现，常见的两种超时场景及实现方式如下：

1. 最大连接时长超时（流从建立开始最长存活时间） #

适用于限制单条流的最长存在时间，不管有没有数据交互到点就断开：

• 在流处理函数入口先计算超时截止时间点
• 启动异步定时器，触发时调用ServerContext::Cancel()方法主动终止当前流连接
• 流处理逻辑中随时检查context->IsCancelled()状态，收到取消信号后清理资源返回

代码示例片段：

Status YourBidirectionalService::YourStreamMethod(ServerContext* context, ServerReaderWriter<Res, Req>* stream) {
    // 配置最大流存活时间为300秒
    const std::chrono::seconds max_stream_duration(300);
    std::atomic<bool> stream_done(false);
    
    // 启动超时检测线程
    std::thread timeout_thread([context, max_stream_duration, &stream_done]() {
        std::this_thread::sleep_for(max_stream_duration);
        if (!stream_done) {
            context->Cancel();
        }
    });
    
    // 正常流读写逻辑
    Req req;
    while (stream->Read(&req)) {
        if (context->IsCancelled()) break;
        // 业务处理、返回响应逻辑
        Res res;
        stream->Write(res);
    }
    
    // 线程清理
    stream_done = true;
    timeout_thread.join();
    
    if (context->IsCancelled()) {
        return Status(StatusCode::DEADLINE_EXCEEDED, "stream max duration exceeded");
    }
    return Status::OK;
}

2. 空闲超时（一段时间没有收到客户端消息则断开） #

适用于检测僵死连接，只有当连续一段时间没有数据交互时才断开：

• 维护最后收到消息的时间戳，每次收到客户端消息时更新该时间戳
• 启动定时检测线程，每隔固定间隔检查当前时间和最后收消息时间的差值，超过阈值则调用context->Cancel()
• 流读写逻辑中随时检查取消状态

代码示例片段：

Status YourBidirectionalService::YourStreamMethod(ServerContext* context, ServerReaderWriter<Res, Req>* stream) {
    // 配置空闲超时时间为60秒
    const std::chrono::seconds idle_timeout(60);
    std::atomic<std::chrono::steady_clock::time_point> last_recv_time = std::chrono::steady_clock::now();
    std::atomic<bool> stream_done(false);
    
    // 空闲检测线程
    std::thread idle_check_thread([context, idle_timeout, &last_recv_time, &stream_done]() {
        while (!stream_done) {
            std::this_thread::sleep_for(std::chrono::seconds(1));
            auto now = std::chrono::steady_clock::now();
            if (now - last_recv_time.load() > idle_timeout) {
                context->Cancel();
                break;
            }
        }
    });
    
    Req req;
    while (stream->Read(&req)) {
        // 收到消息更新时间戳
        last_recv_time = std::chrono::steady_clock::now();
        if (context->IsCancelled()) break;
        // 业务处理、返回响应逻辑
        Res res;
        stream->Write(res);
    }
    
    // 线程清理
    stream_done = true;
    idle_check_thread.join();
    
    if (context->IsCancelled()) {
        return Status(StatusCode::DEADLINE_EXCEEDED, "stream idle timeout");
    }
    return Status::OK;
}

注意事项 #

• 如果使用异步gRPC模型，可以直接复用CompletionQueue的定时器能力实现超时检测，不需要额外开线程，性能更好
• ServerContext::Cancel()是线程安全的，可以在其他线程中直接调用
• 流处理逻辑中要确保检查取消状态的频率足够高，避免取消信号不能及时响应
• 要处理好线程退出的同步逻辑，避免出现悬空指针或者线程泄漏问题

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