如何解决TelegramBot中CancellationToken无法取消多线程定时发信任务的问题

问题根因 #

• 你使用了全局唯一的CancellationTokenSource，多用户订阅时会互相覆盖，仅最后一个订阅的用户能触发取消，其余用户的订阅任务永远收不到取消信号
• Task.Delay(3000)未传入取消令牌，3秒等待过程中收到取消信号不会中断等待，必须等等待结束进入下一次判断才会触发异常
• 仅在循环末尾判断取消请求，会导致取消信号发出后，还会额外发送一次「1111」消息
• 全局CancellationTokenSource在第一次取消后就被Dispose，后续新用户订阅时会使用已释放的令牌源，无法正常触发取消
• Task.Run传入的令牌仅能在任务启动前生效，运行过程中的取消逻辑必须由你自己在任务内部传播

修复方案 #

1. 替换全局令牌源为按用户存储的字典 #

首先定义线程安全的字典存储每个用户的令牌源：

private readonly ConcurrentDictionary<long, CancellationTokenSource> _userCtsMap = new();

2. 修改订阅启动方法 #

private async Task OnStartSubscribeAsync(string userName, long userId,
    ITelegramBotClient _client, long chatId)
{
    var user = new UserDTO
    {
        UserId = userId.ToString(),
        UserName = userName
    };
    await _userService.StartSubscribeAsync(user);
    await _client.SendTextMessageAsync(chatId, "You subscribed successfully ");

    // 如果用户之前有未取消的订阅，先取消
    if (_userCtsMap.TryRemove(userId, out var oldCts))
    {
        oldCts.Cancel();
        oldCts.Dispose();
    }
    // 为当前用户新建独立的令牌源
    var userCts = new CancellationTokenSource();
    _userCtsMap.TryAdd(userId, userCts);
    var token = userCts.Token;

    try
    {
        await CheckTick(_client, chatId, token);
    }
    catch (OperationCanceledException)
    {
        await _client.SendTextMessageAsync(chatId, "STOPPED");
    }
    finally
    {
        // 任务结束后移除并释放当前用户的令牌源
        if (_userCtsMap.TryRemove(userId, out var cts))
        {
            cts.Dispose();
        }
    }

    var articles = await ReturnNewArticles();
    foreach (var item in articles)
    {
        var linkButton = KeyboardGoOver("Перейти", (EncodeUrl(item.Href)));
        await _client.SendPhotoAsync(chatId: chatId, photo: item.Image,
            caption: $"*{item.Title}*",
            parseMode: Telegram.Bot.Types.Enums.ParseMode.Markdown,
            replyMarkup: linkButton);
    }
}

3. 修改延时发信方法 #

private Task CheckTick(ITelegramBotClient _client, long chatId, CancellationToken token)
{
    return Task.Run(async () =>
    {
        token.ThrowIfCancellationRequested();
        while (true)
        {
            // 等待时传入令牌，收到取消信号直接抛出异常
            await Task.Delay(3000, token);
            // 发消息也传入令牌，传播取消信号
            await _client.SendTextMessageAsync(chatId, "1111", cancellationToken: token);
        }
    }, token);
}

4. 修改取消订阅方法 #

private async Task OnStopSubscibeAsync(string userName, long userId,
    ITelegramBotClient _client, long chatId)
{
    var user = new UserDTO()
    {
        UserId = userId.ToString(),
        UserName = userName
    };
    await _userService.StopSubscribeAsync(user);
    // 取出当前用户的令牌源触发取消
    if (_userCtsMap.TryGetValue(userId, out var userCts))
    {
        userCts.Cancel();
    }
    await _client.SendTextMessageAsync(chatId, "You unsubscribed successfully");
}

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