एक नया आदर्श बनाया जा सकता है

Friday, 12 December 2025

में **पार्ट 1: आग और मत काफी भूल जाना**हमने एक - एक करके इस सिद्धांत की जाँच की, जो निजी तौर पर, डीबगी का काम करता है ।

इस लेख से पता चलता है कि आप किसी भी तरह की लाइब्रेरी में जा सकते हैं ।

नहीं!

अब यह विशेष रूप से अल्पतम पैकेज में है 20 से अधिक।.

स्रोत फ़ाइलें

लाइब्रेरी को अच्छी तरह से जमा किया गया फ़ाइलों में विभाजित किया गया है:

IMTA फ़ाइल BAR |------|---------| | एपीटील विकल्प. सेंटीमीटर कॉन्फ़िगरेशन (अनुप्रयोग, विंडो आकार, जीवन, संकेत) | एपीड्रॉडिशन. SICAR आंतरिक ऑपरेशन संकेत समर्थन के साथ ट्रैक किया जा रहा है | स्नेपशॉट्स.css फूजनिक स्नैपशॉट रिकॉर्डों को उपभोक्ताओं के संपर्क में लाया जा सकता है | सिग्नल.cs सिग्नल घटना, फूटना, प्रतिबन्ध, और वैश्विक सिग्नल- ओएस | एपीडैम्टर सेंटीमीटर तेज XxHM64- आधारित IDEALLLANK | दोष लगाना । डीवीडी+आर( डबल्यू) स्थिर तथा समायोजित किया जा सकता है | स्ट्रिंग इंटरनेशनलमैटर. संकेत फिल्टरिंग के लिए सेल- दर- दर- दर- दर- रंग पैटर्न | समानांतर प्रेसमेराल. स्थैतिक एक्सटेंशन विधियाँ (i)EphemeralForEachAsync) | | एपम्रॉमवर्क कोर. बहुतेरे लंबे समय के कार्य कतार कोऑर्डिनेटर | एपीट- कम्पेटर PRECT पर- कुंजी अनुक्रम उचित सारिणी के साथ चलाना | एपमर्मल (Veteowordordin.s) INVEGES परिणाम- सेटअप किए जा रहे संयोजक | सिग्नल अमेरिका.cs Smughughue सिग्नल पैटर्न से मेल खाने के साथ | डिपेंडेंसी इन्जेक्शन. INVES विस्तार विधि और फैक्टरी | उदाहरण/ चिह्न एचटीटीपी कॉल के लिए बीजीय नमूना प्लगिन

और विस्तार जाँच सभी किनारे के मामलों को कवर.

पहले और बाद

यहाँ हम क्या बदल रहे हैं:

// ❌ 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...

और हम क्या निर्माण कर रहे हैं:

// ✅ 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();

एक ही बार में चलाने के लिए. पूर्ण. कोई उपयोक्ता डाटा नहीं बचा.

क्विक स्टार्ट

सबसे सामान्य पैटर्न - DI में एक कोरी का रजिस्टर करता है और उसे गलत साबित करता है:

// 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
    };
}

मुझे कौन - सी चरनी चाहिए?

┌─────────────────────────────────────────────────────────────────┐
│                    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│
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

कॉन्फ़िगरेशन वस्तु

से एपीटील विकल्प.:

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; }
}

मुख्य डिज़ाइन निर्णय

अधिकतम प्री- प्रोसेसर सीपीयू गणना - सीपीयू गणना कार्य के लिए बुद्धिमान है. मैं/O-Oy-d कार्य के लिए, इसे बढ़ाता हूँ.
बिंदुमयता सक्षम करें (y) के द्वारा अनुकूलन को सक्षम करता है SetMaxConcurrency() - इसके बजाए एक कस्टम गेट का उपयोग करें SemaphoreSlim.
टिक- पर- साइन्स/ डीएस- पर रद्द करें कोऑप्टी सिग्नल-पुष्टि बनाएँ - वे एम्बिएंट तंत्र स्थिति के प्रति प्रतिक्रिया देते हैं ( पैटर्न मिलान समर्थन करता है) */?/कमा सूची.
हस्ताक्षर पर तुल्यकालित किया जाता है; अतुल्यकालिक स्थिति का प्रयोग करने के लिए SignalDispatcher या AsyncSignalProcessor हैंडलर के अंदर.
सिग्नल- पूर्व- आवश्यकताएँ चक्र जांच तथा गहराई सीमा के साथ असीमित सिग्नल लूप को रोकता है.

स्नेपशॉट अभिलेख

से स्नेपशॉट्स.css:

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);

यह है सिर्फ मेटाडाटाध्यान दें कि क्या है नहीं यहाँ:

लोड- करें- नहीं
कोई इनपुट डाटा नहीं
कोई उपयोक्ता सामग्री नहीं

बस जवाब देने के लिए पर्याप्त, जब, और यह काम किया था? - और कुछ भी नहीं.

अन्य प्रश्‍नों से यह कैसे तुलना करता है

.netT आप समानांतर काम करने के लिए कई तरीके देता है। यहाँ कैसे एपिडिडल लाइब्रेरी तुलना करता है:

समानांतर. प्रत्येक के लिए (. tint 6+)

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

के लिए उत्तम: संग्रह की सरल समानांतर प्रक्रिया जहाँ आप की दृश्यता की जरूरत नहीं है.

इसमें कोई कमी नहीं:

ट्रैकिंग कोई क्रिया नहीं
कोई भी कुंजी अनुक्रम वध नहीं
क्या चल रहा है में कोई दृश्यता नहीं है

जब एपेक्ट्डल इस्तेमाल करें तो इस्तेमाल करें: आपको डिबगिंग/ ब्लिकमेंट की आवश्यकता है, प्रति-key आदेश, या संकेत-सक्रिय प्रक्रिया की जरूरत है.

डाटा फ्लो (P)

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;

के लिए उत्तम: साफ - सफाई के साथ - साथ जटिल जानकारी भी आती है ।

यह अच्छी तरह से क्या करता है:

समृद्ध संक्षिप्त संरचना (लिंक पिण्डों को एक साथ)
अंतर्निर्मित प्रचयन, ट्रांसमिटिंग, रेडियोिंग
पीछे से गिरने की क्षमता

जब टी. एल. एल.): आपको जटिल इंफेक्शन (f- आउट, प्रशंसक-in) की जरूरत है.

जब एपेक्ट्डल इस्तेमाल करें तो इस्तेमाल करें: आपको ट्रैक की जरूरत है, आसान एपीआई, या संकेत एम्बिलेशन की जरूरत है.

तंत्र. थ्रेडिंग

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);

के लिए उत्तम: निर्माता-कंकर पैटर्न जहां आप दोनों पक्षों को नियंत्रित करते हैं।

यह अच्छी तरह से क्या करता है:

बढ़िया प्रदर्शन
बाध्य चैनलों के द्वारा बैक- आउटपुट
निर्माता और उपभोक्ताओं से अलग

जब चैनल्स इस्तेमाल करें: तुम इन्फ्रास्फीति का रिवाज़ बना रहे हो और अधिकतम नियंत्रण की जरूरत है.

जब एपेक्ट्डल इस्तेमाल करें तो इस्तेमाल करें: आप बिना साबुन के ट्रैकिंग और ब्लाइडमेंट करना चाहते हैं.

पोली

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

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

के लिए उत्तम: अलग - अलग ऑपरेशनों के लिए रेस्टिंग पॉलिसी (अवरी, सर्किट ब्रेकर, टाइम) ।

जब पोल इस्तेमाल करें: आपको हर व्यक्‍ति से बात करने की ज़रूरत है ।

जब एपेक्ट्डल इस्तेमाल करें तो इस्तेमाल करें: आपको बहुत - से ऑपरेशनों में सावधानी बरतने की ज़रूरत है ।

उन्हें सम्मिलित करें: हर व्यक्ति के लिए अपने एपिडीटीम काम बॉडी के अंदर पोली से काम लीजिए.

द्रव्यमान

के लिए उत्तम: बहुत - से देशों में साक्षियों के काम पर पाबंदी लगी हुई थी ।

जब संदेश विंडो इस्तेमाल करें: काम को फिर से शुरू करने की प्रक्रिया से, बहुत सी सेवाएँ शुरू करने की ज़रूरत है या फिर उसे पूरा करने की ज़रूरत है ।

जब एपेक्ट्डल इस्तेमाल करें तो इस्तेमाल करें: काम प्रक्रिया में है, scrrerircenty की आवश्यकता नहीं है, और आप हल्का bocrscrervivy चाहते हैं.

तुलना तालिका

MRITCACT CONT TECKT TANT T-CKT TECKT TECT TECT TECKT TECT TKT TECT TENT T-key SECKS SECKS SECKSTCKS SICKRICKS SICKRICKCKS TANT TENT TANT TENT TENT TENT TENT TENT TENT TE(KCKCKCKCKSECKCKCKCKCKCKCKCKCKSTCKSTCK(K) |----------|:-------:|:--------:|:-------:|:-------:|:-------------:|:----------:| | Parallel.ForEachAsync ख़ीनएल © 2002- 2006 घड़ियात लिखा आपसी सहयोग बेलारूसी डाटा स्थाप इससे सरियाना लिखा हुआ है । बख्शिश चैनल अनक़रीब ही निष्क्रिय कर देगा बिलिक पोलली निया/ALLLYEAL N/ALLLLYELLL N/ALLLLLL NALLLYYELLLLL NAR_ /ALLLLLLLLLLLLLL N/ALLLLLLLLLLLL बुरी तरह से अधिक पृष्ठभूमि सेवा राहित द्रव्यमान/ NELLLALLLLLLLYALLLLLLLLLYY_BAR_ | एपाइडल लाइब्रेरी बुर्किना फासो घड़ू आदमियों की आपस में आदमियों की मौत हो चुकी है । ( g04 7 / 22)

एपलाइडर के लिए हरित:

से समानांतर प्रेसमेराल.:

// 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
    });

क्यों कुंजीबद्ध पिकलाइन तत्व

उपयोक्ता कमांड्स कल्पना करें:

उपयोक्ता द्वारा कमांड 1, 2, 3
उपयोक्ता B भेजता है कमांड 4, 5, 6

कुंजी के बिना, ये शायद इस प्रकार कार्य करें: १, ४, ५, ३, ३, ६ - ४.

के साथ MaxConcurrencyPerKey = 1:

उपयोक्ता का कमांड अनुक्रम में है: 1 mbox 1 > 3
उपयोक्ता बी का कमांड अनुक्रम में: 4 >
लेकिन एक और बी समानांतर में चला सकते हैं

यह है प्रति- केंद्र अनुक्रम, विश्वव्यापी समानांतर - सिस्टमों के लिए महत्वपूर्ण जहां किसी एंटिटी के भीतर व्यवस्था व्यवस्था.

कार्य कोरेंटर: एक लंबा कतार

से एपम्रॉमवर्क कोर.:

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();

कनेक्शनात्मक स्ट्रीमों के साथ

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

await coordinator.DrainAsync();

कुंजी कोरिएटर: पर- स्टिक्स

से एपीट- कम्पेटर:

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");

परिणाम- cuding कोर

से एपमर्मल (Veteowordordin.s):

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();

अक्रियता नियंत्रण

से दोष लगाना ।:

लाइब्रेरी दो सुग्राहक नियंत्रण यांत्रिकी प्रदान करता है:

स्थिर पहचान (डिफ़ॉल्ट)

पीछे से SemaphoreSlim
ऑपुलल गर्म- पथ प्रदर्शन
स्थिर घोषणाओं को छोड़ नहीं सकता

समायोजनयोग्य

मनपसंद कार्यान्वयन के साथ Queue<WaiterEntry>
समर्थित UpdateLimit() एट
के द्वारा सक्षम 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

फैक्टरी पैटर्न: नामित कोर

से डिपेंडेंसी इन्जेक्शन.:

जैसे IHttpClientFactory, आप कॉन्फ़िगरेशन नाम रजिस्टर कर सकते हैं:

// 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");
}

फैक्टरी ट्रांसमीटरName

एक जैसे नाम = उसी उदाहरण - कॉल कर रहा है CreateCoordinator("fast") उसी संयोजक को दो बार लौटाता है
भिन्न नाम = भिन्न उदाहरण - "fast" और "accurate" अलग समन्वयक मिलता है
zy निर्माण - कोटर केवल बनाया जाता है जब पहले अनुरोध किया
कॉन्फ़िगरेशन - एक अपंजीकृत नाम से एक सहायक त्रुटि फेंक देता है

सिग्नल क्वेरीिंग एपीआई

सभी समन्वयित सिग्नल क्वैरी विधियाँ प्रदान करते हैं:

// 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.*");

उत्पाद विसर्जन

फास्ट आईडी जेनरेशन

से एपीडैम्टर:

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));
    }
}

आबंटन- मुक्त (घुट) stackalloc)
लड़ी सुरक्षित (घुट) Interlocked.Increment)
प्रक्रियाओं के पार बेजोड़ (ewase प्रक्रिया आईडी को शामिल करें)
नॉन- सीक्वेंशियल (घुट को हिलाता है)

मेमोरी- सिग लंबा ऑपरेशन

संयोजकों को स्टोर नहीं है Task संदर्भ - सिर्फ काउंटर: (n)

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();
    }
}

पर- की- लॉक साफ

कुंजी संयोजक स्वतः निष्क्रियता को साफ करता है प्रति-key रीस्फोर्स:

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

पूर्ण उदाहरण

// 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.*")
        }
    });
}

कंटेनमेंट

हम के साथ एक पूर्ण ए-रेप्टर निष्पादन पुस्तकालय का निर्माण किया है:

EphemeralForEachAsync - ट्रैक के साथ एक स्नेपशॉट प्रक्रिया
EphemeralWorkCoordinator - लंबे समय का दृश्‍य कतार
EphemeralKeyedWorkCoordinator - सही अनुसूचित के साथ PRATY अनुक्रम
EphemeralResultCoordinator - परिणाम-Sming चर
फैक्टरी पैटर्न - नामित कॉन्फ़िगरेशन जैसे IHttpClientFactory
गतिशील पहचान - समानांतरता का रन समय समायोजन
सिग्नल इंच - अंतर्निर्मित संकेत उत्सर्जन तथा क्वैरीिंग

पैटर्न एक प्यारी जगह पर बैठता है:

इससे अधिक दृश्‍य Parallel.ForEachAsync
टीएलईएल डाटा फ्लो से सरल
रॉ चैनल्स से ज्यादा एकीकृत करें
डिजाइन द्वारा गोपनीयता- सुरक्षित

आग ... और काफी भूल नहीं है.

लिंक्स

पार्ट 1: आग और मत काफी भूल जाना - सिद्धांत और पैटर्न
पार्ट 3: एपाइडल सिग्नल - एक संयोग से परमाणुओं को एक नेटवर्क में बदल दें
केरोफ़ॉयरॉली दस्तावेज़ीकरण
तंत्र. थ्रेडिंग
डाटा फ्लो (P)
IHttpCirolial पैटर्न