C++ gRPC长流服务处理信号时如何终止活跃RPC？

问题1：带超时的Shutdown()为何无法取消所有活跃RPC？ #

gRPC的Server::Shutdown()（含超时版本）核心逻辑是阻止新RPC接入并等待已有RPC自然完成，它本身不会主动终止活跃的流RPC。针对你使用的回调式异步API：

• 当客户端未主动取消流RPC时，服务器端的流会持续处于活跃状态，没有触发Finish()的时机；
• 带超时的Shutdown()仅在超时后停止等待，但不会强制终止未完成的RPC；
• 未终止的RPC会持续占用Completion Queue（CQ）的处理资源，导致代码阻塞在grpc_completion_queue_next()——因为CQ仍在等待这些RPC的事件回调。

简言之：Shutdown()是"等待结束"而非"强制结束"，超时只是给等待设了时间限制，不会主动清理未完成的RPC。

问题2：处理信号时，如何在服务器端终止活跃RPC？ #

结合gRPC 1.48.4的回调式异步API，需按以下步骤实现强制终止+优雅清理：

核心步骤 #

1. 捕获信号时先取消所有活跃调用：调用grpc_server_cancel_all_calls()给所有活跃RPC发送取消通知，触发回调中的取消事件；
2. 启动服务器超时关闭：调用server->Shutdown(超时时间)，阻止新RPC接入并等待现有RPC处理取消逻辑；
3. 关闭Completion Queue：调用cq->Shutdown()，让CQ的Next()调用返回false，终止处理循环，避免阻塞；
4. 在RPC回调中处理取消事件：收到取消通知时，必须调用Finish()结束RPC，释放相关资源。

修正后的代码示例 #

#include <grpcpp/grpcpp.h>
#include <signal.h>
#include <thread>
#include <chrono>

std::unique_ptr<grpc::Server> g_server;
std::unique_ptr<grpc::ServerCompletionQueue> g_cq;
bool g_exit_signal = false;

void SignalHandler(int signum) {
    g_exit_signal = true;
    
    // 1. 强制取消所有活跃RPC
    grpc_server_cancel_all_calls(g_server->c_server());
    // 2. 启动服务器关闭，设置1秒超时
    g_server->Shutdown(std::chrono::seconds(1));
    // 3. 关闭CQ，触发处理线程退出
    g_cq->Shutdown();
}

// 示例RPC数据结构（需根据业务场景实现）
class NotificationRpc {
public:
    // 构造函数、业务处理逻辑等...
    
    void OnCancel() {
        // 处理取消：调用Finish结束RPC，释放资源
        responder_.Finish(grpc::Status::CANCELLED(), this);
    }
};

void ProcessCompletionQueue() {
    void* tag;
    bool ok;
    while (g_cq->Next(&tag, &ok)) {
        auto rpc = static_cast<NotificationRpc*>(tag);
        if (!ok) {
            // ok为false表示RPC被取消，执行清理逻辑
            rpc->OnCancel();
            delete rpc;
            continue;
        }
        // 处理正常RPC事件
        rpc->Process();
    }
}

int main() {
    // 注册信号处理函数
    signal(SIGINT, SignalHandler);
    signal(SIGTERM, SignalHandler);

    grpc::ServerBuilder builder;
    // 配置服务器地址、注册服务...
    g_cq = builder.AddCompletionQueue();
    // 注册你的通知服务实例...
    
    g_server = builder.BuildAndStart();

    // 启动CQ处理线程
    std::thread cq_thread(ProcessCompletionQueue);

    // 主线程等待退出信号
    while (!g_exit_signal) {
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }

    // 等待CQ处理线程完成退出
    cq_thread.join();
    return 0;
}

关键说明 #

• grpc_server_cancel_all_calls()会直接触发所有活跃RPC的取消回调，此时必须调用Finish()，否则RPC资源无法释放，仍可能导致阻塞；
• cq->Shutdown()是核心：它会让cq->Next()返回false，终止CQ处理循环，避免线程一直阻塞在事件等待上；
• 超时的Shutdown()仅作为兜底逻辑，真正的终止依赖于"取消所有调用+CQ关闭+回调中Finish处理"的组合操作。

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