This is a viewer only at the moment see the article on how this works.
To update the preview hit Ctrl-Alt-R (or ⌘-Alt-R on Mac) or Enter to refresh. The Save icon lets you save the markdown file to disk
This is a preview from the server running through my markdig pipeline
在 LRU 缓存中, 一个奇怪的并行系统模式系统@ . @
Sunday, 14 December 2025
这也是我过去一周所沉迷的 前一頁 是什么导致了这个;如果LRU是处死背景呢 .'“.”现在“'”是一套覆盖大多数主要同时执行模式的 30Nuget 套件 ” @(在 TINY & 5-10线套件@). 获得惊人的适应能力,使用 SIMPLE 语法=!}
在這裡找到源碼@: @ https @ :// @ giuthub. @ com_ MS K3} @ scottgal+/} 大多數是LUCID}. – MSkK6Blob/main{/}(大多是LIPICid) MS9epheperalMS K10srcMS11}(最多是luslucid)*. •epheral=.}完成
以往
读取 上一个部分的信号 用于某些洞察它的“'@%s uses @.}这里展示的是来自多半lucid.epheperal}.}完整的和平套件的Readme( md) 它包含核心大多为lucid}MSKO5e-phemeral 包和所有图案}MS K6_ @ 'atoms'} (oconformations et\MSC10}一个方便的 DLLQ.}♪
核心核心核心
OR 使用核心 多数是半卢曲. a TINY @(literally@10}班级=)}这给了你所有的原始功能_.}
属性 @&% DI
OR 如果您想要使用简单路径的完整属性 [EphemeralJob] 并提供服务=.Add
这很可能是我的博客继续讨论的主题 ...你被警告过
多数为 Mostlucid @ . @ Epheperal@ MS K1_ 完成
Meta-package 指向所有最精密的 mostlylucid}.}( 短時套件) - 绑定的同步执行与信号@- 基于协调 @.
dotnet add package mostlylucid.ephemeral.complete
这是一个元元@ -}包件, 可以将所有核心的 < , atom @ MSSK2 _ 和模式软件包作为依赖性=.}您可以安全地混合 @-_NuGet 将解析为单一版本, 没有命名空间碰撞@.}
@ 2.0_ 中有什么新消息吗?
断开更改
信号ink 订阅模式已更改@: @%
// OLD (1.x) - Event-based
sink.SignalRaised += handler;
// NEW (2.0) - Lock-free Subscribe() returning IDisposable
using var sub = sink.Subscribe(handler);
完整软件包现在是一个元@ - package @ MS K1 {
- 以前将所有来源编译为单一的 DLL {( 。 < possibled names space attackes\ )}_
- 现在请参考个别包件作为依赖性
- 安全地与同一解决方案中的单个套件引用相混合
新特点
检测分类簿系统 (
Mostlylucid.Ephemeral.Atoms.Taxonomy.Ledger)DetectionLedger- 累积检测器贡献与小类群集DetectionContribution工厂方法Bot(),Human(),Info(),VerifiedBot(),VerifiedGoodBot()CategoryScore.TotalWeight属性@(\重新命名自Weight)
信号传动跟踪 用于信号链的周期探测和深度限制
源链接 - 直接从您的 IDE 中调试到库源@ info
见见 多克西 / 塔克索诺米 MS K1 基底词库的共享词汇表=, oclees ,\ 原子= MS K2\ 分子= I,} 和升级 .
目录目录
快速启动
using Mostlylucid.Ephemeral;
// Long-lived work coordinator
await using var coordinator = new EphemeralWorkCoordinator<WorkItem>(
async (item, ct) => await ProcessAsync(item, ct),
new EphemeralOptions { MaxConcurrency = 8 });
await coordinator.EnqueueAsync(new WorkItem("data"));
// One-shot parallel processing
await items.EphemeralForEachAsync(
async (item, ct) => await ProcessAsync(item, ct),
new EphemeralOptions { MaxConcurrency = 8 });
服务登记
var builder = WebApplication.CreateBuilder(args);
builder.Services.AddCoordinator<WorkItem>(
async (item, ct) => await ProcessAsync(item, ct),
new EphemeralOptions { MaxConcurrency = 8, MaxTrackedOperations = 128 });
builder.Services.AddEphemeralSignalJobRunner<LogWatcherJobs>();
var app = builder.Build();
app.MapPost("/", async ([FromServices] IEphemeralCoordinatorFactory<WorkItem> factory, WorkItem item) =>
{
var coordinator = factory.CreateCoordinator();
await coordinator.EnqueueAsync(item);
return Results.Accepted();
});
await app.RunAsync();
熟悉的 services.AddCoordinator<T>() 辅助者和辅助者, AddEphemeralSignalJobRunner<T>() 服务保持登记
简洁的, @let DI 拥有水槽@/,,
屬性 - 驱动的工作
mostlylucid.ephemeral.complete 捆包 mostlylucid.ephemeral.attributes@,所以属性管道是核心的一部分
将跑者作为第一个“-”类信号消费者“MS K2”组,
插针故事,}和每个属性都可以声明 Priority-=YTET -伊甸园字幕组=- 翻译: MaxConcurrency, Lane, Key Q, signal
和retries.
密钥属性 knobs @ :
- 排队#&车道: 使用
Priority,MaxConcurrency,和Lane保持确定性的工作秩序,同时热道 单独停留@. @% - 正在键入“% &” 标签:
OperationKey,KeyFromSignal,KeyFromPayload,和[KeySource]帮助您分组工作 记录% , @ 公平日程安排@ MS K1 @ 和诊断+.}的有效密钥 - Pinning &重试:
Pin,ExpireAfterMs,AwaitSignals,MaxRetries,和RetryDelayMs扩展处理器 他们的能见度=,门处决 直到依赖者抵达时为止 ,和在发射失败信号的同时用回转治愈 - 信号编舞~ M:伊密 ~
EmitOnStart,EmitOnComplete,和EmitOnFailure发送下游阶段的信号@, @% log 或其它没有手动接线的协调员@. @%
var sink = new SignalSink();
await using var runner = new EphemeralSignalJobRunner(sink, new[] { new LogWatcherJobs(sink) });
var loggerFactory = LoggerFactory.Create(builder =>
{
builder.AddConsole();
builder.AddProvider(new SignalLoggerProvider(new TypedSignalSink<SignalLogPayload>(sink)));
});
var logger = loggerFactory.CreateLogger("orders");
logger.LogError(new EventId(1001, "DbFailure"), "Order store failed");
// Later tasks or other services can also raise watcher-friendly signals directly:
sink.Raise("log.error.orders.dbfailure", key: "orders");
public sealed class LogWatcherJobs
{
private readonly SignalSink _sink;
public LogWatcherJobs(SignalSink sink) => _sink = sink;
[EphemeralJob("log.error.*", Priority = 1, MaxConcurrency = 2, Lane = "hot:4", EmitOnComplete = new[] { "incident.created" })]
public Task EscalateAsync(SignalEvent signal)
{
Console.WriteLine($"escalating {signal.Signal} for {signal.Key}");
_sink.Raise("incident.created", key: signal.Key);
return Task.CompletedTask;
}
[EphemeralJob("incident.created", EmitOnStart = new[] { "incident.monitor.start" })]
public Task NotifyAsync(SignalEvent signal)
{
Console.WriteLine($"notified incident for {signal.Key}");
return Task.CompletedTask;
}
}
这个跑者现在坐在起步时 随时作出反应 log.error.* 或任何发出的信号击中水槽@.}属性
操作器还可以读取信号中的密钥@ / @ payloads @ ,pinwork 直到下游 acks=, 发射完成@/}#failure signal_ MS K4}和
切入通道以订购 @ . 用于 DI @ I-_ 首个设置 services.AddEphemeralSignalJobRunner<T>() (或范围
变体“)”这样,向导和汇由容器“MS K1”管理。
[EphemeralJobs (Signal Pretifix=}MS K2 ",默认Lane MPK4{"pitiplineMSQ6 公共密封班级 { [EphemeralJob(" NSK1EmitOn compllete\ =新的[@] @ { @ @ MASK2 @ 阶段 @ I.} MS K4 @ done @ MPK5# @ NSK6 @ 公共任务 IngestAsync( @SignalEvent evt @)} @ => Console@._Opte._WriterLineAsync}(evtMS K6#Signale=);{
[EphemeralJob("finalize")]
public Task FinalizeAsync(SignalEvent evt) => Console.Out.WriteLineAsync("final stage");
}
var stageSink = 新的信号Sink (); 正在等待使用 var stageRunner @ = @ 新建的 EphemeralSignalJobRunnner@ MS K1_ Sink,}新的[@] { 新的阶段Jobs() @#});# 阶段Sink.RaiseMS K1 阶段.inest");
Pin- 重活工作可以依赖 ResponsibilitySignalManager.PinUntilQueried ( 默认 ack 模式 responsibility.ack.*)到
在下游阅读器获取有效载荷@, OperationEchoMaker/
OperationEchoAtom 最后的信号流持续到最后,所以审计员或分子仍然可以使用““Tataste”}即使在完成此状态后 ”
原子死亡 .
排定任务
mostlylucid.ephemeral.complete 也包含 mostlylucid.ephemeral.atoms.scheduledtasks"M." 定义拐角或JSON
通过 ScheduledTaskDefinition (cron, signal,可選 key, payload, description, timeZone, format,
runOnStartup@,等 @ .),和让 ScheduledTasksAtom (a) 开展持久的工作 DurableTaskAtom“. ” 每个工作都安排好
在协调窗口中添加配置的信号@ ,, 所以它继承了 pinning @ MS K1 secologing\ ,, 以及责任语义
在您的分子或属性管道响应所发射的信号波@.}时
每个 DurableTask 带有调度表 Name, Signal,任选 Key, 哪怕是打字 Payload,和 Description@,_如此#
下游听众立即知道哪个工作运行以及什么元数据@(filenames @, URLs,a等用于消费.}
DurableTaskAtom.WaitForIdleAsync() 当您只想等待当前计划完成的工作爆破结束时
未完成原子@ , @ 保持排程器为下一个矩形计时备妥@ MPK1 @
正在日志 @ & @% 信号
mostlylucid.ephemeral.logging 微软 . Extensions . 登录到信号,反之亦然. 开始附加
SignalLoggerProvider 您的伐木厂因此日志事件增加 log.* 和钩子 SignalToLoggerAdapter 如果
您想要信号返回到标准日志管道\ .\
var sink = new SignalSink();
var typedSink = new TypedSignalSink<SignalLogPayload>(sink);
using var loggerFactory = LoggerFactory.Create(builder =>
{
builder.AddConsole();
builder.AddProvider(new SignalLoggerProvider(typedSink));
});
using var watcher = new EphemeralSignalJobRunner(sink, new[] { new LogWatcherJobs(sink) });
var logger = loggerFactory.CreateLogger("orders");
logger.LogError(new EventId(1001, "DbFailure"), "Order store failed");
public sealed class LogWatcherJobs
{
private readonly SignalSink _sink;
public LogWatcherJobs(SignalSink sink) => _sink = sink;
[EphemeralJob("log.error.*")]
public Task EscalateAsync(SignalEvent signal)
{
_sink.Raise("incident.created", key: signal.Key);
return Task.CompletedTask;
}
[EphemeralJob("incident.created")]
public Task NotifyAsync(SignalEvent signal)
{
Console.WriteLine($"Incident for {signal.Key}");
return Task.CompletedTask;
}
}
使用 SignalToLoggerAdapter 将生成的信号反射回到标准日志 这样你的监控堆看到两个
桥的两侧 .
核心协调员
套件@: 多数是半卢曲.
短期工作协调员<泰 T T>
Long - 长期工作队列,与捆绑的货币和可观测窗口@.
await using var coordinator = new EphemeralWorkCoordinator<Request>(
async (req, ct) => await HandleAsync(req, ct),
new EphemeralOptions
{
MaxConcurrency = 8,
MaxTrackedOperations = 200,
MaxOperationLifetime = TimeSpan.FromMinutes(5)
});
await coordinator.EnqueueAsync(request);
// Observe state
var running = coordinator.GetRunning();
var failed = coordinator.GetFailed();
var pending = coordinator.PendingCount;
// Graceful shutdown
coordinator.Complete();
await coordinator.DrainAsync();
易眼工作协调员<{\fn黑体\fs22\bord1\shad0\3aHBE\4aH00\fscx67\fscy66\2cHFFFFFF\3cH808080}T>
- key 顺序处理 * -}用相同的键处理过的项目 {. _
await using var coordinator = new EphemeralKeyedWorkCoordinator<Order, string>(
order => order.CustomerId, // Key selector
async (order, ct) => await ProcessOrder(order, ct),
new EphemeralOptions
{
MaxConcurrency = 16, // Total parallel
MaxConcurrencyPerKey = 1 // Sequential per customer
});
await coordinator.EnqueueAsync(order);
短期成果协调员<TResult 数据输入>
Async 操作的抓取结果 @ . @
await using var coordinator = new EphemeralResultCoordinator<Request, Response>(
async (req, ct) => await FetchAsync(req, ct),
new EphemeralOptions { MaxConcurrency = 4 });
var id = await coordinator.EnqueueAsync(request);
var snapshot = await coordinator.WaitForResult(id);
if (snapshot.HasResult)
Console.WriteLine(snapshot.Result);
优先工作协调员<泰 T T>
多优先车道,每个车道可配置同价货币=.
var coordinator = new PriorityWorkCoordinator<WorkItem>(
async (item, ct) => await ProcessAsync(item, ct),
new PriorityWorkCoordinatorOptions<WorkItem>(
Lanes: new[] { new PriorityLane("high"), new PriorityLane("normal"), new PriorityLane("low") }
));
await coordinator.EnqueueAsync(item, "high");
設定 & M( @ 时间选项@ )
new EphemeralOptions
{
// Concurrency
MaxConcurrency = 8, // Max parallel operations
MaxConcurrencyPerKey = 1, // For keyed coordinators
EnableDynamicConcurrency = false, // Allow runtime adjustment
// Memory
MaxTrackedOperations = 200, // Window size (LRU eviction)
MaxOperationLifetime = TimeSpan.FromMinutes(5),
// Fair scheduling (keyed only)
EnableFairScheduling = false, // Prevent hot key starvation
FairSchedulingThreshold = 10,
// Signals
Signals = sharedSink, // Shared signal sink
OnSignal = evt => { }, // Sync callback
OnSignalAsync = async (evt, ct) => { }, // Async callback
CancelOnSignals = new HashSet<string> { "circuit-open" },
DeferOnSignals = new HashSet<string> { "backpressure" },
DeferCheckInterval = TimeSpan.FromMilliseconds(100),
MaxDeferAttempts = 50,
// Signal handler limits
MaxConcurrentSignalHandlers = 4,
MaxQueuedSignals = 1000
}
信号
行动发射横跨-的信号
// Query signals
bool hasError = coordinator.HasSignal("error");
int count = coordinator.CountSignals("error");
var errors = coordinator.GetSignalsByPattern("error.*");
// Shared sink across coordinators
var sink = new SignalSink();
var c1 = new EphemeralWorkCoordinator<A>(body, new EphemeralOptions { Signals = sink });
var c2 = new EphemeralWorkCoordinator<B>(body, new EphemeralOptions { Signals = sink });
sink.Raise("system.busy"); // Both see it
责任信号
下游消费者需要足够长的时间才能看到结果, ResponsibilitySignalManager 让你打针
默认模式 responsibility.ack.* 密钥“=”operationId提供
可选 description 以便描述其责任 *,*和设置 maxPinDuration 优雅地
消费者从不露面,
var manager = new ResponsibilitySignalManager(coordinator, sink, maxPinDuration: TimeSpan.FromMinutes(5));
if (manager.PinUntilQueried(operationId, "file.ready", ackKey: fileId, description: "Awaiting fetch"))
{
sink.Raise("file.ready", key: fileId);
}
// Consumer acknowledges the work
sink.Raise("file.ready.ack", key: fileId);
using Mostlylucid.Ephemeral.Patterns;
var notes = new LastWordsNoteAtom(async note => await noteRepository.SaveAsync(note));
coordinator.OperationFinalized += snapshot =>
{
var note = new LastWordsNote(
OperationId: snapshot.OperationId,
Key: snapshot.Key,
Signal: snapshot.Signals?.FirstOrDefault(),
Timestamp: DateTimeOffset.UtcNow);
_ = notes.EnqueueAsync(note);
};
LastWordsNote @ (_ 操作 idQ,} 键 *, sign@ MS K3} 时间戳=),} 这样您就可以记录你所关心的最小状态
在收集操作前的关于@. @%
协调员还保持了最后信号“(”的即时回声。 EnableOperationEcho)可以吗?
检查 GetEchoes() 当您需要重放曲折的信号波时, 而不保持整个操作在 @.}@ action: button
var recentErrors = coordinator.GetEchoes(pattern: "error.*")
.Where(e => e.Timestamp > DateTimeOffset.UtcNow - TimeSpan.FromMinutes(1))
.ToList();
if (recentErrors.Any())
logger.LogWarning("Trimmed errors: {Count}", recentErrors.Count);
OperationEchoRetention 和 OperationEchoCapacity 平衡你保持的回声和持续的时间
所以您可以重放““”最后一个单词 “”” 只要足够长的时间来进行表面诊断 ‘.’
当 ACK 火灾时, 管理器自动解扣@ , @ 但是您可以调用 CompleteResponsibility(operationId) 结束
@ ( @ e @ I. @ g.,}关于 retries=).} 操作仍然在增加 OperationFinalized 当窗口调整时 @ , @ so
如果您想要发送最终的信号“ ,” 日志诊断@ MS K1 或运行“ I“” 最后一个单词 @ ” cleanup .
Atoms * (* 建筑砖块 )
固定工作单位
**套件@: ** . 固定工作
固定工人人才库,在短期工作协调员的周围有Satts.最小API包装纸*.
using Mostlylucid.Ephemeral.Atoms.FixedWork;
await using var atom = new FixedWorkAtom<WorkItem>(
async (item, ct) => await ProcessAsync(item, ct),
maxConcurrency: 4,
maxTracked: 200);
await atom.EnqueueAsync(item);
// Get stats
var (pending, active, completed, failed) = atom.Stats();
Console.WriteLine($"Completed: {completed}, Failed: {failed}");
// Get recent operations
var snapshot = atom.Snapshot();
// Graceful shutdown
await atom.DrainAsync();
按键顺序
以可选的公平日程安排顺序处理@.
using Mostlylucid.Ephemeral.Atoms.KeyedSequential;
await using var atom = new KeyedSequentialAtom<Order, string>(
keySelector: order => order.CustomerId,
body: async (order, ct) => await ProcessOrder(order, ct),
maxConcurrency: 16,
perKeyConcurrency: 1, // Sequential per key
enableFairScheduling: true); // Prevent hot key starvation
await atom.EnqueueAsync(order1); // Customer A
await atom.EnqueueAsync(order2); // Customer A - waits for order1
await atom.EnqueueAsync(order3); // Customer B - parallel with A
var (pending, active, completed, failed) = atom.Stats();
await atom.DrainAsync();
信号提醒Atom
**套件@: ** 多数为 mostlucid. @ epheperal@ ._ atoms @ MS K2# signalaware
基于环境信号暂停或取消摄取@.
using Mostlylucid.Ephemeral.Atoms.SignalAware;
var sink = new SignalSink();
await using var atom = new SignalAwareAtom<WorkItem>(
async (item, ct) => await ProcessAsync(item, ct),
cancelOn: new HashSet<string> { "shutdown", "circuit-open" },
deferOn: new HashSet<string> { "backpressure.*" },
deferInterval: TimeSpan.FromMilliseconds(100),
maxDeferAttempts: 50,
signals: sink,
maxConcurrency: 8);
// Enqueue work
await atom.EnqueueAsync(item);
// Raise ambient signals
atom.Raise("backpressure.downstream"); // New items defer
sink.Raise("shutdown"); // New items rejected (returns -1)
await atom.DrainAsync();
Batching Atom 条形图
**套件@: ** 多数是卢西德语 @ . @ epheperal@ MS K1 @ atoms @ MPK2 @ battach
按大小或时间间隔将项目收集成批集@ . @ {
using Mostlylucid.Ephemeral.Atoms.Batching;
await using var atom = new BatchingAtom<LogEntry>(
onBatch: async (batch, ct) =>
{
Console.WriteLine($"Flushing {batch.Count} entries");
await FlushToDatabase(batch, ct);
},
maxBatchSize: 100,
flushInterval: TimeSpan.FromSeconds(5));
// Items are batched automatically
atom.Enqueue(new LogEntry("User logged in"));
atom.Enqueue(new LogEntry("Request received"));
// ... batch flushes when full OR after 5 seconds
重试阿tom
套件@: . retrolly 多数是Lucid@. @ epheperal_ MS K1# atoms @MSK 2retry
指数后卸重试包装@ . @
using Mostlylucid.Ephemeral.Atoms.Retry;
await using var atom = new RetryAtom<ApiRequest>(
async (req, ct) => await CallExternalApi(req, ct),
maxAttempts: 3,
backoff: attempt => TimeSpan.FromMilliseconds(100 * Math.Pow(2, attempt)),
maxConcurrency: 4);
// Automatically retries on failure with exponential backoff
// Attempt 1: immediate
// Attempt 2: 200ms delay
// Attempt 3: 400ms delay
await atom.EnqueueAsync(new ApiRequest("https://api.example.com"));
await atom.DrainAsync();
数据存储原子
套件@: . - 数据
共享存储原子的共享配置 @ ( @ DataStorageConfig, IDataStorageAtom<TKey, TValue>)+信号
和 PostgreSQL 适配器@.
using Mostlylucid.Ephemeral.Atoms.Data;
using Mostlylucid.Ephemeral.Atoms.Data.File;
var sink = new SignalSink();
var config = new DataStorageConfig
{
DatabaseName = "orders",
SignalPrefix = "save.data",
LoadSignalPrefix = "load.data",
DeleteSignalPrefix = "delete.data",
MaxConcurrency = 1
};
await using var storage = new FileDataStorageAtom<string, Order>(sink, config, "./orders");
storage.EnqueueSave("order-123", new Order { Id = "order-123", Total = 42.00m });
var loaded = await storage.LoadAsync("order-123");
使用相同 DataStorageConfig 与
Mostlylucid.Ephemeral.Atoms.Data.Sqlite 或
Mostlylucid.Ephemeral.Atoms.Data.Postgres
由 SQLite @/ Postgres.}属性工作可以订阅
saved.data.{dbname} 启动下游工作的信号 load.data.{dbname} 触发水合物缓存@. @%
分子Runner @ & @ AtomTrigger
**套件@: ** .atoms MS K2 molecules
由 MoleculeBlueprintBuilder 允许您定义原子@pay,}清册 @,shipping,
通知@)}当信号发生时应该运行 order.placed 到达. MoleculeRunner 监听触发器
创建共享模式@, MoleculeContext@ , 并执行您订阅开始时的每步@ MS K1 补全事件\ . 使用
AtomTrigger 当一个原子@ ' @%s 信号应该启动另一个协调器或分子\ .}#
var sink = new SignalSink();
var blueprint = new MoleculeBlueprintBuilder("order", "order.placed")
.AddAtom(async (ctx, ct) => await paymentCoordinator.EnqueueAsync(ctx.TriggerSignal.Key!, ct))
.AddAtom(async (ctx, ct) =>
{
ctx.Raise("order.payment.complete", ctx.TriggerSignal.Key);
await inventoryCoordinator.EnqueueAsync(ctx.TriggerSignal.Key!, ct);
})
.Build();
await using var runner = new MoleculeRunner(sink, new[] { blueprint }, serviceProvider);
using var trigger = new AtomTrigger(sink, "order.payment.complete", async (signal, ct) =>
{
await notificationCoordinator.EnqueueAsync(signal.Key!, ct);
});
sink.Raise("order.placed", key: "order-42");
分子步骤可增加信号@( @%ctx.Raise("order.shipping.start")) 所以系统其余部分会接收到
{\fn黑体\fs22\bord1\shad0\3aHBE\4aH00\fscx67\fscy66\2cHFFFFFF\3cH808080}你是谁?
流出缓存阿通
**套件@: ** . 滑动缓存
访问结果重置 TTL. {以滑动过期缓存 @ -}
using Mostlylucid.Ephemeral.Atoms.SlidingCache;
await using var cache = new SlidingCacheAtom<string, UserProfile>(
async (userId, ct) => await LoadUserProfileAsync(userId, ct),
slidingExpiration: TimeSpan.FromMinutes(5),
absoluteExpiration: TimeSpan.FromHours(1),
maxSize: 1000);
// First call: computes and caches
var profile = await cache.GetOrComputeAsync("user-123");
// Second call within 5 minutes: returns cached, resets TTL
var cached = await cache.GetOrComputeAsync("user-123");
// Try get without computation (still resets TTL on hit)
if (cache.TryGet("user-123", out var profile))
Console.WriteLine(profile.Name);
// Get stats
var stats = cache.GetStats();
Console.WriteLine($"Entries: {stats.TotalEntries}, Hot: {stats.HotEntries}");
挥发性行动
**套件@: ** . 挥发性
当他们发出一个致命信号时,立即放下操作 使窗口保持轻度重量 即使是在非常高的输气管下
原子监听可配置模式 @ (@ 默认 kill.*共享的) SignalSink# ,呼叫# IOperationEvictor.TryKill
仍然让协调员在进入前发射回声 / 最后的钩子
失踪=.
var sink = new SignalSink();
await using var coordinator = new EphemeralWorkCoordinator<JobItem>(
async (job, ct) => await QuickProcessAsync(job, ct),
new EphemeralOptions
{
Signals = sink,
EnableOperationEcho = true,
OperationEchoRetention = TimeSpan.FromSeconds(30)
});
using var volatileAtom = new VolatileOperationAtom(sink, coordinator);
// inside the job: emitter.Emit("kill.job");
将原子与 OperationEchoMaker/OperationEchoAtom 已经键入 *.echo.* 以小的 TTL’+d 信号在杀戮中幸存
记录@.
时长路缓存
套件@: 核心 **
mostlylucid.ephemeral) @ MS K1 自行-@ 优化缓存, 每次点击和扩展 TTL 时都滑动 TL 热键.
using Mostlylucid.Ephemeral;
var cache = new EphemeralLruCache<string, Widget>(new EphemeralLruCacheOptions
{
DefaultTtl = TimeSpan.FromMinutes(5),
HotKeyExtension = TimeSpan.FromMinutes(30),
HotAccessThreshold = 3,
MaxSize = 10_000,
SampleRate = 5 // emit 1 in 5 signals
});
var widget = await cache.GetOrAddAsync("widget:42", async key =>
{
var data = await LoadWidgetAsync(key);
return data!;
});
// Stats and signals to see how the cache self-focuses on hot keys
var stats = cache.GetStats(); // hot/expired counts, size
var signals = cache.GetSignals("cache.*"); // cache.hot/evict/miss/hit
{\fn黑体\fs22\bord1\shad0\3aHBE\4aH00\fscx67\fscy66\2cHFFFFFF\3cH808080}小费
MemoryCache可配置为滑动到期@ , @ 但它从未发出热/} 冷信号或扩展 TTL 热密钥@. @%EphemeralLruCache是核心软件包“% (” 和“ in”中的自定义默认默认值SqliteSingleWriter) 每当您想要缓存聚焦在活动工作集@. @% 1
窗口大小原子
**套件@: ** . 窗口大小
原子倾听着 window.size.* 或 window.time.* 命令“,”和调用
SignalSink.UpdateWindowSize 以便您能够从另一个协调器或信号监视器中扩展或缩小共享窗口@. @%
var sink = new SignalSink(maxCapacity: 250);
await using var atom = new WindowSizeAtom(sink);
sink.Raise("window.size.decrease:50");
sink.Raise("window.time.set:00:02:00");
分类原子
软件包@: @%
- 基基
- . - 分类学 ( 包括
MultiTaxonomyAtom)- 原子种类 @:
mostlylucid.ephemeral.atoms.taxonomy.sensor,.extractor,.embedder,.retriever,.proposer,
.constrainer,.ranker,.renderer,.coordinator,.feedback,.guard
符合分类学的通用原子类型 @(@sensor,}(缩略语) @MS K2嵌入器{,*检索机♪,提议者{,{stricer□,}{甲级},
每个原子在完成时发出打印信号,并携带一个Atom合同。
安装您需要的指定原子软件包@(}以下示例使用 proposr @).
使用 MultiTaxonomyAtom 当您需要将多种类型合并到一个合同中时\ .
using Mostlylucid.Ephemeral;
using Mostlylucid.Ephemeral.Atoms.Taxonomy;
public sealed record Proposal(string Text, double Confidence);
var sink = new SignalSink();
var proposer = new ProposerAtom<string, Proposal>(
sink,
async (prompt, ct) => new Proposal($"Echo: {prompt}", 0.42),
outputSignal: "proposal.created",
keySelector: prompt => prompt);
await proposer.EnqueueAsync("hello");
电扶梯Atom
将打印的MSMK0- 时间信号推广到耐用汇@._ 埃斯卡拉托Atoms 是长期产出的最佳方式 活的协调员...
using Mostlylucid.Ephemeral;
using Mostlylucid.Ephemeral.Atoms.Data.File;
using Mostlylucid.Ephemeral.Atoms.Escalator;
public sealed record EscalationPayload(string Kind, string? EvidenceId, double Confidence);
var sink = new SignalSink();
var typed = new TypedSignalSink<EscalationPayload>(sink);
await using var storage = new FileDataStorageAtom<string, EscalationPayload>(
sink,
new FileDataStorageConfig { DatabaseName = "signals" });
Func<SignalEvent<EscalationPayload>, CancellationToken, Task> audit = (evt, ct) =>
{
Console.WriteLine($"audit: {evt.Signal} {evt.Key}");
return Task.CompletedTask;
};
var targets = new[]
{
new EscalationTarget<EscalationPayload>(
"store",
(evt, ct) => storage.SaveAsync(evt.Key ?? evt.OperationId.ToString(), evt.Payload, ct)),
new EscalationTarget<EscalationPayload>("audit", audit)
};
await using var escalator = new EscalatorAtom<EscalationPayload>(
sink,
typed,
targets,
new EscalatorAtomOptions<EscalationPayload>
{
EscalateSignalPattern = "escalate.*",
EmitOnSuccess = "escalation.persisted",
EmitOnFailure = "escalation.failed"
});
typed.Raise("escalate.signal", new EscalationPayload("risk", "file-42", 0.81), key: "order-123");
回音器
抓取输入的“% “” 最后一个单词“ MS K1” , 该操作在被剪裁前就已发出“ I.” 。 原子保留着一个封闭的信号窗口
有效载荷 @ ( @ 匹配 ActivationSignalPattern / CaptureSignalPattern** 和何时 OperationFinalized 火能产生火灾
OperationEchoEntry<TPayload> 记录可以通过 OperationEchoAtom<TPayload>.
var sink = new SignalSink();
var typedSink = new TypedSignalSink<EchoPayload>(sink);
var echoAtom = new OperationEchoAtom<EchoPayload>(async echo => await repository.AppendAsync(echo));
await using var coordinator = new EphemeralWorkCoordinator<JobItem>(ProcessAsync);
using var maker = coordinator.EnableOperationEchoing(
typedSink,
echoAtom,
new OperationEchoMakerOptions<EchoPayload>
{
ActivationSignalPattern = "echo.capture",
CaptureSignalPattern = "echo.*",
MaxTrackedOperations = 128
});
typedSink.Raise("echo.capture", new EchoPayload("order-1", "archived"), key: "order-1");
属性任务只是以他们认为关键的任何状态提高输入的信号@ , @ 和创建者保持工作设置 在你坚持回声.时被捆绑
模式@( @ ready @ - @toQ-_uses @)
以信号为基础的环形圆环面包机
**套件@: ** . 解路器
使用信号历史窗口的静态断路器@.
using Mostlylucid.Ephemeral.Patterns.CircuitBreaker;
var breaker = new SignalBasedCircuitBreaker(
failureSignal: "api.failure",
threshold: 5,
windowSize: TimeSpan.FromSeconds(30));
// Check before making calls
if (breaker.IsOpen(coordinator))
{
var retryAfter = breaker.GetTimeUntilClose(coordinator);
throw new CircuitOpenException("Too many failures", retryAfter);
}
// Pattern matching variant
if (breaker.IsOpenMatching(coordinator, "error.*"))
throw new CircuitOpenException("Error pattern detected");
// Get current failure count
int failures = breaker.GetFailureCount(coordinator);
信号驱动器后压缩
**套件@: ** . refackpressure (后压)
使用后压信号自动延迟的深度管理队列@.
using Mostlylucid.Ephemeral.Patterns.Backpressure;
var sink = new SignalSink();
await using var coordinator = SignalDrivenBackpressure.Create<WorkItem>(
async (item, ct) => await ProcessAsync(item, ct),
sink,
maxConcurrency: 4);
// Enqueue work
await coordinator.EnqueueAsync(item);
// When downstream is slow
sink.Raise("backpressure.downstream"); // New work auto-defers
// When recovered
sink.Retract("backpressure.downstream"); // Work resumes
控制源流
**套件@: ** . 受管制的硫丹
用于控制平行的 Global & M+% per- key glate
using Mostlylucid.Ephemeral.Patterns.ControlledFanOut;
await using var fanout = new ControlledFanOut<string, Request>(
keySelector: req => req.TenantId,
body: async (req, ct) => await ProcessAsync(req, ct),
maxGlobalConcurrency: 100, // Total parallel across all tenants
perKeyConcurrency: 5); // Max 5 parallel per tenant
// Items for same tenant processed with limit
await fanout.EnqueueAsync(requestA); // Tenant1
await fanout.EnqueueAsync(requestB); // Tenant1 - waits if 5 already running
await fanout.EnqueueAsync(requestC); // Tenant2 - parallel with Tenant1
await fanout.DrainAsync();
适应服务
**套件@: ** 多数情况下,.-epheperal ._patterns @.}适应剂
信号@- @% 驱动率限制, 自动反向@ .} @
using Mostlylucid.Ephemeral.Patterns.AdaptiveRate;
await using var service = new AdaptiveRateService<ApiRequest>(
async (req, ct) => await CallApiAsync(req, ct),
maxConcurrency: 8);
// Process with automatic rate limit handling
await service.ProcessAsync(request);
// When API returns 429, emit signal with retry-after
// Signal: "rate-limit:500ms"
// Service auto-parses and delays
Console.WriteLine($"Pending: {service.PendingCount}, Active: {service.ActiveCount}");
动态货币化
**套件@: ** . 动态货币
基于负载信号的运行时间货币缩放@. @%
using Mostlylucid.Ephemeral.Patterns.DynamicConcurrency;
var sink = new SignalSink();
await using var demo = new DynamicConcurrencyDemo<WorkItem>(
async (item, ct) => await ProcessAsync(item, ct),
sink,
minConcurrency: 2,
maxConcurrency: 32,
scaleUpPattern: "load.high",
scaleDownPattern: "load.low");
await demo.EnqueueAsync(item);
// Concurrency adjusts automatically based on signals
sink.Raise("load.high"); // Concurrency doubles (up to max)
sink.Raise("load.low"); // Concurrency halves (down to min)
Console.WriteLine($"Current concurrency: {demo.CurrentMaxConcurrency}");
await demo.DrainAsync();
密钥优先权
**套件@: ** . - 关键优先信息
优先车道,有per-}订货保留=.
using Mostlylucid.Ephemeral.Patterns.KeyedPriorityFanOut;
await using var fanout = new KeyedPriorityFanOut<string, UserCommand>(
keySelector: cmd => cmd.UserId,
body: async (cmd, ct) => await HandleCommand(cmd, ct),
maxConcurrency: 32,
perKeyConcurrency: 1, // Sequential per user
maxPriorityDepth: 100);
// Normal lane
await fanout.EnqueueAsync(normalCommand);
// Priority lane - jumps the queue for that user
bool accepted = await fanout.EnqueuePriorityAsync(urgentCommand);
// Check lane depths
var counts = fanout.PendingCounts;
Console.WriteLine($"Priority: {counts.Priority}, Normal: {counts.Normal}");
await fanout.DrainAsync();
反活动Fan outputPipeline
**套件@: ** .活性芳烃
Two- 配有自动后压的阶段输油管
using Mostlylucid.Ephemeral.Patterns.ReactiveFanOut;
await using var pipeline = new ReactiveFanOutPipeline<WorkItem>(
stage2Work: async (item, ct) => await SlowProcessing(item, ct),
preStageWork: async (item, ct) => await FastPreprocessing(item, ct),
stage1MaxConcurrency: 8,
stage1MinConcurrency: 1,
stage2MaxConcurrency: 4,
backpressureThreshold: 32, // Throttle when stage2 has 32+ pending
reliefThreshold: 8); // Resume when stage2 drops below 8
await pipeline.EnqueueAsync(item);
// Stage1 auto-throttles when stage2 backs up
Console.WriteLine($"Stage1 concurrency: {pipeline.Stage1CurrentMaxConcurrency}");
Console.WriteLine($"Stage2 pending: {pipeline.Stage2Pending}");
await pipeline.DrainAsync();
信号异常探测器
**套件@: ** mostlucid@ . @ epheperal @ MS K1_ patterns @ MPK2 异常检测器
移动中 -窗口异常点检测@.
using Mostlylucid.Ephemeral.Patterns.AnomalyDetector;
var sink = new SignalSink();
var detector = new SignalAnomalyDetector(
sink,
pattern: "error.*",
threshold: 5,
window: TimeSpan.FromSeconds(10));
// Check for anomalies
if (detector.IsAnomalous())
{
Console.WriteLine("Anomaly detected! Too many errors.");
TriggerAlert();
}
// Get current match count
int errorCount = detector.GetMatchCount();
Console.WriteLine($"Errors in window: {errorCount}");
信号协调指南
**套件@: ** mostlucid. @ epheperal@ ._ patterns @ MS K2# 信号协调阅读
校对:Portnoy
using Mostlylucid.Ephemeral.Patterns.SignalCoordinatedReads;
// Run demo: readers pause when update signal is present
var result = await SignalCoordinatedReads.RunAsync(
readCount: 10,
updateCount: 1);
Console.WriteLine($"Reads: {result.ReadsCompleted}, Updates: {result.UpdatesCompleted}");
Console.WriteLine($"Signals: {string.Join(", ", result.Signals)}");
// Manual implementation:
var sink = new SignalSink();
await using var readers = new EphemeralWorkCoordinator<Query>(
body,
new EphemeralOptions
{
DeferOnSignals = new HashSet<string> { "update.in-progress" },
Signals = sink
});
// Readers auto-defer when update is running
sink.Raise("update.in-progress"); // Readers wait
sink.Raise("update.done"); // Readers resume
信号提示服务
具有进度信号的 HTTP 客户端@. @%
using Mostlylucid.Ephemeral.Patterns.SignalingHttp;
var httpClient = new HttpClient();
var request = new HttpRequestMessage(HttpMethod.Get, "https://example.com/large-file");
// Create an emitter from your coordinator
// (emitter is any ISignalEmitter - operations implement this)
byte[] data = await SignalingHttpClient.DownloadWithSignalsAsync(
httpClient,
request,
emitter);
// Signals emitted during download:
// - stage.starting
// - progress:0
// - stage.request
// - stage.headers
// - stage.reading
// - progress:25, progress:50, progress:75, progress:100
// - stage.completed
信号日志监视器
**套件@: ** mostlucid. @ epheperal@ ._ patroducts @ MS K2# 信号观察器
监视模式的信号窗口并触发回调@. @%
using Mostlylucid.Ephemeral.Patterns.SignalLogWatcher;
var sink = new SignalSink();
await using var watcher = new SignalLogWatcher(
sink,
onMatch: evt =>
{
Console.WriteLine($"Error detected: {evt.Signal} at {evt.Timestamp}");
AlertOps(evt);
},
pattern: "error.*",
pollInterval: TimeSpan.FromMilliseconds(200));
// Watcher runs in background, calling onMatch for each new error signal
sink.Raise("error.database"); // -> onMatch called
sink.Raise("error.timeout"); // -> onMatch called
sink.Raise("info.started"); // -> ignored (doesn't match pattern)
遥测信号Handler
**套件@: ** . 遥测仪
Open 遥测@/ @Application Insights 集成=%.
using Mostlylucid.Ephemeral.Patterns.Telemetry;
// Use in-memory for testing, or implement ITelemetryClient for real telemetry
var telemetry = new InMemoryTelemetryClient();
await using var handler = new TelemetrySignalHandler(telemetry);
// Wire up to coordinator
var options = new EphemeralOptions
{
OnSignal = signal => handler.OnSignal(signal)
};
// Signals are processed asynchronously
// - "error.*" signals -> TrackExceptionAsync
// - "perf.*" signals -> TrackMetricAsync
// - all signals -> TrackEventAsync
Console.WriteLine($"Queued: {handler.QueuedCount}");
Console.WriteLine($"Processed: {handler.ProcessedCount}");
Console.WriteLine($"Dropped: {handler.DroppedCount}");
// Check recorded events
var events = telemetry.GetEvents();
长窗口Demo
演示审计线索的大型窗口配置 @ .@
using Mostlylucid.Ephemeral.Patterns.LongWindowDemo;
// Configure coordinator with large tracking window
var options = new EphemeralOptions
{
MaxTrackedOperations = 10000,
MaxOperationLifetime = TimeSpan.FromHours(24)
};
信号反应显示大小写
**套件@: ** 多数为 mostlucid. @ epheperal@ ._ patterns @ MS K2$ 信号反应显示大小写
显示信号发送模式和回调@.
using Mostlylucid.Ephemeral.Patterns.SignalReactionShowcase;
// See source for signal dispatch examples
// Demonstrates OnSignal, OnSignalAsync, CancelOnSignals, DeferOnSignals
持久性信号窗口
**套件@: ** . - 持久性窗口
带有 SQLite 持久性 {-} 的信号窗口 <\ xB1\bord1\shad1\3cH2F2F3F}{fn方正黑体简体\fs18\b0}
using Mostlylucid.Ephemeral.Patterns.PersistentWindow;
await using var window = new PersistentSignalWindow(
"Data Source=signals.db",
flushInterval: TimeSpan.FromSeconds(30));
// On startup: restore previous signals
await window.LoadFromDiskAsync(maxAge: TimeSpan.FromHours(24));
// Raise signals as normal
window.Raise("order.completed", key: "order-service");
window.Raise("payment.processed", key: "payment-service");
// Query signals
var recentOrders = window.Sense("order.*");
// Signals automatically flush every 30 seconds
// Also flushes on dispose
// Get stats
var stats = window.GetStats();
Console.WriteLine($"In memory: {stats.InMemoryCount}, Flushed: {stats.LastFlushedId}");
依赖性注射
// Register in Startup/Program.cs
services.AddEphemeralWorkCoordinator<WorkItem>(
async (item, ct) => await ProcessAsync(item, ct),
new EphemeralOptions { MaxConcurrency = 8 });
// Named coordinators
services.AddEphemeralWorkCoordinator<WorkItem>("priority",
async (item, ct) => await ProcessPriorityAsync(item, ct));
// Inject and use
public class MyService(IEphemeralCoordinatorFactory<WorkItem> factory)
{
public async Task DoWork()
{
var coordinator = factory.CreateCoordinator();
await coordinator.EnqueueAsync(new WorkItem());
}
}
现代DI根可能更喜欢较短的帮手,例如 services.AddCoordinator<T>(...),
services.AddScopedCoordinator<T>(...),或 services.AddKeyedCoordinator<T, TKey>(...) 因为他们读起来像正常人
AddX 他们只是委派给Ephemeral -的帮助者,
Lane 隐藏电源@+密钥配置 @(_Semple config},隱藏電源@ )
mostlylucid.ephemeral.complete 捆包 PriorityWorkCoordinator 和按键的变体 ,,所以相同的 MS K1,AddCoorden\”
配置几条车道名称@,}给予他们可选的 MaxDepth,
并依靠协调员在 @“_slow}#”}车道之前排空 “@hot”}同时仍遵守任何要求
var sink = new SignalSink();
var lanes = new[]
{
new PriorityLane("hot:4", CancelOnSignals: new HashSet<string> { "maintenance" }),
new PriorityLane("normal"),
new PriorityLane("slow:2")
};
await using var coordinator = new PriorityWorkCoordinator<WorkItem>(
new PriorityWorkCoordinatorOptions<WorkItem>(
async (item, ct) => await processor.ProcessAsync(item, ct),
lanes,
new EphemeralOptions { Signals = sink }));
await coordinator.EnqueueAsync(new WorkItem("order-42"), laneName: "hot");
await coordinator.EnqueueAsync(new WorkItem("order-43"), laneName: "slow");
使用 PriorityKeyedWorkCoordinator 如果您也需要 per- key 命令 —}在 phup , @ but a 中仍然会发生航道决定
内建的-% in key 选择器保持每个分区顺序\ .
var keyed = new PriorityKeyedWorkCoordinator<WorkItem, string>(
new PriorityKeyedWorkCoordinatorOptions<WorkItem, string>(
order => order.CustomerId,
async (order, ct) => await processor.ProcessPerCustomerAsync(order, ct),
lanes,
new EphemeralOptions { Signals = sink }));
await keyed.EnqueueAsync(new WorkItem("order-42") { CustomerId = "A" }, laneName: "hot");
await keyed.EnqueueAsync(new WorkItem("order-44") { CustomerId = "B" }, laneName: "normal");
将协调者视为厨师 , 车道作为碗盘 MSSK1 和钥匙作为勺子,每个客户都由勺子来保管 保持工作顺序的隐藏电源@ , @ trottled,}和无额外仪式的信号@ MPK3 @ aware
目标框架
- .NET @6.0,#7.0,}8.0, @MS K4#10.0#
驾照许可证
公共域@)