Будівництво пам'ятної пам'ятки про страту

Friday, 12 December 2025

Вхід **Частина 1: вогонь і не робити Досить. Забути**Ми дослідили теорію, пов'язану з ефемеральною стратою - приватний, знеохочуючий робочий потік, який запам'ятовує, що є корисним, а потім випаровується.

Ця стаття перетворює цей шаблон у придатну для повторного використання бібліотеку, яку ви можете внести до будь- якого проекту.NET.

ГОЛОСНО!!!!!

Зараз це у пакунку maculicid.ephemerals Nuget також більше, ніж 20 centericd.ephemerals patterns та 'atoms'.

Вихідні файли

Бібліотеку поділено на файли, які мають добрі наслідки:

|------|---------| | EphemeralOptions.cs Передбачення (концертність, розмір вікна, життєвий термін, чверті)} | EphemeralOperation.cs Д_ д. д. д. д. д. д. д. д. д. д. д. д. д. д. д. д. д. д. | Знімки.cs Незамінні записи знімка, що виходять з підходящого боку | Сигнали.cs Дзвінки, поширення, обмеження і глобальні сигнали | EphemeralIdGenerator.cs XxHash64- основаного ID- створенння} | ConcurrencyGates.cs Тежна і придатна для конфігурації рівноправність ♪ | StringPreakenMatcher.cs Шаблон стилю slob- style, що відповідає параметру zespa ♪ | ParpareEphemeral.cs Методи визначення суфікса ( parts of the hash algorithm)EphemeralForEachAsync) | | EphemeralWorkCoordinator.cs Територія довголіття | EphemeraedKeyWorkCoordinator.cs ♪ Per-key sequential execution with Truste' ♪ | EphemeraalResultCoordinator.cs Координарний кіоску ♪ | SignDispatcher.cs } Асинхронний сигнал, що rout з шаблоном} | Коефіцієнт залежності.cs Методи розширення DI та реалізація фабрики | Приклади/ SignalingHttpClient.cs Дзвінок з дрібним випроміненням сигналу для HTTP-окремих дзвінків

І комплексні перевірки Выделяет все дела на край.

До і після

Ось що ми заміняємо:

// ❌ 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│
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

Об' єкт налаштування

Від EphemeralOptions.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; }
}

Ключові рішення

Макс. корективність Типовим значенням є кількість ЦП - раціональна для роботи з ЦП. Для роботи з об' ємом вводу/ виводу, збільшіть його.
Увімкнути DynamicCurrance вмикає зміну часу виконання SetMaxConcurrency() - використовується нетипова брама замість SemaphoreSlim.
Скасувати підписи/ Деферіньйони зробити координатори сигнальними - вони відповідають на стан системи з обмеженими можливостями (підтримка відповідності шаблонів підтримує */?/ comea lists).
OnSignal є синхронним; для асинхронного використання вентилятора SignalDispatcher або AsyncSignalProcessor внутри куратора.
SignConstraints Запобігає нескінченним циклам сигналів з визначенням циклів та обмеженнями глибини.

Записи знімків

Від Знімки.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);

Це Лише метадані. Зверніть увагу, що є неDescription of a condition. Do not translate key words (# V1S #, # V1 #,) тут:

Без вантажу
Немає вхідних даних
Немає вмісту користувача

Просто достаточно, чтобы ответить "что случилось, когда, и это сработало?" - ничего больше.

Як це порівнюється з іншими приступами

. NET надає вам декілька способів, як виконувати паралельні роботи. Ось як порівнюється бібліотека Ефеса:

Паралельно. ForEachAsync (. NET 6+)

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

Найкраще для: Проста паралельна обробка збірок, де вам не потрібна видимість.

Чого їй бракує:

Без стеження за операціями
Без послідовного виконання за ключем
Не показувати те, що виконується

Використовувати ефімераль у: Потрібна програма для зневаджування/ observation, впорядкування клавіш або обробки сигналів.

Потік даних TPL

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;

Найкраще для: Комплексні трубопроводи для обробки даних з розгалуженням, об' єднанням, пакетизацією.

Що він робить добре:

Багата складова трубопроводу (з' єднання блоків разом)
Вбудована пакетизація, перетворення, трансляція
Обмежена місткість зі зворотним тиском

Використовувати потік даних TPL, якщо: Вам потрібні складні топологічні трубопроводи (фран- out, in- in, умовні маршрутизації).

Використовувати ефімераль у: Вам потрібні операції стеження, простіший API, або координація сигналів.

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

Найкраще дляШаблони виробника, де ви контролюєте обидві сторони.

Що він робить добре:

Виконана швидкодія
Стискання за допомогою обмежених каналів
Розділення виробників і споживачів

Використовувати канали у: Ви будуєте нетипову інфраструктуру і потребуєте максимального контролю.

Використовувати ефімераль уВи хочете операції відстеження і утилізацію без кип'ятки.

ПолліCity in Alaska USA

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

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

Найкраще для: Правила підтримки (докладність, Переривання, Тайм- аут) для окремих операцій.

Використовувати у поліВам потрібна стійкість під час окремих дзвінків.

Використовувати ефімераль у: Вам потрібно координації на протязі багатьох операцій з розумовою свідомістю.

Об' єднати їх: Використовуй політивне у своєму робочому тілі для стійкості до дії.

MassTransit / NServiceseBus

Найкраще для: Розповсюджували повідомлення по службах з тривалими чергами.

Використовувати поштові автобуси, якщо: Робота повинна витримувати перезапуск процесу, протяжність декількох служб або потребу у гарантованій доставці.

Використовувати ефімераль у: Робота є підробкою, не потребує міцності, і ви хочете легкої економії.

Таблиця порівняння

Дівчата ведуть нас вперед. |----------|:-------:|:--------:|:-------:|:-------:|:-------------:|:----------:| | Parallel.ForEachAsync А. О. А. О. Що ж до даних ТПУ, то вони мають на увазі, що не мають значення. Мисля, що йде далі. А це означає, що ми маємо справу з носієм. Це означає, що ми маємо на увазі, що ми маємо справу з нашим часом. Зверху ми бачимо, що це означає, що ми маємо справу з середовищем, в якому ми живемо. Д-р Харріс: "Міс-Трансіт/сторінка/Місяць/Пів'єра/Міш'єр-Брайтут." | Бібліотека Ephemeral А це означає, що ми маємо на увазі, що ми маємо справу з програмою.

Ephemeral ForEachAync: однозгоряча версія

Від ParpareEphemeral.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
    });

Чому секретні трубопроводи мають значення

Уявіть, що ви виконуєте команди користувача:

Користувач A надсилає команди 1, 2, 3
Користувач B надсилає команди 4, 5, 6

Без ключових слів вони можуть виконати як: 1, 4, 2, 5, 3, 6 - міжклітинні.

З MaxConcurrencyPerKey = 1:

Команди користувача A виконуються за порядком: 1 → 2 → 3
Команди користувача B виконуються за порядком: 4 → 5 → 6
Але А і В можуть працювати паралельно.

Це послідовна послідовність для кожного, глобальна паралельність - критично для систем, де порядок має значення всередині об'єкта.

Робочий координатор: Давня черга

Від 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();

Безперервні потоки з безсинхронним потоком

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

await coordinator.DrainAsync();

Ключем Coordinator: лінії адрес на Entity

Від EphemeraedKeyWorkCoordinator.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");

Координатори результатів

Від EphemeraalResultCoordinator.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();

Керування коректністю

Від ConcurrencyGates.cs:

У бібліотеці передбачено два механізми керування спільними можливостями:

ФіксованийCurrencyGate (Типовий)

Затримано SemaphoreSlim
Як працює Optimal hot-path
Неможливо корегувати під час виконання

КоригуванняCurrentcyGate

Нетипова реалізація з Queue<WaiterEntry>
Підтримка UpdateLimit() на runningtime
Увімкнено через 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

Шаблон виробника: названі координатори

Від Коефіцієнт залежності.cs:

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

Обов'язки виробника

Та сама назва = той самий екземпляр - Calling CreateCoordinator("fast") двічі повертає того самого координатора
Різні назви = різні екземпляри - "fast" і "accurate" отримання окремих координаторів
Тяжке створення - Координатори створюються лише за першої запиту
Перевірка налаштувань - Запит на незареєстроване ім' я призводить до корисної помилки

API запитів на сигнал

Всі координатори надають оптимізовані способи опитування сигналів:

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

Оптимізація виробництва

Швидке створення ІД

Від EphemeralIdGenerator.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));
    }
}

Без пов' язування (використовує stackalloc)
Безпека гілки (використовує Interlocked.Increment)
Унікальна у процесіComment (включає ІД процесу)
Неналежний [ прибор распространяет прилавка ]

Операція з довготривалим захистом пам' яті

Координатори не зберігають. Task посилання - лише лічильники:

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

Заблокувати за допомогою натискання клавіші

Координатор- ключ автоматично вичищує бездіяльні одноклавішні семафори:

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 - Належна послідовність виконання з справедливим плануванням
EphemeralResultCoordinator - Варіант перегляду результатів
Шаблон виробника - Налаштовані налаштування на зразок IHttpClientFactory
Динамічна сумісність - Коригування паралелізму під час виконання.
Атмосфера сигналів - Вбудований вивід сигналу і запити

Шаблон знаходиться в приємній точці:

Більше видиме, ніж Parallel.ForEachAsync
Простіший за потік даних TPL
Вбудованіші, ніж необроблені канали
Конфіденційна безпека за дизайном

Вогонь... і не зовсім забудь.

Посилання

Частина 1: вогонь і не робити Досить. Забути - теорію і шаблон
Частина 3: Ефемеричні сигнали - перетворення атомів у мережу сприйняття.
Документація SemaphoreSlim
System. Threading.Channels
Потік даних TPL
Шаблон IMTpClientFactory