1. Infrastructure layer
Distributed lock service
//
public interface IDistributedLockService
{
ValueTask<IAsyncDisposable?> AcquireLockAsync(string resourceKey, TimeSpan expiryTime);
}
//
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 = ().ToString();
var lockKey = $"distributed-lock:{resourceKey}";
try
{
var acquired = await (lockKey, lockToken, expiryTime);
if (acquired)
{
_logger.LogDebug("Successfully acquired distributed lock {LockKey}", lockKey);
return new RedisLockHandle(db, lockKey, lockToken, _logger);
}
_logger.LogDebug("Unable to acquire distributed locks {LockKey}", lockKey);
return null;
}
catch (Exception ex)
{
_logger.LogError(ex, "Obtain a distributed lock {LockKey} An error occurred while", 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("Release distributed locks {LockKey} fail", _lockKey);
}
else
{
_logger.LogDebug("成功Release distributed locks {LockKey}", _lockKey);
}
}
catch (Exception ex)
{
_logger.LogError(ex, "Release distributed locks {LockKey} An error occurred while", _lockKey);
}
finally
{
_isDisposed = true;
}
}
}
}
2. Task service layer
Timed task service
//
public interface IPollingService
{
Task ExecutePollingTasksAsync();
Task ExecuteDailyTaskAsync(int hour);
}
//
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 minutes to prevent concurrency public async Task ExecutePollingTasksAsync()
{
await using var lockHandle = await _lockService.AcquireLockAsync(
"polling-tasks-lock",
(25)); // The lock is valid for 25 minutes
if (lockHandle is null)
{
_logger.LogInformation("Other nodes are executing polling tasks, skip this execution");
return;
}
try
{
_logger.LogInformation("Start polling tasks - node: {NodeId}", );
// Perform all polling tasks await (
PollingTaskAsync(),
PollingExpireTaskAsync(),
PollingExpireDelCharactTaskAsync()
);
// Trigger background tasks _ = (() => PollingDelCharactTaskAsync(), _logger);
_ = (() => AutoCheckApiAsync(), _logger);
_ = (() => DelLogsAsync(), _logger);
}
catch (Exception ex)
{
_logger.LogError(ex, "An error occurred while executing a polling task");
throw;
}
}
[DisableConcurrentExecution(timeoutInSeconds: 60 * 60)] // 1 hour to prevent concurrency public async Task ExecuteDailyTaskAsync(int hour)
{
var lockKey = $"daily-task-{hour}:{:yyyyMMdd}";
await using var lockHandle = await _lockService.AcquireLockAsync(
lockKey,
(55)); // The lock is valid for 55 minutes
if (lockHandle is null)
{
_logger.LogInformation("其他node已implement今日 {Hour} Click on the task", hour);
return;
}
try
{
_logger.LogInformation("Start execution {Hour} Click on the task - node: {NodeId}",
hour, );
if (hour == 21)
{
await ExecuteNightlyMaintenanceAsync();
}
else if (hour == 4)
{
await ExecuteEarlyMorningTasksAsync();
}
}
catch (Exception ex)
{
_logger.LogError(ex, "implement {Hour} Click on the task时发生错误", hour);
throw;
}
}
// Specific task implementation method private async Task PollingTaskAsync()
{
// Implement game character startup/close logic }
private async Task ExecuteNightlyMaintenanceAsync()
{
// Blackjack special task logic }
// Other methods...}
// (Safely run background tasks)public static class BackgroundTask
{
public static Task Run(Func<Task> task, ILogger logger)
{
return (async () =>
{
try
{
await task();
}
catch (Exception ex)
{
(ex, "Background task execution failed");
}
});
}
}
3. Task Schedule Configuration Layer
Task Initializer
//
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 = <IPollingService>();
// Tasks executed every 30 minutes _jobManager.AddOrUpdate<IPollingService>(
"polling-tasks-30min",
s => (),
"*/30 * * * *");
// Tasks performed at 21:00 every day _jobManager.AddOrUpdate<IPollingService>(
"daily-task-21:00",
s => (21),
"0 21 * * *");
// Tasks performed at 04:00 every day _jobManager.AddOrUpdate<IPollingService>(
"daily-task-04:00",
s => (4),
"0 4 * * *");
_logger.LogInformation("Recurring task initialization is completed");
}
catch (Exception ex)
{
_logger.LogError(ex, "Initialization of periodic task failed");
throw;
}
return ;
}
public Task StopAsync(CancellationToken cancellationToken) => ;
}
4. Application startup configuration
var builder = (args);
// Add Redis<IConnectionMultiplexer>(sp =>
(("Redis")));
// Configure Hangfire(config =>
{
(
("Redis"),
new RedisStorageOptions
{
Prefix = "hangfire:",
Db = 1 // Use a separate Redis database });
();
});
(options =>
{
= $"{}:{():N}";
= 1;
= new[] { "default", "critical" };
});
// Register service<IDistributedLockService, RedisDistributedLockService>();
<IPollingService, PollingService>();
<RecurringJobInitializer>();
var app = ();
// Configure Hangfire dashboard("/jobs", new DashboardOptions
{
DashboardTitle = "Task Scheduling Center",
Authorization = new[] { new HangfireDashboardAuthorizationFilter() },
StatsPollingInterval = 60_000 // Refresh once in 60 seconds});
();
// Hangfire Dashboard Authorized Filterpublic class HangfireDashboardAuthorizationFilter : IDashboardAuthorizationFilter
{
public bool Authorize(DashboardContext context)
{
var httpContext = ();
return ?.IsAuthenticated == true;
}
}
5. Configuration
{
"ConnectionStrings": {
"Redis": "localhost:6379,allowAdmin=true",
"Hangfire": "Server=(localdb)\\mssqllocaldb;Database=Hangfire;Trusted_Connection=True;"
},
"Hangfire": {
"WorkerCount": 1,
"SchedulePollingInterval": 5000
}
}
Key Design Notes
1. Distributed lock:
- Implementation using Redis RedLock algorithm
- Automatically handle the acquisition and release of locks
- Includes complete error handling and logging
2. Task isolation:
- Use Hangfire's [DisableConcurrentExecution] to prevent duplicate execution of the same task
- Distributed lock ensures unique execution across nodes
3. Error handling:
- All key operations have try-catch and logging
- Background tasks are executed using a security wrapper
4. Observability:
- Detailed logging
- Hangfire dashboard monitoring
5. Scalability:
- New tasks can be added easily
- Support dynamic adjustment of scheduling strategies
This implementation fully complies with .NET 6 best practices, supports distributed deployment, and ensures that tasks are executed safely and reliably in a cluster environment.
This is the article about the complete solution for .NET6 to implement distributed timing tasks. For more related contents of .NET6 distributed timing tasks, please search for my previous articles or continue browsing the related articles below. I hope everyone will support me in the future!