带有 Hangfire 的 CancellationToken

Question

首先，我决定使用 Hangfire，因为它可以在不同的 Windows 服务中或实际上在不同的服务器上运行代码。我可以使用Task类轻松执行我的任务，但我的逻辑将在很长一段时间内 24/7 运行，直到它被用户停止，我认为任务无法处理这个。这就是我使用 Hangfire 的原因。我对不同的解决方案持开放态度。更具体地说，我的逻辑是使用 Web 套接字 24/7 全天候监控内容。

如果你看看我下面的代码，它有 Run 方法，它在 Hangfire 的 BackgroundJob 中生成一个新的机器人。问题是，当我必须停止一个特定的机器人（比如“机器人 1”）时，它应该以某种方式识别当前没有的机器人。

Hangfire 的文档不完整，或者至少我不明白如何从所写的内容中做到这一点。https://docs.hangfire.io/en/latest/background-methods/using-cancellation-tokens.html

private UpdateSubscription _subscription;
private readonly CancellationTokenSource _cts = new CancellationTokenSource();

public async Task RunAsync(string botName)
{
    var jobId = BackgroundJob.Enqueue(() => StartAsync(botName, _cts.Token));
    await _cache.SetAsync($"bot_{botName.Replace(" ", "_")}", jobId);
}

public void Start(Bot bot, CancellationToken token)
{
    // heavy logic
    _subscription = _socketClient.SubscribeToKlineUpdates(bot.CryptoPair.Symbol, bot.TimeInterval.Interval /*KlineInterval.OneHour*/, async data =>
    {
        ... logic ...

        if (token.IsCancellationRequested)
        {
            await _socketClient.Unsubscribe(_subscription);
        }
    }
}

public async Task StopAsync(string botName)
{
    var jobId = await _cache.GetAsync<string>($"bot_{botName.Replace(" ", "_")}");

    if (jobId == null)
        return;

    BackgroundJob.Delete(jobId);
}

编辑： BackgroundJob 将 jobId 作为字符串返回，但if (token.IsCancellationRequested)即使在BackgroundJob.Delete(jobId)调用后也不会以某种方式触发。

我还使用 Redis 来存储作业 ID。也编辑了上面的代码。

Answer 1

要取消 Hangfire 作业，您不必提供自己的 CancellationToken。相反，您提供给 Hangfire 的排队方法需要该类型的参数。当通过代码对作业进行排队时，您在这种情况下提供默认值，并且 Hangfire 会在执行方法时CancellationToken.None小心地提供真实令牌。在您的方法中，您定期询问令牌的状态并根据需要退出。

要取消正在运行的作业，Hangfire 本身会监视两个事件来触发所提供的令牌。服务器收到关闭请求（这将使作业进入新的排队状态，以便在下次启动时再次运行）或作业删除请求。

以下是一些代码草图：

// The method initiated from Hangfire Worker with special parameters
// for cancellationToken and context, filled by Hangfire.
public async Task MyJob(int someParameter, CancellationToken cancellationToken, PerformContext context)
{
    for (int i = 0; i < 100; i++)
    {
        cancellationToken.ThrowIfCancellationRequested();
        Console.WriteLine($"Processing {i} in job {context.Job.Id}");
        await DoSomething(i, cancellationToken);
    }
}

private string EnqueueMyJob(int someParameter)
{
    // When enqueuing a job, just provide some default values for the *magic parameters*
    return jobClient.Enqueue<MyTaskClass>(task => task.MyJob(someParameter, CancellationToken.None, null));
}

private void CancelMyJob(string jobId)
{
    // If the job is currently running, the cancellation token will be set.
    jobClient.Delete(jobId);
}