Bouwen van een herbruikbare Tijdelijke Executiebibliotheek (Nederlands (Dutch))

In Deel 1: Vuur en niet doen Nogal Vergeet, we onderzochten de theorie achter efemerale uitvoering - begrensd, prive, debuggable async workflows die net genoeg onthouden om nuttig te zijn en vervolgens te verdampen.

Dit artikel verandert dat patroon in een herbruikbare bibliotheek die je kunt laten vallen in elk .NET project.

NUGET!!!

Dit is nu in het meest lucide.efemerals Nuget pakket ook meer dan 20 meestal lucid.efemerals patronen en 'atomen'.

Bronbestanden

De bibliotheek is onderverdeeld in goed gefactoreerde bestanden:

Bestand Doel |------|---------| | EfemeralOptions.cs Configuratie (concurrency, venstergrootte, levensduur, signalen) | EfemeralOperation.cs Internal operation tracking met signaalondersteuning | Snapshots.cs Onveranderlijke snapshot records blootgesteld aan consumenten | Signalen.cs Signal events, propagatie, beperkingen en de globale SignalSink | EfemeralIdGenerator.cs Fast XxHash64-gebaseerde ID-generatie | ConcurrencyGates.cs Vaste en instelbare concurrency limiting | StringPatternMatcher.cs Glob-stijl patroon dat overeenkomt met het signaal filteren | ParallelEphemeral.cs Verlengmethodes voor statische uitbreiding (EphemeralForEachAsync) | | EphemeralWorkCoordinator.cs De coördinator van de langlevende wachtrij | EfemeralKeyedWorkCoordinator.cs Per sleutel sequentiële uitvoering met eerlijke uitvoering | EfemeralResultCoordinator.cs Coördinator van het resultaat-nemingsresultaat . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . | SignalDispatcher.cs Async signaal routing met patroon matching | AfhankelijkheidInjectic.cs DI uitbreidingsmethoden en fabrieksimplementaties | Voorbeelden/SignalingHttpClient.cs Sample fijnkorrelige signaalemissie voor HTTP-gesprekken

En uitgebreide tests alle edge cases.

Voor en na

Dit is wat we vervangen:

// ❌ Before: Fire-and-forget black hole
_ = Task.Run(() => ProcessAsync(item));
// No visibility. No debugging. No idea if it worked.

// ❌ Or: Blocking everything
await ProcessAsync(item);  // Hope you like waiting...

En wat we bouwen:

// ✅ After: Trackable, bounded, debuggable
await coordinator.EnqueueAsync(item);

// Instant visibility
Console.WriteLine($"Pending: {coordinator.PendingCount}");
Console.WriteLine($"Active: {coordinator.ActiveCount}");
Console.WriteLine($"Failed: {coordinator.TotalFailed}");

// Full operation history
var snapshot = coordinator.GetSnapshot();
var failures = coordinator.GetFailed();

Zelfde async uitvoering. Complete observeerbaarheid. Geen gebruikersgegevens bewaard.

Snel starten

Het meest voorkomende patroon - registreer een coördinator in DI en injecteer het:

// Program.cs
services.AddEphemeralWorkCoordinator<TranslationRequest>(
    async (request, ct) => await TranslateAsync(request, ct),
    new EphemeralOptions { MaxConcurrency = 8 });

// Your service
public class TranslationService(EphemeralWorkCoordinator<TranslationRequest> coordinator)
{
    public async Task TranslateAsync(TranslationRequest request)
    {
        await coordinator.EnqueueAsync(request);
        // Returns immediately - work happens in background
    }

    public object GetStatus() => new
    {
        pending = coordinator.PendingCount,
        active = coordinator.ActiveCount,
        completed = coordinator.TotalCompleted,
        failed = coordinator.TotalFailed
    };
}

Welke Variant heb ik nodig?

┌─────────────────────────────────────────────────────────────────┐
│                    DECISION TREE                                │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  Processing a collection once?                                  │
│  └─► EphemeralForEachAsync<T> (ParallelEphemeral.cs)            │
│                                                                 │
│  Need a long-lived queue that accepts items over time?          │
│  └─► EphemeralWorkCoordinator<T>                                │
│                                                                 │
│  Need per-entity ordering (user commands, tenant jobs)?         │
│  └─► EphemeralKeyedWorkCoordinator<TKey, T>                     │
│                                                                 │
│  Need to capture results (fingerprints, summaries)?             │
│  └─► EphemeralResultCoordinator<TInput, TResult>                │
│                                                                 │
│  Need multiple coordinators with different configs?             │
│  └─► IEphemeralCoordinatorFactory<T> (like IHttpClientFactory)  │
│                                                                 │
│  Need dynamic concurrency adjustment at runtime?                │
│  └─► Set EnableDynamicConcurrency = true, call SetMaxConcurrency│
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

Het configuratieobject

Van EfemeralOptions.cs:

public sealed class EphemeralOptions
{
    // Concurrency control
    public int MaxConcurrency { get; init; } = Environment.ProcessorCount;
    public int MaxConcurrencyPerKey { get; init; } = 1;
    public bool EnableDynamicConcurrency { get; init; } = false;

    // Window management
    public int MaxTrackedOperations { get; init; } = 200;
    public TimeSpan? MaxOperationLifetime { get; init; } = TimeSpan.FromMinutes(5);

    // Fair scheduling (keyed coordinator)
    public bool EnableFairScheduling { get; init; } = false;
    public int FairSchedulingThreshold { get; init; } = 10;

    // Signal-reactive processing
    public IReadOnlySet<string>? CancelOnSignals { get; init; }
    public IReadOnlySet<string>? DeferOnSignals { get; init; }
    public int MaxDeferAttempts { get; init; } = 10;
    public TimeSpan DeferCheckInterval { get; init; } = TimeSpan.FromMilliseconds(100);

    // Signal infrastructure
    public SignalSink? Signals { get; init; }
    public SignalConstraints? SignalConstraints { get; init; }
    public Action<SignalEvent>? OnSignal { get; init; }

    // Async signal handling
    public Func<SignalEvent, CancellationToken, Task>? OnSignalAsync { get; init; }
    public int MaxConcurrentSignalHandlers { get; init; } = 4;
    public int MaxQueuedSignals { get; init; } = 1000;

    // Observability
    public Action<IReadOnlyCollection<EphemeralOperationSnapshot>>? OnSample { get; init; }
}

Belangrijkste ontwerpbesluiten

MaxConcurrency standaardwaarden voor CPU-telling - verstandig voor CPU-gebonden werk. Voor I/O-gebonden werk, verhogen.
DynamicConcurrency inschakelen maakt runtime aanpassing via SetMaxConcurrency() - gebruikt een aangepaste poort in plaats van SemaphoreSlim.
AnnulerenOnSignals/DeferOnSignals coördinators signaalreactief maken - ze reageren op omgevingssysteemtoestand (pattern matching supports */?/comma-lijsten).
OnSignal is synchroon; voor async fan-out gebruik SignalDispatcher of AsyncSignalProcessor In de handler.
SignalConstraints voorkomt oneindige signaallussen met cyclusdetectie en dieptelimieten.

The Snapshot Records

Van Snapshots.cs:

public sealed record EphemeralOperationSnapshot(
    long Id,
    DateTimeOffset Started,
    DateTimeOffset? Completed,
    string? Key,
    bool IsFaulted,
    Exception? Error,
    TimeSpan? Duration,
    IReadOnlyList<string>? Signals = null,
    bool IsPinned = false)
{
    public bool HasSignal(string signal) => Signals?.Contains(signal) == true;
}

// For result-capturing coordinators
public sealed record EphemeralOperationSnapshot<TResult>(
    long Id,
    DateTimeOffset Started,
    DateTimeOffset? Completed,
    string? Key,
    bool IsFaulted,
    Exception? Error,
    TimeSpan? Duration,
    TResult? Result,
    bool HasResult,
    IReadOnlyList<string>? Signals = null,
    bool IsPinned = false);

Dit is Uitsluitend metagegevens. Merk op wat is niet Hier:

Geen lading
Geen invoergegevens
Geen gebruikersinhoud

Net genoeg om te antwoorden "wat is er gebeurd, wanneer, en werkte het?" - niets meer.

Hoe dit vergelijkt met andere benaderingen

.NET geeft u verschillende manieren om parallel werk te doen. Hier is hoe de Efemeral bibliotheek vergelijkt:

Parallel.VoorElkeAsync (.NET 6+)

await Parallel.ForEachAsync(items,
    new ParallelOptions { MaxDegreeOfParallelism = 4 },
    async (item, ct) => await ProcessAsync(item, ct));

Beste voor: Eenvoudige parallelle verwerking van collecties waar je geen zichtbaarheid nodig hebt.

Wat het mist:

Geen tracking van bewerkingen
Geen sequentiële uitvoering per sleutel
Geen zicht op wat draait

Efemeral gebruiken wanneer: U moet debuggen/observeerbaarheid, per-key bestelling, of signaal-reactieve verwerking.

TPL-gegevensstroom

var block = new ActionBlock<T>(
    async item => await ProcessAsync(item),
    new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 });

foreach (var item in items)
    block.Post(item);

block.Complete();
await block.Completion;

Beste voor: Complexe dataflow pijpleidingen met vertakking, merging, batching.

Wat het goed doet:

Rijke samenstelling van de pijpleiding (linkblokken samen)
Ingebouwd batchen, transformeren, uitzenden
Gesloten capaciteit met tegendruk

TPL-dataflow gebruiken wanneer: Je hebt complexe pijplijn topologieën nodig (fan-out, fan-in, voorwaardelijke routing).

Efemeral gebruiken wanneer: U hebt operatie tracking, eenvoudiger API, of signaal-reactieve coördinatie nodig.

Systeem.Threading.Channels

var channel = Channel.CreateBounded<T>(100);

// Producer
foreach (var item in items)
    await channel.Writer.WriteAsync(item);
channel.Writer.Complete();

// Consumer (multiple workers)
var workers = Enumerable.Range(0, 4).Select(async _ =>
{
    await foreach (var item in channel.Reader.ReadAllAsync())
        await ProcessAsync(item);
});
await Task.WhenAll(workers);

Beste voor: Producent-consument patronen waar je controle over beide kanten.

Wat het goed doet:

Uitstekende prestaties
Rugdruk via begrensde kanalen
Scheiding van producenten en consumenten

Kanalen gebruiken wanneer: Je bouwt aangepaste infrastructuur en hebt maximale controle nodig.

Efemeral gebruiken wanneer: U wilt operatie tracking en observeerbaarheid zonder ketelplaat.

Polly

var policy = Policy
    .Handle<HttpRequestException>()
    .WaitAndRetryAsync(3, attempt => TimeSpan.FromSeconds(Math.Pow(2, attempt)));

await policy.ExecuteAsync(() => ProcessAsync(item));

Beste voor: Resilience policies (pension, stroomonderbreker, timeout) for individual operations.

Polly gebruiken wanneer: Je hebt veerkracht nodig rond individuele gesprekken.

Efemeral gebruiken wanneer: Je hebt coördinatie nodig over vele operaties met omgevingsbewustzijn.

Combineer ze: Gebruik Polly in je Efemerale werklichaam voor per-operatie veerkracht.

MassTransit / NServiceBus

Beste voor: Distributed messaging over diensten met duurzame wachtrijen.

Berichtbussen gebruiken wanneer: Werk moet procesherstarten overleven, meerdere diensten overspannen of gegarandeerde levering vereisen.

Efemeral gebruiken wanneer: Werk is in proces, heeft geen duurzaamheid nodig, en je wilt lichtgewicht opmerkzaamheid.

Vergelijkingstabel

Aanpak Begrensd volgen Per-Key Signalen Zelfreinigend Complexiteit |----------|:-------:|:--------:|:-------:|:-------:|:-------------:|:----------:| | Parallel.ForEachAsync Vertaald door Simply releases Toppers Sync: DevilsBackbone TPL Dataflow . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . Channels . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . Polly, n.v.t., n.v.t., n.v.t., n.v.t., n.v.t., n.v.t., n.v.t., n.v.t., n.v.t., n.v.t., n.v.t. Background-services (Background-services - achtergrondservices) MassTransit/NServiceBus . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . | Ephemeral Library . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .

EfemeralForEachAsync: De One-Shot versie

Van ParallelEphemeral.cs:

// Simple parallel processing with tracking
await items.EphemeralForEachAsync(
    async (item, ct) => await ProcessAsync(item, ct),
    new EphemeralOptions { MaxConcurrency = 8 });

// With keyed execution (per-user sequential)
await commands.EphemeralForEachAsync(
    cmd => cmd.UserId,  // Key selector
    async (cmd, ct) => await ExecuteCommandAsync(cmd, ct),
    new EphemeralOptions
    {
        MaxConcurrency = 32,
        MaxConcurrencyPerKey = 1  // Sequential per user
    });

Waarom Keyed Pipelines Matter

Stel je voor dat je gebruikersopdrachten verwerkt:

Gebruiker A stuurt commando's 1, 2, 3
Gebruiker B stuurt commando's 4, 5, 6

Zonder sleuteling kunnen deze worden uitgevoerd als: 1, 4, 2, 5, 3, 6 - tussenspleet.

Met MaxConcurrencyPerKey = 1:

Gebruiker A's commando's uitvoeren in volgorde: 1 → 2 → 3
User B's commando's uitvoeren in volgorde: 4 → 5 → 6
Maar A en B kunnen parallel lopen.

Dit is sequentiële per-entiteit, wereldwijd parallel - van cruciaal belang voor systemen waarbij orde binnen een entiteit van belang is.

De werkcoördinator: een lange wachtrij

Van EphemeralWorkCoordinator.cs:

await using var coordinator = new EphemeralWorkCoordinator<TranslationRequest>(
    async (request, ct) => await TranslateAsync(request, ct),
    new EphemeralOptions
    {
        MaxConcurrency = 8,
        MaxTrackedOperations = 500,
        EnableDynamicConcurrency = true  // Allow runtime adjustment
    });

// Enqueue items over time
await coordinator.EnqueueAsync(new TranslationRequest("Hello", "es"));

// Check status anytime
Console.WriteLine($"Pending: {coordinator.PendingCount}");
Console.WriteLine($"Active: {coordinator.ActiveCount}");

// Get snapshots
var snapshot = coordinator.GetSnapshot();
var running = coordinator.GetRunning();
var failed = coordinator.GetFailed();
var completed = coordinator.GetCompleted();

// Control flow
coordinator.Pause();   // Stop pulling new work
coordinator.Resume();  // Continue

// Adjust concurrency at runtime (requires EnableDynamicConcurrency)
coordinator.SetMaxConcurrency(16);

// Pin important operations to survive eviction
coordinator.Pin(operationId);
coordinator.Unpin(operationId);
coordinator.Evict(operationId);

// When done
coordinator.Complete();
await coordinator.DrainAsync();

Continue streams met IAsyncEnumerable

await using var coordinator = EphemeralWorkCoordinator<Message>.FromAsyncEnumerable(
    messageStream,  // IAsyncEnumerable<Message>
    async (msg, ct) => await ProcessMessageAsync(msg, ct),
    new EphemeralOptions { MaxConcurrency = 16 });

await coordinator.DrainAsync();

The Keyed Coördinator: Per-Entity Pijpleidingen

Van EfemeralKeyedWorkCoordinator.cs:

await using var coordinator = new EphemeralKeyedWorkCoordinator<string, Command>(
    cmd => cmd.UserId,  // Key selector
    async (cmd, ct) => await ExecuteCommandAsync(cmd, ct),
    new EphemeralOptions
    {
        MaxConcurrency = 32,
        MaxConcurrencyPerKey = 1,      // Per-user sequential
        EnableFairScheduling = true,   // Prevent hot user starvation
        FairSchedulingThreshold = 10   // Reject if user has 10+ pending
    });

// TryEnqueue returns false if fair scheduling rejects
if (!coordinator.TryEnqueue(hotUserCommand))
{
    await DeferCommandAsync(hotUserCommand);
}

// Per-key visibility
var pendingForUser = coordinator.GetPendingCountForKey("user-123");
var opsForUser = coordinator.GetSnapshotForKey("user-123");

Coördinators voor het opvangen van resultaten

Van EfemeralResultCoordinator.cs:

await using var coordinator = new EphemeralResultCoordinator<SessionInput, SessionResult>(
    async (input, ct) =>
    {
        var fingerprint = await ComputeFingerprintAsync(input.Events, ct);
        return new SessionResult(fingerprint, input.Events.Length);
    },
    new EphemeralOptions { MaxConcurrency = 16 });

await coordinator.EnqueueAsync(session);
coordinator.Complete();
await coordinator.DrainAsync();

// Get just the results (no metadata)
var results = coordinator.GetResults();

// Get snapshots with results + metadata
var snapshots = coordinator.GetSnapshot();

// Get base snapshots without results (privacy-safe)
var baseSnapshots = coordinator.GetBaseSnapshot();

// Filter by success/failure
var successful = coordinator.GetSuccessful();
var failed = coordinator.GetFailed();

Concurrency Control (Concurrency Control)

Van ConcurrencyGates.cs:

De bibliotheek biedt twee concurrency controlemechanismen:

FixedConcurrencyGate (Standaard)

Gesteund door SemaphoreSlim
Optimale hotpathprestaties
Kan niet worden aangepast op runtime

VerstelbareConcurrencyGate

Aangepaste implementatie met Queue<WaiterEntry>
Ondersteunt UpdateLimit() op runtime
Ingeschakeld via EnableDynamicConcurrency = true

// Dynamic concurrency adjustment
var coordinator = new EphemeralWorkCoordinator<T>(body,
    new EphemeralOptions
    {
        MaxConcurrency = 4,
        EnableDynamicConcurrency = true
    });

// Later, based on system load:
coordinator.SetMaxConcurrency(16);  // Scale up
coordinator.SetMaxConcurrency(2);   // Scale down

Het Fabriekspatroon: Benoemde coördinatoren

Van AfhankelijkheidInjectic.cs:

Zoals IHttpClientFactory, u kunt de naam configuraties registreren:

// Registration
services.AddEphemeralWorkCoordinator<TranslationRequest>("fast",
    async (request, ct) => await FastTranslateAsync(request, ct),
    new EphemeralOptions { MaxConcurrency = 32 });

services.AddEphemeralWorkCoordinator<TranslationRequest>("accurate",
    async (request, ct) => await AccurateTranslateAsync(request, ct),
    new EphemeralOptions { MaxConcurrency = 4 });

// Usage
public class TranslationService(IEphemeralCoordinatorFactory<TranslationRequest> factory)
{
    private readonly EphemeralWorkCoordinator<TranslationRequest> _fast =
        factory.CreateCoordinator("fast");
    private readonly EphemeralWorkCoordinator<TranslationRequest> _accurate =
        factory.CreateCoordinator("accurate");
}

Fabrieksgarantie

Zelfde naam = zelfde instantie - Ik bel je. CreateCoordinator("fast") twee keer retourneert dezelfde coördinator
Verschillende namen = verschillende instanties - "fast" en "accurate" afzonderlijke coördinatoren ophalen
Lui aanmaken - Coördinators worden alleen aangemaakt wanneer eerst gevraagd
Configuratievalidatie - Het aanvragen van een niet-geregistreerde naam werpt een nuttige fout

Signal Querying API

Alle coördinatoren bieden optimale signaalzoekmethoden:

// Get all signals
var signals = coordinator.GetSignals();

// Filter by key (zero-allocation)
var userSignals = coordinator.GetSignalsByKey("user-123");

// Filter by time range
var recentSignals = coordinator.GetSignalsSince(DateTimeOffset.UtcNow.AddMinutes(-5));
var rangeSignals = coordinator.GetSignalsByTimeRange(from, to);

// Filter by signal name or pattern
var rateSignals = coordinator.GetSignalsByName("rate-limit");
var httpSignals = coordinator.GetSignalsByPattern("http.*");

// Check existence (short-circuits on first match)
if (coordinator.HasSignal("rate-limit"))
    await ThrottleAsync();

if (coordinator.HasSignalMatching("error.*"))
    await AlertAsync();

// Count signals efficiently (no allocation)
var totalSignals = coordinator.CountSignals();
var errorCount = coordinator.CountSignals("error");
var httpCount = coordinator.CountSignalsMatching("http.*");

Optimalisering van de productie

Snelle ID-generatie

Van EfemeralIdGenerator.cs:

internal static class EphemeralIdGenerator
{
    private static long _counter;
    private static readonly long _processStart = Environment.TickCount64;
    private static readonly int _processId = Environment.ProcessId;

    [MethodImpl(MethodImplOptions.AggressiveInlining)]
    public static long NextId()
    {
        var counter = Interlocked.Increment(ref _counter);

        // Combine counter with process-unique seed
        Span<byte> buffer = stackalloc byte[24];
        BitConverter.TryWriteBytes(buffer, _processStart);
        BitConverter.TryWriteBytes(buffer.Slice(8), _processId);
        BitConverter.TryWriteBytes(buffer.Slice(16), counter);

        return unchecked((long)XxHash64.HashToUInt64(buffer));
    }
}

Toewijzingsvrij (gebruik) stackalloc)
Thread-safe (gebruik) Interlocked.Increment)
Uniek voor alle processen (inclusief proces-ID)
Niet-sequentieel (hash verspreidt de teller)

Geheugenveilige langlevende operatie

De coördinatoren bewaren niet Task referenties - gewoon tellers:

private int _activeTaskCount;
private readonly TaskCompletionSource _drainTcs;

// In ExecuteItemAsync:
finally
{
    // Signal drain when last task completes AND channel iteration is done
    if (Interlocked.Decrement(ref _activeTaskCount) == 0 &&
        Volatile.Read(ref _channelIterationComplete))
    {
        _drainTcs.TrySetResult();
    }
}

Per key slot opruimen

De sleutelcoördinator reinigt automatisch inactieve semaforen per sleutel:

private sealed class KeyLock(SemaphoreSlim gate, int maxCount)
{
    public SemaphoreSlim Gate { get; } = gate;
    public int MaxCount { get; } = maxCount;
    public long LastUsedTicks = Environment.TickCount64;
}

// Cleanup runs periodically, removes locks idle > 60 seconds

Compleet voorbeeld

// Program.cs
var builder = WebApplication.CreateBuilder(args);

// Named coordinators
builder.Services.AddEphemeralWorkCoordinator<TranslationRequest>("fast",
    async (req, ct) => await FastTranslateAsync(req, ct),
    new EphemeralOptions { MaxConcurrency = 16 });

// Keyed coordinator for per-user commands
builder.Services.AddEphemeralKeyedWorkCoordinator<string, UserCommand>("commands",
    cmd => cmd.UserId,
    sp =>
    {
        var handler = sp.GetRequiredService<ICommandHandler>();
        return async (cmd, ct) => await handler.HandleAsync(cmd, ct);
    },
    new EphemeralOptions
    {
        MaxConcurrency = 32,
        MaxConcurrencyPerKey = 1,
        EnableFairScheduling = true,
        CancelOnSignals = new HashSet<string> { "system-overload" }
    });

var app = builder.Build();

// Controller
[ApiController]
[Route("api")]
public class WorkController : ControllerBase
{
    private readonly EphemeralWorkCoordinator<TranslationRequest> _translator;
    private readonly EphemeralKeyedWorkCoordinator<string, UserCommand> _commands;

    public WorkController(
        IEphemeralCoordinatorFactory<TranslationRequest> translationFactory,
        IEphemeralKeyedCoordinatorFactory<string, UserCommand> commandFactory)
    {
        _translator = translationFactory.CreateCoordinator("fast");
        _commands = commandFactory.CreateCoordinator("commands");
    }

    [HttpPost("translate")]
    public async Task<IActionResult> Translate([FromBody] TranslationRequest request)
    {
        await _translator.EnqueueAsync(request);
        return Ok(new { pending = _translator.PendingCount });
    }

    [HttpPost("command")]
    public IActionResult SubmitCommand([FromBody] UserCommand command)
    {
        if (!_commands.TryEnqueue(command))
            return StatusCode(429, "Too many pending commands for this user");
        return Ok();
    }

    [HttpGet("status")]
    public IActionResult GetStatus() => Ok(new
    {
        translator = new
        {
            pending = _translator.PendingCount,
            active = _translator.ActiveCount,
            completed = _translator.TotalCompleted,
            failed = _translator.TotalFailed,
            hasRateLimit = _translator.HasSignal("rate-limit")
        },
        commands = new
        {
            pending = _commands.PendingCount,
            active = _commands.ActiveCount,
            errorCount = _commands.CountSignalsMatching("error.*")
        }
    });
}

Conclusie

We hebben een complete executoriale bibliotheek gebouwd met:

EphemeralForEachAsync - One-shot parallelle verwerking met tracking
EphemeralWorkCoordinator - Langlevende waarneembare wachtrijen
EphemeralKeyedWorkCoordinator - Per-entiteit sequentiële uitvoering met eerlijke planning
EphemeralResultCoordinator - Resultaat-opname variant
Fabriekspatroon - Genoemde configuraties zoals IHttpClientFactory
Dynamische concurrency - Runtime aanpassing van parallelisme
Signaalinfrastructuur - Ingebouwde signaalemissie en query

Het patroon zit op een zoete plek:

Zichtbaarder dan Parallel.ForEachAsync
Eenvoudiger dan TPL-gegevensstroom
Meer geïntegreerd dan rauwe kanalen
Privacy-veilig door ontwerp

Vuur... en vergeet het niet.