# 短时间信号 - 将原子变成一个遥感网络
*一个微小的原始体 将同时进行的工作 变成一个协调的适应系统*
**"时空信号模式"**
2025-12-12T16:00
内 **[第一部分 第一部分](/blog/fire-and-dont-quite-forget-ephemeral-execution)** 我们建造了短时执行 - 捆绑, 私人, 自我清洁的同步工作流程。 In **[第二部分 第二部分](/blog/ephemeral-execution-library)** 我们把它变成一个可再使用的图书馆,有协调员、有钥匙的管道和DI集成。
本条增加了一个改变一切的小特点: **信号信号信号**.
## 举起手来!
这现在也出现在 大多为粗略的. 短暂的 Niget 软件包中 [20多个大多为lucid.ephears 模式和“原子”](https://www.nuget.org/packages?q=mostlylucid.ephemeral&includeComputedFrameworks=true&prerel=true&sortby=created-desc).
[](https://www.nuget.org/packages/mostlylucid.ephemeral)
[](../../UNLICENSE)
## 源源文件
完整源代码在 [基特Hub 库](https://github.com/scottgal/mostlylucid.atoms)
信号基础设施存在于:
|------|---------|
| [信号cs](https://github.com/scottgal/mostlylucidweb/blob/main/Mostlylucid/Helpers/Ephemeral/Signals.cs) | `SignalEvent`, `SignalRetractedEvent`, `SignalPropagation`, `SignalConstraints`, `SignalSink`, `ISignalEmitter`, `AsyncSignalProcessor` |
| [短期行动cs](https://github.com/scottgal/mostlylucidweb/blob/main/Mostlylucid/Helpers/Ephemeral/EphemeralOperation.cs) * 运行中的信号排放和撤回
| [时间选择cs](https://github.com/scottgal/mostlylucidweb/blob/main/Mostlylucid/Helpers/Ephemeral/EphemeralOptions.cs) `CancelOnSignals`, `DeferOnSignals`, `OnSignal`, `OnSignalRetracted`) |
| [书签父母](https://github.com/scottgal/mostlylucidweb/blob/main/Mostlylucid/Helpers/Ephemeral/StringPatternMatcher.cs) 用于信号过滤的球式模式匹配
| [信号发送器cs](https://github.com/scottgal/mostlylucidweb/blob/main/Mostlylucid/Helpers/Ephemeral/SignalDispatcher.cs) 使用模式匹配的 Async 信号路由( 支持) `*`, `?`逗号列表, 确定性顺序)
| [实例/信号HttpClient.cs](https://github.com/scottgal/mostlylucidweb/blob/main/Mostlylucid/Helpers/Ephemeral/Examples/SignalingHttpClient.cs) HTTP电话的精细颗粒信号排放样本
| [实例/调整性翻译服务cs](https://github.com/scottgal/mostlylucidweb/blob/main/Mostlylucid/Helpers/Ephemeral/Examples/AdaptiveTranslationService.cs) 适应率受基于信号的推迟限制
| [实例/基于信号的环环 道 环 环 环 环 环 箱.cs](https://github.com/scottgal/mostlylucidweb/blob/main/Mostlylucid/Helpers/Ephemeral/Examples/SignalBasedCircuitBreaker.cs) 巡回断路器读取时空信号窗口
| [实例/遥测信号手法cs](https://github.com/scottgal/mostlylucidweb/blob/main/Mostlylucid/Helpers/Ephemeral/Examples/TelemetrySignalHandler.cs) 使用遥测集成的 Async 信号处理
[TOC]
---
## 问题:孤立的原子
我们的时尚协调员善于处理工作,但他们是孤立的。每个协调员都知道自己的行动,但不知道系统其他地方正在发生什么。
```csharp
// Translation coordinator has no idea that...
await translationCoordinator.EnqueueAsync(request);
// ...the API just hit a rate limit
// ...another service is experiencing backpressure
// ...a downstream dependency is slow
```
我们可以把明显的依附关系连接起来,但是这会造成联结。我们想要的是, **环境意识** - 协调员能够感知其环境而不直接相连。
让处决原子留下痕迹的信号 在其短暂的窗口。这些痕迹表现得像短命的事实:
- *"这只限利率的API"*
- *"这个用户三次失败"*
- *"一个兄弟姐妹行动还在门口等着"*
然后协调员可以根据窗口中可见的信号改变行为。这是环境意识,没有依赖性。
---
## 解决方案:行动信号
发自 [信号cs](https://github.com/scottgal/mostlylucidweb/blob/main/Mostlylucid/Helpers/Ephemeral/Signals.cs):
```csharp
public readonly record struct SignalEvent(
string Signal,
long OperationId,
string? Key,
DateTimeOffset Timestamp,
SignalPropagation? Propagation = null)
{
public int Depth => Propagation?.Depth ?? 0;
public bool WouldCycle(string signal) => Propagation?.Contains(signal) == true;
public bool Is(string name) => Signal == name;
public bool StartsWith(string prefix) => Signal.StartsWith(prefix, StringComparison.Ordinal);
}
```
操作可以在执行过程中发出信号。 这些信号存在于运行的时空窗口中。 当操作结束时, 信号随之而去 。
就是这样,没有信息经纪人,没有单独的基础设施,只有操作的附加条件
由于信号存在于时空窗口内,它们继承了它的保证:界限大小、自动老化和零生命周期间接费用。
---
# 时间信号法
## 法律1 - 信号从不暗中触发执行
信号本身并不导致处决。
它只记录在时尚窗口中的事实
没有什么运行是因为信号被发出。
## 第2号法律 - 明确、地方和可选的信号法
如果协调员给OnSignal处理器下定义,当发出信号时它同步运行,但仅仅因为协调员选择附加信号。
施舍者不知道或不关心。
取消所有处理者,核心行为不变。
## 第3号法律 -- -- 局部到原子的信号
信号只附在发射它的原子/操作上。
从来没有信号变异 或注解另一个原子。
没有共享的可执行公交车。
## 第4号法律 - 信号是仅附加事实
发送信号是原子历史的附加事实。
信号从未更新或覆盖过。
减量只能消除排放者自己的信号。
## 第5号法律 -- -- 信号表面断裂和时间限制
信号只存在于协调员的时空窗口内。
它们随着窗口的耗尽而自动失效。
没有什么是持久的,除非你明确坚持下去。
## 法律6 - 观察者阅读,他们不写
当协调员“检查关键 K 的信号”时,它扫描:
窗口中的原子
以及当地对这些原子的信号
观察者从未改变原子的信号状态。
## 第7号法律 - 类似事件的行为是一个层,不是原始行为
如果您想要信号驱动同步工作流程, 您必须使用 :
信号发送器
Async 信号处理器
或其他适配器。
这些是建立在信号之上的可选层, 而不是其语义的一部分。
## 第8号法律 -- -- 任何情况下不得发生跨原子变异
任何原子都不能改变另一个原子的快照、状态、信号或元数据。
通过以下方式进行协调:
- 信号信号信号
- 遥感
- 窗口窗口窗口窗口
- 政策、政策、政策、政策和政策
**不是通过写作。**
## 第9号法律 -- -- 全球观点是衍生的,不会变异
信号汇或群聚表面可能显示对信号的综合观察——
但它总是只读,从不权威,从不写字。
## 第10号法律 - 清除手持者
如果所有处理器(信号、调度器、处理器)都分离,
该系统仍然完全正确和可预测。
信号仍然有意义,因为它们是事实,不是触发物。
### 最佳分析:脚印,不是指示
**事件事件事件事件** 就像是一个电话:
> "我现在就给你打电话 接电话反应"
**信号** 象雪中的脚印一样。
> "我留下了脚印,如果你想知道我去哪里..."
这就是为什么信号从不断断,从不阻断, 从不与控制流互动,除非 *选择* 来调查他们。
### 此删除是什么
区别很重要,因为信号消除了:
问题 事件 信号 避免它
|---------|:--------------:|:----------------:|
时间依赖性 即时反应 需要 环境, 民意调查,一旦准备好
* 命令语义 * * 秩序事务 * * 不相关 * * * 命令语义 * * 秩序事务 *
交付保证必须交付/处理 没有交付,只是存在
* 传播错误 * * 处理错误 * 传播错误 * 孤立 * * 传播错误 * * 传播错误 * 传播错误 * 传播错误 * 传播错误 * 传播错误 * 传播错误 * 传播错误 * 传播错误 * 传播错误 * 传播错误 * 处理错误 * 传播错误 * 传播错误 * 孤立 * * 传播错误 * 传播错误 * 传播错误 * 传播错误 * 传播错误 * 传播错误 *
* 隐居危险 * * 常见的 * * 不可能 * * * 隐居危险 * * 常见的 * * 不可能 * * * * 隐居危险 * * * 隐居危险 * * 无法 * * * 隐居危险 * * * 隐居危险 * * 常见的 * * 无法 * * * 无法 * * * * * * * 隐居危险 * * * 无法 * * * * * * * 隐居危险 * * * * * * * * * * 无法 * * * * * * * * * 无法 * * * * *
连锁故障 一个处理器故障 链断 没有链断
### 快速比较
特写事件 信号
|---------|--------|---------|
即时时间 即时 即时 即时 即时 即时 即时 即时 即时 即时 即时 即时 即时 即时 即时 即时 即时 即时 即时 即时
保证/试图 没有交付,只是存在
回应 要求 任选 任选 任选
* 数据有效载荷 * * 通常重 * * 小字符串元数据 * * * * * 数据有效载荷 * * 数据有效载荷 * * 数据经常重 * 细字符串元数据 * * * 数据有效载荷 * * 数据有效载荷 * * 数据有效载荷 * * 数据经常重 * 细字符串元数据 *
儲存 LRU 式視窗
一生 即时 衰变 自动 衰变 衰变 衰变 衰变 衰变 衰变 衰变 衰变 衰变 衰变 衰变 衰变 衰变 衰变 衰变 衰变 衰变 衰
* 失败模式 * * 许多 * * 几乎没有 * * * * * 失败模式 * * * 失败模式 * * 失败模式 * * * 许多失败模式 * * * 几乎无成功模式 * * * * * * 失败模式 * * * 失败模式 * * * 失败模式 * * * 许多失败模式 * * * * 几乎无成功模式 * * * * * * * * 失败模式 * * * * * * * 失败模式 * * * * * * * 失败模式 * * * * * * * * * * 失败模式 * * * * * * * * * * * * * * * * * * * * 几乎无失败模式 * * * * * * * * * * * * * * * * * * * * * 失败模式 * * * * * * * * * * * * * 失败模式 * * * * * * * * * * * * * * * * * 失败模式 * * * * * * * * * * *失败模式 * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
### 《守则》差异
**事件处理方法(经典):**
```csharp
public event Action RateLimited;
try
{
await CallApiAsync();
}
catch (RateLimitException)
{
RateLimited?.Invoke(); // Makes someone else act right now
}
```
问题:
- 谁的反应?
- 按什么顺序排列?
- 如果两个处理者有冲突呢?
- 如果一个操作员扔球呢?
- 如果时机不对呢?
- 如果打电话的人不想这样呢?
**信号接近( 短暂) :**
```csharp
try
{
await CallApiAsync(ct);
}
catch (RateLimitException ex)
{
op.Signal($"rate-limit:{ex.RetryAfterMs}ms");
throw;
}
```
这里什么也没有发生 后来 完全不一样的地方
```csharp
if (translator.HasSignal("rate-limit"))
{
await Task.Delay(1000); // Act when *we* choose
}
```
注:
- 不直接连接
- 无惊喜电话
- 没有执行跳跳
- 没有回回回地狱的地狱
- 没有交叉线怪异
- 只有当我们 *问 问 问 问* 查看
### "啊哈!" 线条
> **事件传输控制 信号传输环境**
这是整个心理模式 在一句话。
### 真正信号是什么
取决于您的背景 :
* 听众 * 定义 *
|----------|------------|
| **系统思考者** * 间接协调的基底基底子 *
| **工程师** 轻量级元数据附在一个捆绑的滑动窗口中的操作中
| **PL/Con 货币呆子** 隐含的,时间上的黑板 与捆绑的通货货币语义搭配
| **框架用户** 处理过程中的自我清洁状态表面 你随时可以查询
信号是共享的、紧凑的内存表面留下的痕迹。 任何人都可以查看它们。 没有人有反应的义务。 这是根本的。 *悬浮* 协调模式 -- -- 同一种蚂蚁使用,同一种在早期AI中使用的黑板系统,同一种现代CRDT八卦网络的提示。
---
## 此方法如何与其他方法相比
### 应用程序 Insights / Open 遥测
```csharp
using var activity = source.StartActivity("ProcessOrder");
activity?.SetTag("order.id", orderId);
activity?.SetTag("rate.limited", true);
```
**最佳**:在服务、长期遥测存储、相关身份等各处分布追踪。
**使用遥测时使用遥测方法**: 您需要跟踪多个服务的请求, 存储用于分析的量度, 或与监测工具整合 。
**使用时间信号**:您需要过程内的环境意识, 反应协调, 或者不需要遥测基础设施。
### 反应扩展( Rx)
```csharp
var rateLimits = Observable.FromEventPattern(
h => api.RateLimitHit += h,
h => api.RateLimitHit -= h);
rateLimits
.Throttle(TimeSpan.FromSeconds(1))
.Subscribe(e => HandleRateLimit(e));
```
**最佳**:复杂事件处理、基于时间的业务,结合多个事件流。
**使用 Rx 当**: 您需要复杂的时间查询( 窗口、 拆卸、 组合流) 。
**使用时间信号**:您需要更简单的基于投票的感应、自动清理或与行动跟踪整合。
### 中继R 通知
```csharp
public class RateLimitNotification : INotification
{
public int RetryAfterMs { get; init; }
}
await _mediator.Publish(new RateLimitNotification { RetryAfterMs = 5000 });
```
**最佳**:与多个处理器分离处理处理处理过程中的事件。
**使用中位tR**:您想要多个处理器对同一事件同步反应 。
**使用时间信号**:您想要没有明确订阅、自我清理历史或与捆绑处决结合的环境遥感。
### Polly 电路断开器
```csharp
var circuitBreaker = Policy
.Handle()
.CircuitBreakerAsync(5, TimeSpan.FromSeconds(30));
```
**最佳**:通过自动国家管理,应对个人电话的复原力。
**使用 Polly 时**:您需要每次呼叫的复原力,自动进行半开放/封闭的过渡。
**使用时间信号**:您想要通过许多操作, 定制电路逻辑, 或与操作跟踪整合, 来提高环境意识 。
**结合他们**:在工作体内使用Polly,在电路出行时发出信号。
### 比较比较表
* 自我清除 * 环境感应 * 破碎 * 自定义逻辑 * 整合 *
|----------|:-------------:|:---------------:|:---------:|:------------:|:-----------:|
外在工具 @ @ openTelestries @ @ @ @ @ @ @ @ _
反应扩展 \
《中位数》 《手册》 《手册》 《手册》
| **短讯讯号**
---
## 升升和减缩信号
业务实施 `ISignalEmitter`:
```csharp
public interface ISignalEmitter
{
// Emit signals
void Emit(string signal);
bool EmitCaused(string signal, SignalPropagation? cause);
// Retract (remove) signals
bool Retract(string signal);
int RetractMatching(string pattern);
bool HasSignal(string signal);
long OperationId { get; }
string? Key { get; }
}
```
在你工作的身体里:
```csharp
await coordinator.ProcessAsync(async (item, op, ct) =>
{
try
{
var result = await CallExternalApiAsync(item, ct);
if (result.WasCached)
op.Signal("cache-hit");
if (result.Duration > TimeSpan.FromSeconds(2))
op.Signal("slow-response");
}
catch (RateLimitException ex)
{
op.Signal("rate-limit");
op.Signal($"rate-limit:{ex.RetryAfterMs}ms");
throw;
}
catch (TimeoutException)
{
op.Signal("timeout");
throw;
}
});
```
信号只是字符串。使用简单的名称(`"rate-limit"`或结构化名称(`"rate-limit:5000ms"`).
模式过滤器使用 Glob 语义( Glob 语义) 。`*`, `?`以及支持逗号清单(`"error.*,timeout"`配对是决定性的,通过 `StringPatternMatcher`.
### 精密信号示例: HTTP 调用
对于非常详细的可观测性,您可以在操作的每个阶段发出信号。图书馆包括一个样本。 [信号提示服务](https://github.com/scottgal/mostlylucidweb/blob/main/Mostlylucid/Helpers/Ephemeral/Examples/SignalingHttpClient.cs) 证明这一模式:
```csharp
using Mostlylucid.Helpers.Ephemeral.Examples;
// Inside your work body where you have access to the operation's emitter:
await coordinator.ProcessAsync(async (request, op, ct) =>
{
var data = await SignalingHttpClient.DownloadWithSignalsAsync(
httpClient,
new HttpRequestMessage(HttpMethod.Get, request.Url),
op, // ISignalEmitter
ct);
// Process the downloaded data...
});
```
这在每个阶段发出信号:
何时发出信号?
|--------|------|
| `stage.starting` 在请求开始之前
| `progress:0` 初步进展标志
| `stage.request` * HTTP 请求已发送 * * * * HTTP 请求已发送 * * * * HTTP 请求已发送 * * HTTP 请求已发送 * * HTTP 请求已发送 * * HTTP 请求已发送 * HTTP 请求已发送 *
| `stage.headers` 收到回应信头
| `stage.reading` * 开始阅读正文 * * * 开始阅读正文 * * * 开始阅读正文 * * * 开始阅读正文 * * 开始阅读正文 * * * 开始阅读正文 * * * 开始阅读正文 * * 开始阅读正文 * * 开始阅读正文 * * 开始阅读正文 * * * 开始阅读正文 *
| `progress:XX` 下载期间的进度百分比(0-100)
| `stage.completed` * 下载完成 * * * * * * 下载完成 * * * * * * 下载完成 * * * * * * 下载完成 * * * * * * 下载完成 * * * * * * 下载完成 * * * * * * * 下载完成 * * * * * * * * 下载完成 * * * * * * * * * 下载完成 * * * * * * * * * *
然后,您可以用匹配的图案查询这些图案 :
```csharp
// Find all stage transitions
var stages = coordinator.GetSignalsByPattern("stage.*");
// Check download progress
var progress = coordinator.GetSignalsByPattern("progress:*");
// Check if any download is still in progress
if (coordinator.HasSignalMatching("stage.reading") &&
!coordinator.HasSignalMatching("stage.completed"))
{
// Download in progress
}
```
### 正在压缩信号
操作也可以删除自己的信号。 这对临时状态有用 :
```csharp
await coordinator.ProcessAsync(async (item, op, ct) =>
{
// Mark as processing
op.Emit("processing");
try
{
await ProcessItemAsync(item, ct);
// Success - retract the processing signal
op.Retract("processing");
op.Emit("completed");
}
catch (RetryableException)
{
// Keep processing signal, add retry info
op.Emit("retrying");
}
catch (Exception)
{
// Remove all temporary signals
op.RetractMatching("processing*");
op.Emit("failed");
throw;
}
});
```
### 递减事件
就像信号发射一样,撤回可以触发回击:
```csharp
var coordinator = new EphemeralWorkCoordinator(
body,
new EphemeralOptions
{
// Sync retraction handler
OnSignalRetracted = evt =>
{
_metrics.DecrementGauge(evt.Signal);
Console.WriteLine($"Signal {evt.Signal} retracted from op {evt.OperationId}");
if (evt.WasPatternMatch)
Console.WriteLine($" (matched pattern: {evt.Pattern})");
},
// Async retraction handler
OnSignalRetractedAsync = async (evt, ct) =>
{
await _telemetry.TrackRetraction(evt.Signal, evt.OperationId, ct);
}
});
```
缩略 `SignalRetractedEvent` 包括:
- `Signal` - 撤回的信号名
- `OperationId` - 收回它的行动
- `Key` - 手术的钥匙(如果有的话)
- `Timestamp` - 发生撤回时
- `WasPatternMatch` - 如果通过 `RetractMatching`
- `Pattern` - 所用模式(如果模式匹配)
### 真实世界实例:回收率限制
```csharp
await coordinator.ProcessAsync(async (request, op, ct) =>
{
// Check if we already have a rate limit signal
if (op.HasSignal("rate-limited"))
{
// We're in recovery mode
await Task.Delay(1000, ct);
}
try
{
var response = await _api.SendAsync(request, ct);
// Success! Remove any rate limit signal
if (op.Retract("rate-limited"))
{
op.Emit("rate-limit-cleared");
}
}
catch (RateLimitException ex)
{
op.Emit("rate-limited");
op.Emit($"rate-limit:{ex.RetryAfterMs}ms");
throw;
}
});
```
---
## 遥感信号
所有协调员都提供最佳信号查询:
```csharp
// Check if any recent operation hit a rate limit
if (coordinator.HasSignal("rate-limit"))
{
await Task.Delay(1000);
}
// Count slow responses in the window
var slowCount = coordinator.CountSignals("slow-response");
if (slowCount > 10)
{
await ThrottleAsync();
}
// Get signals by pattern
var httpErrors = coordinator.GetSignalsByPattern("http.error.*");
// Get signals since a time
var recentSignals = coordinator.GetSignalsSince(DateTimeOffset.UtcNow.AddMinutes(-1));
// Get signals for a specific key
var userSignals = coordinator.GetSignalsByKey("user-123");
```
---
## 信号反应处理
发自 [时间选择cs](https://github.com/scottgal/mostlylucidweb/blob/main/Mostlylucid/Helpers/Ephemeral/EphemeralOptions.cs):
协调员可对信号自动作出反应:
```csharp
var coordinator = new EphemeralWorkCoordinator(
body,
new EphemeralOptions
{
// Cancel new work if these signals are present
CancelOnSignals = new HashSet { "system-overload", "circuit-open" },
// Defer new work while these signals are present
DeferOnSignals = new HashSet { "rate-limit" },
MaxDeferAttempts = 10,
DeferCheckInterval = TimeSpan.FromMilliseconds(100)
});
```
当信号在 `CancelOnSignals` 检测到,新项目被跳过(算作失败) 。
当信号在 `DeferOnSignals` 被检测到,新的项目将等待信号清除。
---
## 现实世界实例:适应率限制
发自 [适应性转换服务cs](https://github.com/scottgal/mostlylucidweb/blob/main/Mostlylucid/Helpers/Ephemeral/Examples/AdaptiveTranslationService.cs):
```csharp
public class AdaptiveTranslationService : IAsyncDisposable
{
private readonly EphemeralWorkCoordinator _coordinator;
private readonly ITranslationApi _translationApi;
public AdaptiveTranslationService(ITranslationApi translationApi)
{
_translationApi = translationApi;
_coordinator = new EphemeralWorkCoordinator(
ProcessTranslationAsync,
new EphemeralOptions
{
MaxConcurrency = 8,
MaxTrackedOperations = 100,
// New work is deferred while any "rate-limit" or "rate-limit:*" signal is present
DeferOnSignals = new HashSet { "rate-limit", "rate-limit:*" },
MaxDeferAttempts = 10,
DeferCheckInterval = TimeSpan.FromMilliseconds(100)
});
}
public async Task TranslateAsync(TranslationRequest request)
{
// Optional: extra politeness based on most recent retry-after
var rateLimitSignals = _coordinator.GetSignalsByPattern("rate-limit:*");
if (rateLimitSignals.Count > 0)
{
var latest = rateLimitSignals
.OrderByDescending(s => s.Timestamp)
.First()
.Signal; // "rate-limit:5000ms"
if (TryParseRetryAfter(latest, out var delay))
{
await Task.Delay(delay);
}
}
await _coordinator.EnqueueAsync(request);
}
public static bool TryParseRetryAfter(string signal, out TimeSpan delay)
{
delay = default;
var parts = signal.Split(':', 2);
if (parts.Length != 2) return false;
var payload = parts[1].Trim();
if (!payload.EndsWith("ms", StringComparison.OrdinalIgnoreCase)) return false;
var numPart = payload[..^2];
if (!int.TryParse(numPart, out var ms) || ms < 0) return false;
delay = TimeSpan.FromMilliseconds(ms);
return true;
}
}
```
当利率限制被击中时,这项服务的每个实例都会自动退缩。 没有共享状态, 没有消息传递。 正在读取时间窗口 。
---
## 现实世界实例:交叉协调员认识
多个协调员可以通过共同的交流相互感觉 `SignalSink`:
```csharp
public class OrderProcessingSystem
{
private readonly SignalSink _sharedSignals = new(maxCapacity: 1000);
private readonly EphemeralWorkCoordinator _orderProcessor;
private readonly EphemeralWorkCoordinator _paymentProcessor;
public OrderProcessingSystem()
{
var options = new EphemeralOptions { Signals = _sharedSignals };
_orderProcessor = new EphemeralWorkCoordinator(
ProcessOrderAsync, options);
_paymentProcessor = new EphemeralWorkCoordinator(
ProcessPaymentAsync, options);
}
public async Task ProcessOrderAsync(Order order)
{
// Check shared signals for payment gateway issues
if (_sharedSignals.Detect("gateway-error"))
{
await _retryQueue.EnqueueAsync(order);
return;
}
await _orderProcessor.EnqueueAsync(order);
}
}
```
---
## 真实世界实例:健康监测
```csharp
[HttpGet("/health/detailed")]
public IActionResult GetDetailedHealth()
{
return Ok(new
{
translation = new
{
pending = _translationCoordinator.PendingCount,
active = _translationCoordinator.ActiveCount,
recentRateLimits = _translationCoordinator.CountSignals("rate-limit"),
recentTimeouts = _translationCoordinator.CountSignals("timeout"),
recentSuccess = _translationCoordinator.CountSignals("success"),
hasErrors = _translationCoordinator.HasSignalMatching("error.*")
},
payment = new
{
pending = _paymentCoordinator.PendingCount,
gatewayErrors = _paymentCoordinator.CountSignals("gateway-error"),
declines = _paymentCoordinator.CountSignals("declined"),
approvals = _paymentCoordinator.CountSignals("approved")
}
});
}
```
不需要量度库。 只需查询时钟窗口 。
---
## 真实世界实例:基于信号的电路断裂
发自 [基于信号的环形环形车列车cs](https://github.com/scottgal/mostlylucidweb/blob/main/Mostlylucid/Helpers/Ephemeral/Examples/SignalBasedCircuitBreaker.cs):
```csharp
public class SignalBasedCircuitBreaker
{
private readonly string _failureSignal;
private readonly int _threshold;
private readonly TimeSpan _windowSize;
public SignalBasedCircuitBreaker(
string failureSignal = "failure",
int threshold = 5,
TimeSpan? windowSize = null)
{
_failureSignal = failureSignal;
_threshold = threshold;
_windowSize = windowSize ?? TimeSpan.FromSeconds(30);
}
public bool IsOpen(EphemeralWorkCoordinator coordinator)
{
var recentFailures = coordinator.GetSignalsSince(
DateTimeOffset.UtcNow - _windowSize);
return recentFailures.Count(s => s.Signal == _failureSignal) >= _threshold;
}
public int GetFailureCount(EphemeralWorkCoordinator coordinator)
{
var recentFailures = coordinator.GetSignalsSince(
DateTimeOffset.UtcNow - _windowSize);
return recentFailures.Count(s => s.Signal == _failureSignal);
}
}
// Usage
var circuitBreaker = new SignalBasedCircuitBreaker("api-error", threshold: 3);
if (circuitBreaker.IsOpen(_coordinator))
{
throw new CircuitOpenException("Too many recent API errors");
}
await _coordinator.EnqueueAsync(request);
```
断路器没有自己的状态 它只读到时钟窗口
---
## 信号限制:防止无限环圈
发自 [信号cs](https://github.com/scottgal/mostlylucidweb/blob/main/Mostlylucid/Helpers/Ephemeral/Signals.cs):
当信号能引发其他信号时 你就会冒着无穷无尽的循环风险 `SignalConstraints` 防止:
```csharp
var options = new EphemeralOptions
{
SignalConstraints = new SignalConstraints
{
// Max propagation depth before blocking
MaxDepth = 10,
// Prevent A → B → A cycles
BlockCycles = true,
// Signals that end propagation chains
TerminalSignals = new HashSet { "completed", "failed", "resolved" },
// Signals that emit but don't propagate
LeafSignals = new HashSet { "logged", "metric" },
// Callback when a signal is blocked
OnBlocked = (signal, reason) =>
{
_logger.LogWarning("Signal {Signal} blocked: {Reason}",
signal.Signal, reason);
}
}
};
```
### 信号传动
音轨因果关系 `EmitCaused`:
```csharp
public void HandleSignal(SignalEvent evt, ISignalEmitter emitter)
{
if (evt.Is("order-placed"))
{
// This signal carries the propagation chain
// Will be blocked if it would create a cycle
emitter.EmitCaused("inventory-reserved", evt.Propagation);
}
}
```
传播链跟踪路径: `order-placed → inventory-reserved → ...`
如果(如果) `inventory-reserved` 尝试发送 `order-placed`,就会被堵住(检测到循环)。
---
## 信号辛克:全球信号空间
发自 [信号cs](https://github.com/scottgal/mostlylucidweb/blob/main/Mostlylucid/Helpers/Ephemeral/Signals.cs):
为了让所有协调员都能看到需要显示的信号:
```csharp
public sealed class SignalSink
{
private readonly ConcurrentQueue _window;
private readonly int _maxCapacity;
private readonly TimeSpan _maxAge;
public SignalSink(int maxCapacity = 1000, TimeSpan? maxAge = null);
// Raise signals
public void Raise(SignalEvent signal);
public void Raise(string signal, string? key = null);
// Sense signals
public IReadOnlyList Sense();
public IReadOnlyList Sense(Func predicate);
public bool Detect(string signalName);
public bool Detect(Func predicate);
public int Count { get; }
}
```
用法 :
```csharp
// Create a shared sink
var sink = new SignalSink(maxCapacity: 1000, maxAge: TimeSpan.FromMinutes(2));
// Configure coordinators to use it
var options = new EphemeralOptions { Signals = sink };
// Or raise signals directly
sink.Raise("system-maintenance");
// Sense from anywhere
if (sink.Detect("system-maintenance"))
{
await DeferWorkAsync();
}
```
---
## 与字符串匹配模式
发自 [书签父母](https://github.com/scottgal/mostlylucidweb/blob/main/Mostlylucid/Helpers/Ephemeral/StringPatternMatcher.cs):
用于信号过滤的球式匹配 :
```csharp
// Exact match
coordinator.HasSignal("rate-limit");
// Wildcard patterns
coordinator.HasSignalMatching("http.*"); // http.timeout, http.error
coordinator.HasSignalMatching("error.*.critical"); // error.payment.critical
coordinator.HasSignalMatching("user-???-failed"); // user-123-failed
// Comma-separated patterns in CancelOnSignals/DeferOnSignals
new EphemeralOptions
{
CancelOnSignals = new HashSet
{
"system-overload, circuit-open", // Either pattern
"error.*" // Any error signal
}
}
```
---
## 信号命名公约
保持信号简单和一致:
```csharp
// Good - simple, categorical
op.Signal("success");
op.Signal("failure");
op.Signal("rate-limit");
op.Signal("timeout");
op.Signal("cache-hit");
// Good - structured for parsing
op.Signal("rate-limit:5000ms");
op.Signal("retry:attempt-3");
op.Signal("slow:2500ms");
op.Signal("http.error:429");
// Good - hierarchical for pattern matching
op.Signal("payment.declined");
op.Signal("payment.approved");
op.Signal("payment.gateway-error");
// Avoid - entity identification belongs in Key, not signals
op.Signal("user-123-rate-limited"); // Bad
// Instead
op.Key = "user-123";
op.Signal("rate-limit");
```
---
## Async 信号处理
同步信号处理器( C)`OnSignal`运行在操作的线条上, 让他们快速保持它们。 要完成 I/ O 任务, 将信号发送到同步路径 `SignalDispatcher` (模式匹配、确定性顺序)或 `AsyncSignalProcessor`.
### 发送信号
```csharp
await using var dispatcher = new SignalDispatcher(new EphemeralOptions
{
MaxConcurrency = Environment.ProcessorCount,
MaxConcurrencyPerKey = 1 // sequential per signal name by default
});
dispatcher.Register("error.*", evt => _alerts.SendAsync(evt.Signal));
dispatcher.Register("progress:*", evt => _metrics.Record(evt.Signal));
// In coordinator options, keep OnSignal chor options, keep OnSignal cheap and enqueue
var coordinator = new EphemeralWorkCoordinator(
body,
new EphemeralOptions
{
OnSignal = dispatcher.Dispatch
});
```
模式支助 `*`, `?`和逗号列表(`"error.*,timeout"`所有匹配的操作员均按登记顺序由背景钥匙协调员负责操作;发放时保持同步。
### Async 信号处理器
用于独立自动同步处理:
```csharp
await using var processor = new AsyncSignalProcessor(
async (signal, ct) =>
{
await _externalService.LogAsync(signal, ct);
},
maxConcurrency: 4,
maxQueueSize: 1000);
// Enqueue signals (returns immediately)
processor.Enqueue(new SignalEvent(
"rate-limit",
operationId,
key,
DateTimeOffset.UtcNow));
```
### 遥测信号Handler 示例
将合成信号处理与遥测集成相结合的完整实例:
> **资料来源:** [遥测信号Handler.cs](https://github.com/scottgal/mostlylucidweb/blob/main/Mostlylucid/Helpers/Ephemeral/Examples/TelemetrySignalHandler.cs)
```csharp
public class TelemetrySignalHandler : IAsyncDisposable
{
private readonly AsyncSignalProcessor _processor;
private readonly ITelemetryClient _telemetry;
public TelemetrySignalHandler(ITelemetryClient telemetry)
{
_telemetry = telemetry;
_processor = new AsyncSignalProcessor(
HandleSignalAsync,
maxConcurrency: 8,
maxQueueSize: 5000);
}
// Synchronous entry point - returns immediately
public bool OnSignal(SignalEvent signal) => _processor.Enqueue(signal);
private async Task HandleSignalAsync(SignalEvent signal, CancellationToken ct)
{
var properties = new Dictionary
{
["signal"] = signal.Signal,
["operationId"] = signal.OperationId.ToString(),
["key"] = signal.Key ?? "none"
};
await _telemetry.TrackEventAsync("EphemeralSignal", properties, ct);
// Categorized tracking based on signal prefix
if (signal.StartsWith("error"))
await _telemetry.TrackExceptionAsync(signal.Signal, properties, ct);
else if (signal.StartsWith("perf"))
await _telemetry.TrackMetricAsync(signal.Signal, 1, ct);
}
// Expose stats for monitoring
public int QueuedCount => _processor.QueuedCount;
public long ProcessedCount => _processor.ProcessedCount;
public long DroppedCount => _processor.DroppedCount;
public async ValueTask DisposeAsync() => await _processor.DisposeAsync();
}
```
把它发给你的协调员:
```csharp
await using var telemetryHandler = new TelemetrySignalHandler(telemetryClient);
await using var coordinator = new EphemeralWorkCoordinator(
ProcessAsync,
new EphemeralOptions
{
OnSignal = signal => telemetryHandler.OnSignal(signal)
});
```
处理器 :
- **永无区块** 操作线 - `OnSignal` 立即返回
- **信号分类** 不同的遥测类型前缀
- **度量指标** 监测健康(排队、处理、丢弃)
- **间距内存** - 如果队列填满,会减少最古老的信号
---
## 为何这样工作
信号是强大的 因为它们是 **短时间**:
财产 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴 津贴
|----------|---------|
| **环形** 无法不受约束地成长 - 旧信号时代熄灭 -
| **自我清洁** 不需要清理代码
| **分离** 光荣者不知道 听众
| **可观察** 任何代码都能感知到目前的状况
| **私营** 没有用户数据 - 只是信号名
| **快速快速** O(1)探测短路
时间窗口已经存在用于调试。 信号只是赋予它语义的含义 。
---
## 结论 结论 结论 结论 结论
信号将孤立的处决原子变成 **遥感网络**每位协调员可以:
- **埃米特** 有关其经历的信号
- **感知** 发自其本身历史或共有汇的信号
- **反应反应** 自动通过 `CancelOnSignals` 和 `DeferOnSignals`
没有信息代理,没有共享状态,没有协调协议,只有使用自然衰减的元数据操作。
原子不会直接和对方交谈, 它们只是把痕迹留在窗口里, 其他人可以观察它们。 它对于合成系统来说是一丝不苟的。
**火 信号 感知 忘记**
---
## 链接链接链接链接
- [第1部分:火和不要火 *不错* 忘记](/blog/fire-and-dont-quite-forget-ephemeral-execution) - 理论和模式
- [第2部分:建立一个可重复使用的短执行图书馆](/blog/ephemeral-execution-library) - 充分执行
- [重度( 维基百分数)](https://en.wikipedia.org/wiki/Stigmergy) - 通过环境改变进行间接协调
- [电路断断路器模式Name](https://learn.microsoft.com/en-us/azure/architecture/patterns/circuit-breaker)
- [反反应扩展](https://github.com/dotnet/reactive)