C#中不使用取消令牌通过GRPC从客户端终止服务端流的问题

你当前的代码逻辑完全错误，核心问题有两个：

1. StopStream发起了全新的GRPC调用：你在StopStream里新建了MyRequest并调用StreamArrays，这和SendStreamArray启动的流是两个完全独立的连接，服务端原来的循环里的request还是最初传入的IsCancelledRequested=false的对象，根本检测不到你新发送的true。
2. 单向流无法修改已发送的请求：你用的是客户端单次请求、服务端推流的单向流模式，客户端发送初始请求后，服务端收到的request对象就固定了，后续没法再修改这个对象的属性来传递终止信号。

既然不能用取消令牌，就得用双向流（Duplex Streaming），这种模式允许客户端在同一个流连接中多次发送消息，服务端可以实时监听这些消息来触发终止逻辑。

1. 先调整Proto定义（关键） #

需要把服务方法改成双向流类型：

service MyService {
  // 把原来的单向流改成双向流：stream MyRequest 表示客户端可以多次发送请求
  rpc StreamArrays (stream MyRequest) returns (stream MyResponse);
}

message MyRequest {
  bool IsCancelledRequested = 1;
}

message MyResponse {
  int32 data = 1;
}

2. 服务端代码修改 #

public override async Task StreamArrays(IAsyncStreamReader<MyRequest> requestStream, IServerStreamWriter<MyResponse> responseStream, ServerCallContext context)
{
    bool isCancelled = false;
    // 启动后台任务监听客户端发送的所有请求，捕获终止信号
    var listenCancelTask = Task.Run(async () =>
    {
        await foreach (var request in requestStream.ReadAllAsync())
        {
            if (request.IsCancelledRequested)
            {
                isCancelled = true;
                break;
            }
        }
    });

    try
    {
        int counter = 0;
        // 循环推流，直到收到终止信号
        while (!isCancelled)
        {
            var response = new MyResponse { data = ++counter };
            await responseStream.WriteAsync(response);
            await Task.Delay(15);
            // 额外检查监听任务是否完成，避免无意义循环
            if (listenCancelTask.IsCompleted) break;
        }
    }
    finally
    {
        // 确保监听任务正常完成
        await listenCancelTask;
    }
}

3. 客户端代码修改 #

// 保存双向流的引用，确保Start和Stop操作的是同一个流
private AsyncDuplexStreamingCall<MyRequest, MyResponse> _activeStream;

public async Task StartStream()
{
    // 初始化双向流
    _activeStream = client.StreamArrays();
    // 先发送初始的非终止信号
    await _activeStream.RequestStream.WriteAsync(new MyRequest { IsCancelledRequested = false });

    // 后台异步读取服务端的响应流
    _ = Task.Run(async () =>
    {
        await foreach (var response in _activeStream.ResponseStream.ReadAllAsync())
        {
            data = response.data;
        }
    });
}

public async Task StopStream()
{
    if (_activeStream == null) return;

    // 向同一个流发送终止信号
    await _activeStream.RequestStream.WriteAsync(new MyRequest { IsCancelledRequested = true });
    // 关闭客户端请求流，告知服务端不再发送数据
    await _activeStream.RequestStream.CompleteAsync();
    // 等待流完全结束
    await _activeStream.ResponseStream.Completion;
    // 清空引用
    _activeStream = null;
}

逻辑说明 #

• 双向流允许客户端在同一个连接中多次发送请求，服务端通过监听requestStream可以实时获取客户端的终止信号
• StartStream启动流后，保存了双向流的引用，StopStream通过这个引用发送终止信号，服务端的listenCancelTask会捕获到信号并设置isCancelled标志，让推流循环退出
• 全程没有使用取消令牌，完全通过标志位+双向流实现流的终止

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