.NET6實(shí)現(xiàn)分布式定時(shí)任務(wù)的完整方案

更新時(shí)間：2025年04月20日 10:32:43 作者：碼上有潛

這篇文章主要為大家詳細(xì)介紹了.NET6實(shí)現(xiàn)分布式定時(shí)任務(wù)的完整方案,文中的示例代碼講解詳細(xì),具有一定的借鑒價(jià)值,有需要的小伙伴可以參考一下

1. 基礎(chǔ)設(shè)施層

分布式鎖服務(wù)

// IDistributedLockService.cs
public interface IDistributedLockService
{
    ValueTask<IAsyncDisposable?> AcquireLockAsync(string resourceKey, TimeSpan expiryTime);
}

// RedisDistributedLockService.cs
public class RedisDistributedLockService : IDistributedLockService
{
    private readonly IConnectionMultiplexer _redis;
    private readonly ILogger<RedisDistributedLockService> _logger;

    public RedisDistributedLockService(
        IConnectionMultiplexer redis,
        ILogger<RedisDistributedLockService> logger)
    {
        _redis = redis;
        _logger = logger;
    }

    public async ValueTask<IAsyncDisposable?> AcquireLockAsync(string resourceKey, TimeSpan expiryTime)
    {
        var db = _redis.GetDatabase();
        var lockToken = Guid.NewGuid().ToString();
        var lockKey = $"distributed-lock:{resourceKey}";

        try
        {
            var acquired = await db.LockTakeAsync(lockKey, lockToken, expiryTime);
            if (acquired)
            {
                _logger.LogDebug("成功獲取分布式鎖 {LockKey}", lockKey);
                return new RedisLockHandle(db, lockKey, lockToken, _logger);
            }
            
            _logger.LogDebug("無(wú)法獲取分布式鎖 {LockKey}", lockKey);
            return null;
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "獲取分布式鎖 {LockKey} 時(shí)發(fā)生錯(cuò)誤", lockKey);
            throw;
        }
    }

    private sealed class RedisLockHandle : IAsyncDisposable
    {
        private readonly IDatabase _db;
        private readonly string _lockKey;
        private readonly string _lockToken;
        private readonly ILogger _logger;
        private bool _isDisposed;

        public RedisLockHandle(
            IDatabase db,
            string lockKey,
            string lockToken,
            ILogger logger)
        {
            _db = db;
            _lockKey = lockKey;
            _lockToken = lockToken;
            _logger = logger;
        }

        public async ValueTask DisposeAsync()
        {
            if (_isDisposed) return;

            try
            {
                var released = await _db.LockReleaseAsync(_lockKey, _lockToken);
                if (!released)
                {
                    _logger.LogWarning("釋放分布式鎖 {LockKey} 失敗", _lockKey);
                }
                else
                {
                    _logger.LogDebug("成功釋放分布式鎖 {LockKey}", _lockKey);
                }
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "釋放分布式鎖 {LockKey} 時(shí)發(fā)生錯(cuò)誤", _lockKey);
            }
            finally
            {
                _isDisposed = true;
            }
        }
    }
}

2. 任務(wù)服務(wù)層

定時(shí)任務(wù)服務(wù)

// IPollingService.cs
public interface IPollingService
{
    Task ExecutePollingTasksAsync();
    Task ExecuteDailyTaskAsync(int hour);
}

// PollingService.cs
public class PollingService : IPollingService
{
    private readonly IDistributedLockService _lockService;
    private readonly ILogger<PollingService> _logger;

    public PollingService(
        IDistributedLockService lockService,
        ILogger<PollingService> logger)
    {
        _lockService = lockService;
        _logger = logger;
    }

    [DisableConcurrentExecution(timeoutInSeconds: 60 * 30)] // 30分鐘防并發(fā)
    public async Task ExecutePollingTasksAsync()
    {
        await using var lockHandle = await _lockService.AcquireLockAsync(
            "polling-tasks-lock",
            TimeSpan.FromMinutes(25)); // 鎖有效期25分鐘

        if (lockHandle is null)
        {
            _logger.LogInformation("其他節(jié)點(diǎn)正在執(zhí)行輪詢?nèi)蝿?wù)，跳過(guò)本次執(zhí)行");
            return;
        }

        try
        {
            _logger.LogInformation("開(kāi)始執(zhí)行輪詢?nèi)蝿?wù) - 節(jié)點(diǎn): {NodeId}", Environment.MachineName);
            
            // 執(zhí)行所有輪詢?nèi)蝿?wù)
            await Task.WhenAll(
                PollingTaskAsync(),
                PollingExpireTaskAsync(),
                PollingExpireDelCharactTaskAsync()
            );
            
            // 觸發(fā)后臺(tái)任務(wù)
            _ = BackgroundTask.Run(() => PollingDelCharactTaskAsync(), _logger);
            _ = BackgroundTask.Run(() => AutoCheckApiAsync(), _logger);
            _ = BackgroundTask.Run(() => DelLogsAsync(), _logger);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "執(zhí)行輪詢?nèi)蝿?wù)時(shí)發(fā)生錯(cuò)誤");
            throw;
        }
    }

    [DisableConcurrentExecution(timeoutInSeconds: 60 * 60)] // 1小時(shí)防并發(fā)
    public async Task ExecuteDailyTaskAsync(int hour)
    {
        var lockKey = $"daily-task-{hour}:{DateTime.UtcNow:yyyyMMdd}";
        
        await using var lockHandle = await _lockService.AcquireLockAsync(
            lockKey,
            TimeSpan.FromMinutes(55)); // 鎖有效期55分鐘

        if (lockHandle is null)
        {
            _logger.LogInformation("其他節(jié)點(diǎn)已執(zhí)行今日 {Hour} 點(diǎn)任務(wù)", hour);
            return;
        }

        try
        {
            _logger.LogInformation("開(kāi)始執(zhí)行 {Hour} 點(diǎn)任務(wù) - 節(jié)點(diǎn): {NodeId}", 
                hour, Environment.MachineName);
            
            if (hour == 21)
            {
                await ExecuteNightlyMaintenanceAsync();
            }
            else if (hour == 4)
            {
                await ExecuteEarlyMorningTasksAsync();
            }
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "執(zhí)行 {Hour} 點(diǎn)任務(wù)時(shí)發(fā)生錯(cuò)誤", hour);
            throw;
        }
    }

    // 具體任務(wù)實(shí)現(xiàn)方法
    private async Task PollingTaskAsync()
    {
        // 實(shí)現(xiàn)游戲角色啟動(dòng)/關(guān)閉邏輯
    }
    
    private async Task ExecuteNightlyMaintenanceAsync()
    {
        // 21點(diǎn)特殊任務(wù)邏輯
    }
    
    // 其他方法...
}

// BackgroundTask.cs (安全運(yùn)行后臺(tái)任務(wù))
public static class BackgroundTask
{
    public static Task Run(Func<Task> task, ILogger logger)
    {
        return Task.Run(async () =>
        {
            try
            {
                await task();
            }
            catch (Exception ex)
            {
                logger.LogError(ex, "后臺(tái)任務(wù)執(zhí)行失敗");
            }
        });
    }
}

3. 任務(wù)調(diào)度配置層

任務(wù)初始化器

// RecurringJobInitializer.cs
public class RecurringJobInitializer : IHostedService
{
    private readonly IRecurringJobManager _jobManager;
    private readonly IServiceProvider _services;
    private readonly ILogger<RecurringJobInitializer> _logger;

    public RecurringJobInitializer(
        IRecurringJobManager jobManager,
        IServiceProvider services,
        ILogger<RecurringJobInitializer> logger)
    {
        _jobManager = jobManager;
        _services = services;
        _logger = logger;
    }

    public Task StartAsync(CancellationToken cancellationToken)
    {
        try
        {
            using var scope = _services.CreateScope();
            var pollingService = scope.ServiceProvider.GetRequiredService<IPollingService>();
            
            // 每30分鐘執(zhí)行的任務(wù)
            _jobManager.AddOrUpdate<IPollingService>(
                "polling-tasks-30min",
                s => s.ExecutePollingTasksAsync(),
                "*/30 * * * *");
            
            // 每天21:00執(zhí)行的任務(wù)
            _jobManager.AddOrUpdate<IPollingService>(
                "daily-task-21:00",
                s => s.ExecuteDailyTaskAsync(21),
                "0 21 * * *");
            
            // 每天04:00執(zhí)行的任務(wù)
            _jobManager.AddOrUpdate<IPollingService>(
                "daily-task-04:00",
                s => s.ExecuteDailyTaskAsync(4),
                "0 4 * * *");

            _logger.LogInformation("周期性任務(wù)初始化完成");
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "初始化周期性任務(wù)失敗");
            throw;
        }

        return Task.CompletedTask;
    }

    public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
}

4. 應(yīng)用啟動(dòng)配置

Program.cs

var builder = WebApplication.CreateBuilder(args);

// 添加Redis
builder.Services.AddSingleton<IConnectionMultiplexer>(sp => 
    ConnectionMultiplexer.Connect(builder.Configuration.GetConnectionString("Redis")));

// 配置Hangfire
builder.Services.AddHangfire(config =>
{
    config.UseRedisStorage(
        builder.Configuration.GetConnectionString("Redis"),
        new RedisStorageOptions
        {
            Prefix = "hangfire:",
            Db = 1 // 使用單獨(dú)的Redis數(shù)據(jù)庫(kù)
        });
    
    config.UseColouredConsoleLogProvider();
});

builder.Services.AddHangfireServer(options =>
{
    options.ServerName = $"{Environment.MachineName}:{Guid.NewGuid():N}";
    options.WorkerCount = 1;
    options.Queues = new[] { "default", "critical" };
});

// 注冊(cè)服務(wù)
builder.Services.AddSingleton<IDistributedLockService, RedisDistributedLockService>();
builder.Services.AddScoped<IPollingService, PollingService>();
builder.Services.AddHostedService<RecurringJobInitializer>();

var app = builder.Build();

// 配置Hangfire儀表盤
app.UseHangfireDashboard("/jobs", new DashboardOptions
{
    DashboardTitle = "任務(wù)調(diào)度中心",
    Authorization = new[] { new HangfireDashboardAuthorizationFilter() },
    StatsPollingInterval = 60_000 // 60秒刷新一次
});

app.Run();

// Hangfire儀表盤授權(quán)過(guò)濾器
public class HangfireDashboardAuthorizationFilter : IDashboardAuthorizationFilter
{
    public bool Authorize(DashboardContext context)
    {
        var httpContext = context.GetHttpContext();
        return httpContext.User.Identity?.IsAuthenticated == true;
    }
}

5. appsettings.json 配置

{
  "ConnectionStrings": {
    "Redis": "localhost:6379,allowAdmin=true",
    "Hangfire": "Server=(localdb)\\mssqllocaldb;Database=Hangfire;Trusted_Connection=True;"
  },
  "Hangfire": {
    "WorkerCount": 1,
    "SchedulePollingInterval": 5000
  }
}

關(guān)鍵設(shè)計(jì)說(shuō)明

1.分布式鎖：

使用Redis RedLock算法實(shí)現(xiàn)
自動(dòng)處理鎖的獲取和釋放
包含完善的錯(cuò)誤處理和日志記錄

2.任務(wù)隔離：

使用Hangfire的[DisableConcurrentExecution]防止同一任務(wù)重復(fù)執(zhí)行
分布式鎖確保跨節(jié)點(diǎn)唯一執(zhí)行

3.錯(cuò)誤處理：

所有關(guān)鍵操作都有try-catch和日志記錄
后臺(tái)任務(wù)使用安全包裝器執(zhí)行

4.可觀測(cè)性：

詳細(xì)的日志記錄
Hangfire儀表盤監(jiān)控

5.擴(kuò)展性：

可以輕松添加新任務(wù)
支持動(dòng)態(tài)調(diào)整調(diào)度策略

這個(gè)實(shí)現(xiàn)方案完全符合.NET 6的最佳實(shí)踐，支持分布式部署，確保任務(wù)在集群環(huán)境中安全可靠地執(zhí)行。

到此這篇關(guān)于.NET6實(shí)現(xiàn)分布式定時(shí)任務(wù)的完整方案的文章就介紹到這了,更多相關(guān).NET6分布式定時(shí)任務(wù)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: