内 **第1部分:火和不要火 不错 忘记**我们探索了短暂处决背后的理论-- 绑定的,私人的,可调试的,同步的工作流程, 记得的足够有用,然后蒸发。

此文章将该图案转换为可再使用库, 您可以将它投放到任何 .NET 工程中 。

举起手来!

这现在存在于多数卢布的. 短期的Nuget包件中, 也存在于20多个多数卢布的. 短期的图案和“ 原子” 中。.

源源文件

图书馆被分割成由多个因素组成的文件:

|------|---------| | 时间选择cs 配置 (货币、 窗口大小、 寿命、 信号) | 短期行动cs 使用信号支持进行内部行动跟踪 | 抓图. cs 与消费者接触的不可调和的快照记录 | 信号cs 信号事件、传播、限制和全球信号辛克 | 瞬间IdGenerator.cs 快速 XxHash64 型ID 一代 | 中央货币计算器cs 固定和可调整的货币限制 | 书签父母 用于信号过滤的球式模式匹配 | 平行时间 cs * 静态扩展方法EphemeralForEachAsync) | | 短期工作协调员cs 长寿工作队列协调员 | Emeraal-Keyey 工作协调员cs · 以公平的日程安排,逐个相继执行 | 短期成果协调员cs * 成果控制协调员变式 | 信号发送器cs 与模式匹配的 Async 信号路由 * * * * * * * * * * async signal routing * * * * * * async signal routing * * * * * * asmall * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * | 注射注射.cs * DI推广方法和工厂实施 | 实例/信号HttpClient.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│
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

---

配置对象

发自 时间选择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; }
}

关键设计决定

• 最大币值 CPU 默认为 CPU 计数 - 对 CPU 计数的工作是明智的。 对于 I/O 计数的工作,请增加 CPU 计数 。
• 启用动态货币 使运行时间调整能够通过 SetMaxConcurrency() - 使用自订大门,而不是 SemaphoreSlim.
• 取消信号/取消信号 使协调员使信号-反应反应反应 - 对环境系统状态作出反应(模式匹配支持) */?(/comma清单)。
• OnSignal 信号 是同步的; 对于 Async 扇片退出使用 SignalDispatcher 或 AsyncSignalProcessor 控制器内部。
• 信号约束 防止无穷无尽的信号循环与循环探测和深度限制。

---

《快照记录》

发自 抓图. 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);

这是 仅包含元元数据。注意什么是 否 这里 :

• 无有效载荷
• 无输入数据
• 无用户内容

足以回答“发生了什么,何时发生,何时起作用?” - 仅此而已。

---

此方法如何与其他方法相比

. NET 给了您几种平行工作的方式。 下面是“瞬间”图书馆的比较方式:

平行。 ForEachAsync (.NET 6+)

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

最佳简单平行处理收藏 不需要可见度的地方。

其缺乏的:

• 无业务跟踪跟踪
• 没有按键相继执行
• 没有可见度进入运行中

使用时使用时短时间:您需要调试/可观察性、按键订购或信号反应处理。

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 数据流:您需要复杂的管道结构(fan-out, fan-in, 有条件路线)。

使用时使用时短时间:您需要操作跟踪、简单的API或信号反应协调。

系统. 示意图. 通道

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

最佳:各项业务的复原力政策(遥控、断路器、超时)。

使用 Polly 时:您需要针对个人电话的复原力。

使用时使用时短时间:您需要在环境意识下对许多操作进行协调。

结合他们:在您的时空工作机构内使用Polly,以适应每个行动。

质量中转/ NServiciceBus

最佳:以耐久的排队方式在各种服务中发布信息。

使用信息巴士时使用:工作必须生存下来,才能重新开始、跨越多种服务,或需要有保证的交付。

使用时使用时短时间:工作在进行中,不需要耐久性, 你想要轻量级的可观察性。

比较比较表

接近 跟踪 Per-Key 信号 自我清除 复杂 = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = |----------|:-------:|:--------:|:-------:|:-------:|:-------------:|:----------:| | Parallel.ForEachAsync ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ 数据流 数据流 数据流 数据流 数据流 数据流 数据流 数据流 数据流 数据流 数据流 数据流 数据流 数据流 数据流 数据流 数据流 数据流 数据流 数据流 频道 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 中音 {\fn黑体\fs22\bord1\shad0\3aHBE\4aH00\fscx67\fscy66\2cHFFFFFF\3cH808080}

• 背景服务 * * * 中度 * * * 背景服务 * * * * 中度 * * * * 背景服务 * * * * * * 背景服务 * * * * * * * * 背景服务 * * * * * * * * 背景服务 * * * * * * * * * 背景服务 *
• Masstransit/NServic Bus * * mass Transit/NServic Bus * * * * 高 * * * mass Transit/NServic Bus * * * * * * * * 高 * * * masstransit/NServic Bus * * * * * * * * * * * * * 高 * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * | 短期图书馆 ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~

---

Ephemeral ForEachAssync: 一片版本

发自 平行时间 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
• 但A和B可以平行运行

这是 先后顺序,全球平行 - 对于实体内部需要秩序的系统至关重要。

---

工作协调员:长寿队

发自 短期工作协调员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();

使用 IAsync 连续流流

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

await coordinator.DrainAsync();

---

关键协调员:人均管道

发自 Emeraal-Keyey 工作协调员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");

---

成果能力协调员

发自 短期成果协调员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();

---

货币货币管制

发自 中央货币计算器cs:

图书馆提供两种货币管制机制:

固定货币计算单位( 默认)

• 后退于 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

---

工厂模式:命名协调员

发自 注射注射.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");
}

工厂担保

1. 相同名称 = 相同实例 - 拨打 CreateCoordinator("fast") 返回同一位协调员的两倍
2. 不同名称=不同实例 - "fast" 和 "accurate" 获得单独的协调员
3. 懒懒惰创作 - 只有在第一次提出要求时才设立协调员
4. 配置验证 - 要求一个未登记的名字会出一个有用的错误

---

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

---

生产最优化

快速 ID 生成

发自 瞬间IdGenerator.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)
• 各个进程的独特性 (包括进程ID)
• 非顺序 (破碎的散开柜台)

长期记忆安全行动

协调员不储存 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();
    }
}

Per- Key 锁清理

Keyed 协调员自动清理空闲的人均血压:

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

---

结论 结论 结论 结论 结论

我们用以下方法建立了一个完整的执行时间图书馆:

1. EphemeralForEachAsync - 与跟踪平行处理单镜头
2. EphemeralWorkCoordinator - 长寿命可观察队列
3. EphemeralKeyedWorkCoordinator - 逐个实体相继执行,并有公平的日程安排
4. EphemeralResultCoordinator - 成果采集变量
5. 工厂形态 - 命名配置类似 IHttpClientFactory
6. 动态共同货币 - 平行的运行时间调整
7. 信号基础设施 - 内置信号发射和查询

这些图案坐落在一个甜美的地方:

• 可见度大于 Parallel.ForEachAsync
• 数据流比 TPL 简单
• 集成多于原声道
• 设计保护隐私

火... 别忘了

---

链接链接链接链接

• 第1部分:火和不要火 不错 忘记 - 理论和模式
• 第3部分:短暂信号 - 将原子变成一个遥感网络
• SemaphhoreSlim文献资料
• 系统. 示意图. 通道
• TPL 数据流
• IHttpClipClientFacit 模式