Worker Services .NET 10: BackgroundService وChannels وqueues في الإنتاج

دروس السلسلة: C# 14 و.NET 10 · EF Core 10 · gRPC ASP.NET Core 10 · BenchmarkDotNet .NET 10

كل تطبيق يتجاوز مرحلة « API HTTP أساسية » ينتهي بحاجة لمعالجات خلفية: إرسال إيميلات، معالجة طوابير رسائل، مزامنات دورية، حسابات batch ليلية، استماع لأحداث. في عالم .NET، template Worker Service المُقدَّم بـ .NET 6 والمُوطَّد في .NET 10 هو الأداة المخصصة. يعتمد على واجهة IHostedService وصنف BackgroundService، يرث كل البنية التحتية (DI، logging، configuration، health checks)، وينشر بنفس السهولة كخدمة systemd أو حاوية Kubernetes.

المتطلبات

• .NET 10 LTS مُثبَّت
• أساسيات C# والـ async
• اختياري للأمثلة: RabbitMQ، Redis، أو Azure Service Bus
• الوقت المُقدَّر: 90 دقيقة

الخطوة 1 — إنشاء مشروع Worker Service

# Créer un nouveau Worker Service
dotnet new worker -n MonService.Workers
cd MonService.Workers

# Structure générée
# - Program.cs (point d'entrée avec HostBuilder)
# - Worker.cs (BackgroundService minimal)
# - appsettings.json
# - appsettings.Development.json

الملف Program.cs يستخدم Minimal Hosting منذ .NET 6:

var builder = Host.CreateApplicationBuilder(args);

builder.Services.AddHostedService<Worker>();
builder.Services.AddHttpClient();
builder.Services.AddSingleton<IClock, SystemClock>();

var host = builder.Build();
host.Run();

الخطوة 2 — تنفيذ BackgroundService الأساسي

الصنف BackgroundService المُجرَّد يكشف method ExecuteAsync(CancellationToken stoppingToken) للـ override. الـ stoppingToken يُشار عند تلقي الخدمة إشارة إيقاف (SIGTERM في Linux، Ctrl+C في dev، إيقاف SCM في Windows). احترام هذا token حاسم لإيقاف نظيف.

public class Worker(ILogger<Worker> logger, IClock clock) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        logger.LogInformation("Worker démarré à {time}", clock.UtcNow);

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                await TraiterAsync(stoppingToken);
            }
            catch (OperationCanceledException)
            {
                break;
            }
            catch (Exception ex)
            {
                logger.LogError(ex, "Erreur dans le worker");
            }

            await Task.Delay(TimeSpan.FromSeconds(30), stoppingToken);
        }

        logger.LogInformation("Worker arrêté proprement");
    }

    private async Task TraiterAsync(CancellationToken ct)
    {
        logger.LogInformation("Traitement à {time}", clock.UtcNow);
        await Task.CompletedTask;
    }
}

ثلاثة انضباطات حرجة. try/catch (Exception ex) داخل الحلقة يمنع خطأً واحدًا من قتل worker. OperationCanceledException مُعالَج منفصلًا كإشارة إيقاف شرعية. Task.Delay(..., stoppingToken) يُتيح مقاطعة فورية للـ sleep.

الخطوة 3 — الدورية بـ PeriodicTimer

protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
    using var timer = new PeriodicTimer(TimeSpan.FromMinutes(1));

    while (await timer.WaitForNextTickAsync(stoppingToken))
    {
        try
        {
            await SynchroniserAsync(stoppingToken);
        }
        catch (OperationCanceledException) { break; }
        catch (Exception ex)
        {
            logger.LogError(ex, "Synchro échouée");
        }
    }
}

الميزة: إذا أخذ SynchroniserAsync 45 ثانية، التنفيذ التالي يبدأ بعد 15 ثانية (1 دقيقة – 45 ث)، لا 1 دقيقة + 45 ث لاحقًا. للـ workers التي يجب أن تحافظ على إيقاع مستقر، هذه الأداة الصحيحة.

الخطوة 4 — عدة workers متوازية مع Channel

public class JobQueue
{
    private readonly Channel<Job> _channel =
        Channel.CreateBounded<Job>(new BoundedChannelOptions(1000)
        {
            FullMode = BoundedChannelFullMode.Wait,
            SingleReader = false,
            SingleWriter = false
        });

    public ChannelReader<Job> Reader => _channel.Reader;
    public ChannelWriter<Job> Writer => _channel.Writer;
}

public class JobConsumer(JobQueue queue, ILogger<JobConsumer> logger) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await foreach (var job in queue.Reader.ReadAllAsync(stoppingToken))
        {
            try { await TraiterJob(job, stoppingToken); }
            catch (Exception ex) { logger.LogError(ex, "Échec job {id}", job.Id); }
        }
    }
}

// Enregistrer plusieurs consommateurs dans Program.cs
builder.Services.AddSingleton<JobQueue>();
builder.Services.AddHostedService<JobConsumer>();
builder.Services.AddHostedService<JobConsumer>();
builder.Services.AddHostedService<JobConsumer>();

الـ BoundedChannel بـ 1000 مكان يُطبّق backpressure: إذا امتلأت queue، المنتج ينتظر. هذا الانضباط يتجنّب انفجارات ذاكرة. للـ scaling أفقيًا، انتقل لـ RabbitMQ أو Azure Service Bus.

الخطوة 5 — Health checks لـ Kubernetes

var webBuilder = WebApplication.CreateBuilder(args);
webBuilder.Services.AddHostedService<Worker>();
webBuilder.Services.AddHealthChecks();

var webApp = webBuilder.Build();
webApp.MapHealthChecks("/health");
webApp.MapGet("/", () => "Worker actif");
webApp.Run();

الـ WorkerHealthCheck المخصص يكشف آخر نشاط للـ worker. إذا مرّ أكثر من 5 دقائق منذ آخر معالجة، نُرجع Unhealthy وKubernetes يُعيد تشغيل pod. للـ workers بإيقاع طويل (cron يومي)، نُكيّف العتبة.

الخطوة 6 — تكامل مع queue بعيدة (RabbitMQ)

public class RabbitConsumer : BackgroundService
{
    private IConnection? _conn;
    private IChannel? _channel;

    public override async Task StartAsync(CancellationToken ct)
    {
        var factory = new ConnectionFactory
        {
            HostName = "rabbitmq",
            UserName = "user",
            Password = "pass"
        };
        _conn = await factory.CreateConnectionAsync(ct);
        _channel = await _conn.CreateChannelAsync(cancellationToken: ct);
        await _channel.QueueDeclareAsync("emails", durable: true, exclusive: false, autoDelete: false);
        await base.StartAsync(ct);
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var consumer = new AsyncEventingBasicConsumer(_channel!);
        consumer.ReceivedAsync += async (_, ea) =>
        {
            try
            {
                var body = Encoding.UTF8.GetString(ea.Body.Span);
                await TraiterMessage(body);
                await _channel!.BasicAckAsync(ea.DeliveryTag, multiple: false);
            }
            catch
            {
                await _channel!.BasicNackAsync(ea.DeliveryTag, false, requeue: true);
            }
        };

        await _channel!.BasicConsumeAsync("emails", autoAck: false, consumer: consumer);
        await Task.Delay(Timeout.Infinite, stoppingToken);
    }
}

الـ BasicAck اليدوي بعد معالجة ناجحة يضمن أن رسالة غير مُعالَجة (crash، exception) تبقى في queue وتُعاد. الـ BasicNack مع requeue: true يُعيد الرسالة لـ queue. للرسائل المسممة، نُعدّ dead-letter queue على جانب RabbitMQ.

الخطوة 7 — نشر Docker وsystemd

# Dockerfile multi-stage
FROM mcr.microsoft.com/dotnet/sdk:10.0 AS build
WORKDIR /src
COPY *.csproj .
RUN dotnet restore
COPY . .
RUN dotnet publish -c Release -o /app

FROM mcr.microsoft.com/dotnet/runtime:10.0 AS runtime
WORKDIR /app
COPY --from=build /app .
ENV DOTNET_ENVIRONMENT=Production
ENTRYPOINT ["dotnet", "MonService.Workers.dll"]

لـ systemd، الصنف UseSystemd() يُضيف دعم الإشارات Unix الأصلية وإدارة logging عبر journald. unit file أدنى (/etc/systemd/system/monworker.service) مع Type=notify وRestart=on-failure يضمن إعادة التشغيل التلقائي عند crash.

أخطاء شائعة

الأسئلة الشائعة

Worker Service أم Hangfire/Quartz.NET؟
Worker Service للمهام المستمرة أو الدورية البسيطة. Hangfire/Quartz لـ scheduling معقد (cron expressions، retries مستمرة، dashboard).

كم instance لكل Worker؟
يعتمد على queue. إذا single-consumer (تسلسلي)، 1. إذا scaling متوازٍ، 1 pod لكل 100-500 رسالة/ثانية + autoscale Kubernetes HPA.

أي أثر GC؟
Server GC افتراضيًا على .NET 10. للـ workers كثيرة allocations، فعّل <ConcurrentGarbageCollection>true</ConcurrentGarbageCollection> وحلّل بـ dotnet-counters.

كيف نختبر BackgroundService؟
أنشئ مباشرة (لا عبر Host) في اختبار، مرّر CancellationTokenSource، انتظر بضع iterations، اعمل assert على الآثار.

AOT متوافق؟
نعم منذ .NET 8. <PublishAot>true</PublishAot> يُعطي binary 5-15 MB يبدأ في < 50 ms.

مقالات ذات صلة

• gRPC مع ASP.NET Core 10: خدمات، streaming وclients مُنمَّطة
• EF Core 10 و.NET 10: vector وLeftJoin والفلاتر المُسمّاة