Hangfire服务器拾取非监听队列任务问题及解决方法求助

现有两台Hangfire服务器A和B：

• Server A仅配置监听queue_a（无默认队列）
• Server B仅配置监听queue_b（无默认队列）

将JobA调度至queue_a（后台或定时任务均可），但该任务有时会被Server B拾取。由于Server B未引用JobA的程序集，无法从数据库加载任务，持续重试直至耗尽次数后任务失败。已配置任务过滤器确保重试时使用原队列，排除重试队列错误的问题。该问题在SqlServer、Redis、PostgreSql存储中均出现。

配置代码示例 #

services.AddHangfireServer((sp, options) =>
{
    options.Queues = new[] { "queue_a" };
    options.WorkerCount = 1;
    options.ServerName = "server-a";
});


services.AddHangfireServer((sp, options) =>
{
    options.Queues = new[] { "queue_b" };
    options.WorkerCount = 1;
    options.ServerName = "server-b";
});

recurringJobManager.AddOrUpdate<JobA>(
                    nameof(JobA),
                    "queue_a",
                    job => job.Execute(CancellationToken.None),
                    "<some cron expression>");

错误日志 #

Hangfire.Common.JobLoadException: Could not load the job. See inner exception for the details.
 ---> System.IO.FileNotFoundException: Could not resolve assembly 'JobA'.
   at System.Reflection.TypeNameParser.ResolveAssembly(String assemblyName)
   at System.Reflection.TypeNameParser.GetType(String typeName, ReadOnlySpan`1 nestedTypeNames, String assemblyNameIfAny)
   at System.Reflection.TypeNameParser.NamespaceTypeName.ResolveType(TypeNameParser& parser, String containingAssemblyIfAny)
   at System.Reflection.TypeNameParser.Parse()
   at System.Reflection.TypeNameParser.GetType(String typeName, Func`2 assemblyResolver, Func`4 typeResolver, Assembly requestingAssembly, Boolean throwOnError, Boolean ignoreCase, Boolean extensibleParser)
   at System.Type.GetType(String typeName, Func`2 assemblyResolver, Func`4 typeResolver, Boolean throwOnError)
   at Hangfire.Common.TypeHelper.DefaultTypeResolver(String typeName) in C:\projects\hangfire-525\src\Hangfire.Core\Common\TypeHelper.cs:line 78
   at Hangfire.Storage.InvocationData.DeserializeJob() in C:\projects\hangfire-525\src\Hangfire.Core\Storage\InvocationData.cs:line 96
   --- End of inner exception stack trace ---
   at Hangfire.Storage.InvocationData.DeserializeJob() in C:\projects\hangfire-525\src\Hangfire.Core\Storage\InvocationData.cs:line 120
   at Hangfire.RecurringJobExtensions.TriggerRecurringJob(IBackgroundJobFactory factory, JobStorage storage, IStorageConnection connection, IProfiler profiler, RecurringJobEntity recurringJob, DateTime now) in C:\projects\hangfire-525\src\Hangfire.Core\RecurringJobExtensions.cs:line 115
   at Hangfire.Server.RecurringJobScheduler.ScheduleRecurringJob(BackgroundProcessContext context, IStorageConnection connection, String recurringJobId, RecurringJobEntity recurringJob, DateTime now) in C:\projects\hangfire-525\src\Hangfire.Core\Server\RecurringJobScheduler.cs:line 333

原因分析 #

Hangfire的RecurringJobScheduler是所有服务器都会运行的后台进程，无论服务器配置了哪些监听队列，它都会扫描存储中的所有周期性任务。当任务触发时间到达时，任何服务器都会尝试执行任务的调度逻辑（创建后台任务实例），而不是仅由监听对应队列的服务器处理。此时如果服务器未引用任务对应的程序集，就会触发程序集加载失败的错误。

解决方案 #

方案1：添加队列限制过滤器 #

自定义一个服务器过滤器，检查当前任务的队列是否属于服务器允许处理的队列，若不属于则取消任务执行：

public class QueueRestrictionFilter : IServerFilter
{
    private readonly HashSet<string> _allowedQueues;

    public QueueRestrictionFilter(IEnumerable<string> allowedQueues)
    {
        _allowedQueues = new HashSet<string>(allowedQueues, StringComparer.OrdinalIgnoreCase);
    }

    public void OnPerforming(PerformingContext filterContext)
    {
        var jobQueue = filterContext.Job?.Queue;
        
        // 若任务指定了队列且不在允许列表中，取消执行
        if (!string.IsNullOrEmpty(jobQueue) && !_allowedQueues.Contains(jobQueue))
        {
            filterContext.Canceled = true;
        }
    }

    public void OnPerformed(PerformedContext filterContext)
    {
        // 无需处理后续逻辑
    }
}

在配置Hangfire服务器时添加该过滤器：

services.AddHangfireServer((sp, options) =>
{
    options.Queues = new[] { "queue_a" };
    options.WorkerCount = 1;
    options.ServerName = "server-a";
    options.Filters.Add(new QueueRestrictionFilter(options.Queues));
});

services.AddHangfireServer((sp, options) =>
{
    options.Queues = new[] { "queue_b" };
    options.WorkerCount = 1;
    options.ServerName = "server-b";
    options.Filters.Add(new QueueRestrictionFilter(options.Queues));
});

方案2：自定义RecurringJobScheduler #

重写RecurringJobScheduler的ScheduleRecurringJob方法，仅当任务队列属于当前服务器监听队列时才执行调度：

public class RestrictedRecurringJobScheduler : RecurringJobScheduler
{
    private readonly string[] _serverQueues;

    public RestrictedRecurringJobScheduler(string[] serverQueues, JobStorage storage) 
        : base(storage)
    {
        _serverQueues = serverQueues;
    }

    protected override bool ScheduleRecurringJob(BackgroundProcessContext context, IStorageConnection connection, string recurringJobId, RecurringJobEntity recurringJob, DateTime now)
    {
        // 检查任务队列是否在服务器允许的队列列表中
        var jobQueue = recurringJob.Job?.Queue;
        if (!string.IsNullOrEmpty(jobQueue) && !_serverQueues.Contains(jobQueue))
        {
            // 跳过不属于当前服务器队列的任务调度
            return false;
        }

        return base.ScheduleRecurringJob(context, connection, recurringJobId, recurringJob, now);
    }
}

替换Hangfire的默认RecurringJobScheduler：

services.AddHangfireServer((sp, options) =>
{
    var queues = new[] { "queue_a" };
    options.Queues = queues;
    options.WorkerCount = 1;
    options.ServerName = "server-a";
    
    // 替换默认的RecurringJobScheduler
    options.BackgroundProcesses.RemoveAll(p => p is RecurringJobScheduler);
    options.BackgroundProcesses.Add(new RestrictedRecurringJobScheduler(queues, sp.GetRequiredService<JobStorage>()));
});

services.AddHangfireServer((sp, options) =>
{
    var queues = new[] { "queue_b" };
    options.Queues = queues;
    options.WorkerCount = 1;
    options.ServerName = "server-b";
    
    options.BackgroundProcesses.RemoveAll(p => p is RecurringJobScheduler);
    options.BackgroundProcesses.Add(new RestrictedRecurringJobScheduler(queues, sp.GetRequiredService<JobStorage>()));
});

方案3：分离调度与执行节点 #

单独部署一个仅负责周期性任务调度的服务器，该服务器引用所有任务的程序集，任务触发后直接推入对应队列；执行节点（Server A、B）仅监听各自队列并执行任务，不运行RecurringJobScheduler进程。

调度节点配置：

services.AddHangfireServer((sp, options) =>
{
    // 不监听任何执行队列
    options.Queues = Array.Empty<string>();
    // 仅保留RecurringJobScheduler进程
    options.BackgroundProcesses.RemoveAll(p => !(p is RecurringJobScheduler));
});

执行节点配置：

services.AddHangfireServer((sp, options) =>
{
    options.Queues = new[] { "queue_a" };
    options.WorkerCount = 1;
    options.ServerName = "server-a";
    // 移除RecurringJobScheduler进程，不参与调度
    options.BackgroundProcesses.RemoveAll(p => p is RecurringJobScheduler);
});

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