Hangfire + 工厂策略模式 + 幂等性控制：一套完整的 .NET 后台任务调度架构

在 SwitchData 网络配置管理系统中，有一批周期性任务需要执行：网元日志采集与解析、UPF 地址配置导入、PCF 统计文件解析等。这些任务具备三个共性：类型可扩展、调度方式多样（定时 / 手动 / 延迟）、必须防重复触发。

本文基于项目的真实代码，讲清楚如何用 Hangfire 做后台任务调度，用 工厂 + 策略 模式让每种任务独立实现，用 幂等性 Key + 状态机 防重复执行，最后用 BackgroundService 在 Worker 启动时一次性把数据库中的任务同步到 Hangfire。

整体架构

`mermaid graph TD A[Web API / Job Worker 启动] –> B[NeTaskHangfireScheduler] B –> C[NeTaskExecutorFactory 工厂] C –> D[INeTaskExecutor 策略接口] D –> E[CollectTaskExecutor 采集] D –> F[ParseTaskExecutor 解析] D –> G[ParseUpfAddressTaskExecutor] D –> H[ParseZxPcfStatFileTaskExecutor]

style A fill:#f9f,stroke:#333
style B fill:#bbf,stroke:#333
style C fill:#bfb,stroke:#333
style D fill:#fbb,stroke:#333

`

• 调度层：NeTaskHangfireScheduler 封装 Hangfire 的 AddOrUpdate / Enqueue / Schedule
• 执行层：INeTaskExecutor 接口 + NeTaskExecutorBase 泛型抽象基类 + 若干具体 Executor
• 工厂层：NeTaskExecutorFactory 根据 NeTaskType 枚举返回对应策略实现
• 持久化层：NeTaskExecution 记录每次执行的上下文、状态、幂等性 Key

第一步：Hangfire 基础配置

Hangfire 建议独立成一个 Worker 进程（项目中的 SwitchData.Job），不要跟 Web API 混在一起，否则在 IIS 回收时会丢失后台处理能力。

SwitchData.Job/Program.cs 里的配置：

`csharp builder.Services.AddHangfire(config => config .SetDataCompatibilityLevel(CompatibilityLevel.Version_180) .UseSimpleAssemblyNameTypeSerializer() .UseRecommendedSerializerSettings() .UseSqlServerStorage(DbConnectionManager.Get(DatabaseNames.Default), new SqlServerStorageOptions { CommandBatchMaxTimeout = TimeSpan.FromMinutes(3), SlidingInvisibilityTimeout = TimeSpan.FromMinutes(10), QueuePollInterval = TimeSpan.Zero, UseRecommendedIsolationLevel = true, DisableGlobalLocks = true }));

builder.Services.AddHangfireServer(); builder.Services.AddSingleton(); `

几个关键参数解释：

第二步：调度器封装

直接在业务代码里写 RecurringJob.AddOrUpdate(…) 太散。NeTaskHangfireScheduler 把 Hangfire 原始 API 包装成项目语义：

`csharp public class NeTaskHangfireScheduler { private readonly IBackgroundJobClient _backgroundJobClient; private readonly IRecurringJobManager _recurringJobManager;

public void EnableSchedule(int taskId, string jobId, string cronExpression)
{
    _recurringJobManager.AddOrUpdate(
        jobId,
        () => NeTaskExecutor.ExecuteRecurringAsync(taskId, NeTaskTriggerSource.Schedule, null),
        cronExpression,
        new RecurringJobOptions { TimeZone = TimeZoneInfo.Local });
}

public void DisableSchedule(string jobId)
    => _recurringJobManager.RemoveIfExists(jobId);

public void TriggerSchedule(int taskId)
    => _backgroundJobClient.Enqueue(
        () => NeTaskExecutor.ExecuteRecurringAsync(taskId, NeTaskTriggerSource.Manual, null));

public void Enqueue(int executionId)
    => _backgroundJobClient.Enqueue(
        () => NeTaskExecutor.ExecuteAsync(executionId, null));

public void Schedule(int executionId, DateTime executeAt)
    => _backgroundJobClient.Schedule(
        () => NeTaskExecutor.ExecuteAsync(executionId, null), executeAt);

} `

设计要点：Enqueue / Schedule 都传 executionId 而不是 taskId——因为手动触发、延迟执行的场景需要一条独立的执行记录，而不是直接复用定时任务配置。

第三步：工厂 + 策略——让每种任务独立实现

策略接口

`csharp public interface INeTaskExecutor { NeTaskType TaskType { get; } Type PayloadType { get; }

Task<JsonDataResult> ExecuteAsync(
    NeTaskExecution execution, object payload,
    PerformContext context, CancellationToken cancellationToken = default);

string CreateIdempotencyKey(int? taskId, object payload);
bool TryGetPayload(string payloadJson, out object payload);

} `

泛型抽象基类（模板方法）

`csharp public abstract class NeTaskExecutorBase : INeTaskExecutor { public abstract NeTaskType TaskType { get; } public Type PayloadType => typeof(TPayload);

public abstract Task<JsonDataResult> ExecuteAsync(
    NeTaskExecution execution, TPayload payload,
    PerformContext context, CancellationToken cancellationToken = default);

public abstract string CreateIdempotencyKey(int? taskId, TPayload payload);

async Task<JsonDataResult> INeTaskExecutor.ExecuteAsync(
    NeTaskExecution execution, object payload,
    PerformContext context, CancellationToken cancellationToken)
    => await ExecuteAsync(execution, (TPayload)payload, context, cancellationToken);

bool INeTaskExecutor.TryGetPayload(string payloadJson, out object payload)
{
    try
    {
        payload = JsonHelper.Deserialize<TPayload>(payloadJson);
        return payload != null;
    }
    catch { payload = default; return false; }
}

} `

具体实现示例

`csharp public class ParseTaskExecutor : NeTaskExecutorBase { public override NeTaskType TaskType => NeTaskType.Parse;

public override string CreateIdempotencyKey(int? taskId, ParseTaskPayload payload)
{
    if (taskId.HasValue) return "TaskId=" + taskId;
    if (payload.NeGroupId.HasValue)
        return "TaskType=" + (int)TaskType + ":NeGroupId=" + payload.NeGroupId;
    if (payload.DeviceId.HasValue)
        return "TaskType=" + (int)TaskType + ":DeviceId=" + payload.DeviceId;
    throw new InvalidOperationException("NeGroupId 和 DeviceId 不能同时为空");
}

public override async Task<JsonDataResult> ExecuteAsync(
    NeTaskExecution execution, ParseTaskPayload payload,
    PerformContext context, CancellationToken cancellationToken)
{
    if (payload.NeGroupId.HasValue)
        return await NeLogService.ParseAndPersistGroupAsync(
            payload.NeGroupId.Value, execution.Id, true, cancellationToken);
    return await NeLogService.ParseAndPersistDeviceAsync(
        payload.DeviceId.Value, execution.Id, true, cancellationToken);
}

} `

工厂——注册表模式

`csharp public static class NeTaskExecutorFactory { private static readonly Dictionary<NeTaskType, INeTaskExecutor> _executors;

static NeTaskExecutorFactory()
{
    _executors = new()
    {
        [NeTaskType.Collect]            = new CollectTaskExecutor(),
        [NeTaskType.Parse]              = new ParseTaskExecutor(),
        [NeTaskType.ParseUpfAddress]    = new ParseUpfAddressTaskExecutor(),
        [NeTaskType.ParseZxPcfStatFile] = new ParseZxPcfStatFileTaskExecutor()
    };
}

public static INeTaskExecutor GetExecutor(NeTaskType taskType)
    => _executors.TryGetValue(taskType, out var ex)
        ? ex
        : throw new NotSupportedException("不支持任务类型：" + taskType);

} `

扩展新任务 = 新建一个继承 NeTaskExecutorBase 的类 + 在工厂字典里注册一行，零侵入现有代码。

第四步：幂等性控制 + 状态机

Hangfire 会重试失败的任务，用户可能手动点了立即执行，定时调度也在跑——同一个任务在某个时间点可能被多个触发器同时命中。

状态流转

mermaid stateDiagram-v2 [*] --> Pending: CreateAsync (手动/延迟) [*] --> Running: CreateAsync (定时触发) Pending --> Running: SetRunningStatusAsync (CAS 更新) Running --> Success: 业务执行成功 Running --> Failed: 异常 / 返回 success=false

关键：CAS 更新防止并发

`csharp public static async Task SetRunningStatusAsync( int executionId, CancellationToken cancellationToken = default) { string sql = @” UPDATE NeTaskExecution SET Status = 2 WHERE ExecutionId = @id AND Status = 1 “;

int rows = await database.ExecuteNonQueryAsync(sql, ...);
return rows > 0;

} `

这条 UPDATE … WHERE Status = Pending 就是经典的 Compare-And-Swap。多个 Worker 同时读到 Pending，但只有一个能成功更新，其他拿到 0 行就跳过。

定时任务的并发保护

`csharp var execution = await NeTaskExecutionService.TryCreateAsync( taskId, neTask.TaskType, triggerSource, neTask.PayloadJson, cancellationToken);

if (execution == null) { DbLogHelper.AddLog(“任务调度”, “任务执行”, false, “任务[TaskId=” + taskId + ”]正在执行，跳过本次触发”); return; } `

TryCreateAsync 调用 CreateAsync 时吞掉了数据库唯一约束异常（幂等 Key 冲突），返回 null 表示跳过本次触发。

第五步：Worker 启动时全量同步

后台服务 JobWorkerService 在 Worker 进程启动时从数据库里读出所有 Enabled = true 的任务，一次性注册到 Hangfire：

`csharp public class JobWorkerService : BackgroundService { private readonly NeTaskHangfireScheduler _scheduler;

protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
    Logger.Info("定时任务Worker启动");
    _ = Task.Run(() => SyncJobsAsync(stoppingToken), stoppingToken);

    using var timer = new PeriodicTimer(TimeSpan.FromSeconds(30));
    while (await timer.WaitForNextTickAsync(stoppingToken)) { }
}

private async Task SyncJobsAsync(CancellationToken cancellationToken)
{
    await foreach (var task in NeTaskService.GetListAsync(cancellationToken))
    {
        if (task.Enabled)
            _scheduler.EnableSchedule(task.Id, task.JobId, task.CronExpression);
        else
            _scheduler.DisableSchedule(task.JobId);
    }
}

} `

为什么要用 BackgroundService：SwitchData.Job 是一个最小化的 WebApplication，启动后 app.Run() 就阻塞了。注册一个 HostedService，它的 ExecuteAsync 会在 Host 启动后立刻执行，刚好用来做初始化同步。

第六步：手动触发 / 延迟执行

除了定时，项目还支持从界面点立即执行和指定 X 分钟后执行：

`csharp [HttpPost(“execute-now”)] public async Task ExecuteNow(int taskId) { var neTask = await NeTaskService.GetAsync(taskId); var scheduler = ServiceLocator.ServiceProvider.GetRequiredService(); scheduler.TriggerSchedule(taskId); return new JsonDataResult { success = true, message = “已加入队列” }; }

[HttpPost(“execute-at”)] public async Task ExecuteAt(NeTaskExecutionRequest request) { return await NeTaskExecutionService.CreateAsync( request.TaskType, request.Payload, request.ExecuteAt); } `

NeTaskExecutionService.CreateAsync 内部会根据 executeAt 是否有值决定调 scheduler.Enqueue 还是 scheduler.Schedule。

实战总结

1. Worker 进程独立：Hangfire 不跟 Web API 混跑，避免 IIS 回收丢失任务
2. 工厂 + 策略：新增任务类型 = 新增 Executor 子类 + 一行注册，开闭原则贯彻到底
3. 幂等性 Key + CAS 更新：定时触发、手动触发、延迟触发三条路径共享一套去重机制
4. 执行记录持久化：每条执行都落库，带状态机流转，既能审计也能让 CAS 生效
5. BackgroundService 全量同步：数据库才是配置的唯一真源，Worker 每次启动自动对齐

这套架构已经在生产环境稳定运行数月，采集 / 解析任务每天自动跑好几趟，手动触发也不会跟定时冲突。如果你也在做类似的后台任务调度，不妨把这几个关键做法抄过去。