短时间信号 - 将原子变成一个遥感网络

Friday, 12 December 2025

一个微小的原始体将同时进行的工作变成一个协调的适应系统

"时空信号模式"

内 第一部分第一部分 我们建造了短时执行 - 捆绑, 私人, 自我清洁的同步工作流程。 In 第二部分第二部分 我们把它变成一个可再使用的图书馆,有协调员、有钥匙的管道和DI集成。

本条增加了一个改变一切的小特点: 信号信号信号.

举起手来!

这现在也出现在大多为粗略的. 短暂的 Niget 软件包中 20多个大多为lucid.ephears 模式和“原子”.

源源文件

完整源代码在基特Hub 库

信号基础设施存在于:

------	---------
短期行动cs * 运行中的信号排放和撤回
时间选择cs `CancelOnSignals`, `DeferOnSignals`, `OnSignal`, `OnSignalRetracted`)
书签父母用于信号过滤的球式模式匹配
信号发送器cs 使用模式匹配的 Async 信号路由( 支持) `*`, `?`逗号列表, 确定性顺序)
实例/信号HttpClient.cs HTTP电话的精细颗粒信号排放样本
实例/调整性翻译服务cs 适应率受基于信号的推迟限制
实例/基于信号的环环道环环环环环箱.cs 巡回断路器读取时空信号窗口
实例/遥测信号手法cs 使用遥测集成的 Async 信号处理

问题:孤立的原子

我们的时尚协调员善于处理工作,但他们是孤立的。每个协调员都知道自己的行动,但不知道系统其他地方正在发生什么。

// 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:

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 式視窗一生即时衰变自动衰变衰变衰变衰变衰变衰变衰变衰变衰变衰变衰变衰变衰变衰变衰变衰变衰变衰
失败模式 * * 许多 * * 几乎没有 * * * * * 失败模式 * * * 失败模式 * * 失败模式 * * * 许多失败模式 * * * 几乎无成功模式 * * * * * * 失败模式 * * * 失败模式 * * * 失败模式 * * * 许多失败模式 * * * * 几乎无成功模式 * * * * * * * * 失败模式 * * * * * * * 失败模式 * * * * * * * 失败模式 * * * * * * * * * * 失败模式 * * * * * * * * * * * * * * * * * * * * 几乎无失败模式 * * * * * * * * * * * * * * * * * * * * * 失败模式 * * * * * * * * * * * * * 失败模式 * * * * * * * * * * * * * * * * * 失败模式 * * * * * * * * * * *失败模式 * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *

《守则》差异

事件处理方法(经典):

public event Action RateLimited;

try
{
    await CallApiAsync();
}
catch (RateLimitException)
{
    RateLimited?.Invoke(); // Makes someone else act right now
}

问题:

谁的反应?
按什么顺序排列?
如果两个处理者有冲突呢?
如果一个操作员扔球呢?
如果时机不对呢?
如果打电话的人不想这样呢?

信号接近( 短暂) :

try
{
    await CallApiAsync(ct);
}
catch (RateLimitException ex)
{
    op.Signal($"rate-limit:{ex.RetryAfterMs}ms");
    throw;
}

这里什么也没有发生后来完全不一样的地方

if (translator.HasSignal("rate-limit"))
{
    await Task.Delay(1000);   // Act when *we* choose
}

注:

不直接连接
无惊喜电话
没有执行跳跳
没有回回回地狱的地狱
没有交叉线怪异
只有当我们 问问问问 查看

"啊哈!" 线条

事件传输控制信号传输环境

这是整个心理模式在一句话。

真正信号是什么

取决于您的背景 :

听众 * 定义 * |----------|------------| | 系统思考者 * 间接协调的基底基底子 * | 工程师 轻量级元数据附在一个捆绑的滑动窗口中的操作中 | PL/Con 货币呆子 隐含的,时间上的黑板与捆绑的通货货币语义搭配 | 框架用户 处理过程中的自我清洁状态表面你随时可以查询

信号是共享的、紧凑的内存表面留下的痕迹。任何人都可以查看它们。没有人有反应的义务。这是根本的。悬浮协调模式 -- -- 同一种蚂蚁使用,同一种在早期AI中使用的黑板系统,同一种现代CRDT八卦网络的提示。

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

应用程序 Insights / Open 遥测

using var activity = source.StartActivity("ProcessOrder");
activity?.SetTag("order.id", orderId);
activity?.SetTag("rate.limited", true);

最佳:在服务、长期遥测存储、相关身份等各处分布追踪。

使用遥测时使用遥测方法: 您需要跟踪多个服务的请求, 存储用于分析的量度, 或与监测工具整合。

使用时间信号:您需要过程内的环境意识, 反应协调, 或者不需要遥测基础设施。

反应扩展( Rx)

var rateLimits = Observable.FromEventPattern<RateLimitEventArgs>(
    h => api.RateLimitHit += h,
    h => api.RateLimitHit -= h);

rateLimits
    .Throttle(TimeSpan.FromSeconds(1))
    .Subscribe(e => HandleRateLimit(e));

最佳:复杂事件处理、基于时间的业务,结合多个事件流。

使用 Rx 当: 您需要复杂的时间查询( 窗口、拆卸、组合流) 。

使用时间信号:您需要更简单的基于投票的感应、自动清理或与行动跟踪整合。

中继R 通知

public class RateLimitNotification : INotification
{
    public int RetryAfterMs { get; init; }
}

await _mediator.Publish(new RateLimitNotification { RetryAfterMs = 5000 });

最佳:与多个处理器分离处理处理处理过程中的事件。

使用中位tR:您想要多个处理器对同一事件同步反应。

使用时间信号:您想要没有明确订阅、自我清理历史或与捆绑处决结合的环境遥感。

Polly 电路断开器

var circuitBreaker = Policy
    .Handle<HttpRequestException>()
    .CircuitBreakerAsync(5, TimeSpan.FromSeconds(30));

最佳:通过自动国家管理,应对个人电话的复原力。

使用 Polly 时:您需要每次呼叫的复原力,自动进行半开放/封闭的过渡。

使用时间信号:您想要通过许多操作, 定制电路逻辑, 或与操作跟踪整合, 来提高环境意识。

结合他们:在工作体内使用Polly,在电路出行时发出信号。

比较比较表

自我清除 * 环境感应 * 破碎 * 自定义逻辑 * 整合 * |----------|:-------------:|:---------------:|:---------:|:------------:|:-----------:| 外在工具 @ @ openTelestries @ @ @ @ @ @ @ @ _ 反应扩展
《中位数》《手册》《手册》《手册》

| 短讯讯号

升升和减缩信号

业务实施 ISignalEmitter:

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

在你工作的身体里:

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 调用

对于非常详细的可观测性,您可以在操作的每个阶段发出信号。图书馆包括一个样本。信号提示服务证明这一模式:

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 * 下载完成 * * * * * * 下载完成 * * * * * * 下载完成 * * * * * * 下载完成 * * * * * * 下载完成 * * * * * * 下载完成 * * * * * * * 下载完成 * * * * * * * * 下载完成 * * * * * * * * * 下载完成 * * * * * * * * * *

然后,您可以用匹配的图案查询这些图案 :

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

正在压缩信号

操作也可以删除自己的信号。这对临时状态有用 :

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

递减事件

就像信号发射一样,撤回可以触发回击:

var coordinator = new EphemeralWorkCoordinator<Request>(
    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 - 所用模式(如果模式匹配)

真实世界实例:回收率限制

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

遥感信号

所有协调员都提供最佳信号查询:

// 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:

协调员可对信号自动作出反应:

var coordinator = new EphemeralWorkCoordinator<Request>(
    body,
    new EphemeralOptions
    {
        // Cancel new work if these signals are present
        CancelOnSignals = new HashSet<string> { "system-overload", "circuit-open" },

        // Defer new work while these signals are present
        DeferOnSignals = new HashSet<string> { "rate-limit" },
        MaxDeferAttempts = 10,
        DeferCheckInterval = TimeSpan.FromMilliseconds(100)
    });

当信号在 CancelOnSignals 检测到,新项目被跳过(算作失败) 。当信号在 DeferOnSignals 被检测到,新的项目将等待信号清除。

现实世界实例:适应率限制

发自适应性转换服务cs:

public class AdaptiveTranslationService : IAsyncDisposable
{
    private readonly EphemeralWorkCoordinator<TranslationRequest> _coordinator;
    private readonly ITranslationApi _translationApi;

    public AdaptiveTranslationService(ITranslationApi translationApi)
    {
        _translationApi = translationApi;

        _coordinator = new EphemeralWorkCoordinator<TranslationRequest>(
            ProcessTranslationAsync,
            new EphemeralOptions
            {
                MaxConcurrency = 8,
                MaxTrackedOperations = 100,

                // New work is deferred while any "rate-limit" or "rate-limit:*" signal is present
                DeferOnSignals = new HashSet<string> { "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:

public class OrderProcessingSystem
{
    private readonly SignalSink _sharedSignals = new(maxCapacity: 1000);

    private readonly EphemeralWorkCoordinator<Order> _orderProcessor;
    private readonly EphemeralWorkCoordinator<PaymentRequest> _paymentProcessor;

    public OrderProcessingSystem()
    {
        var options = new EphemeralOptions { Signals = _sharedSignals };

        _orderProcessor = new EphemeralWorkCoordinator<Order>(
            ProcessOrderAsync, options);

        _paymentProcessor = new EphemeralWorkCoordinator<PaymentRequest>(
            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);
    }
}

真实世界实例:健康监测

[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:

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<T>(EphemeralWorkCoordinator<T> coordinator)
    {
        var recentFailures = coordinator.GetSignalsSince(
            DateTimeOffset.UtcNow - _windowSize);

        return recentFailures.Count(s => s.Signal == _failureSignal) >= _threshold;
    }

    public int GetFailureCount<T>(EphemeralWorkCoordinator<T> 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:

当信号能引发其他信号时你就会冒着无穷无尽的循环风险 SignalConstraints 防止:

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<string> { "completed", "failed", "resolved" },

        // Signals that emit but don't propagate
        LeafSignals = new HashSet<string> { "logged", "metric" },

        // Callback when a signal is blocked
        OnBlocked = (signal, reason) =>
        {
            _logger.LogWarning("Signal {Signal} blocked: {Reason}",
                signal.Signal, reason);
        }
    }
};

信号传动

音轨因果关系 EmitCaused:

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:

为了让所有协调员都能看到需要显示的信号:

public sealed class SignalSink
{
    private readonly ConcurrentQueue<SignalEvent> _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<SignalEvent> Sense();
    public IReadOnlyList<SignalEvent> Sense(Func<SignalEvent, bool> predicate);
    public bool Detect(string signalName);
    public bool Detect(Func<SignalEvent, bool> predicate);

    public int Count { get; }
}

用法 :

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

与字符串匹配模式

发自书签父母:

用于信号过滤的球式匹配 :

// 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<string>
    {
        "system-overload, circuit-open",  // Either pattern
        "error.*"                          // Any error signal
    }
}

信号命名公约

保持信号简单和一致:

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

发送信号

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<Request>(
    body,
    new EphemeralOptions
    {
        OnSignal = dispatcher.Dispatch
    });

模式支助 *, ?和逗号列表("error.*,timeout"所有匹配的操作员均按登记顺序由背景钥匙协调员负责操作;发放时保持同步。

Async 信号处理器

用于独立自动同步处理:

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

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<string, string>
        {
            ["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();
}

把它发给你的协调员:

await using var telemetryHandler = new TelemetrySignalHandler(telemetryClient);

await using var coordinator = new EphemeralWorkCoordinator<Request>(
    ProcessAsync,
    new EphemeralOptions
    {
        OnSignal = signal => telemetryHandler.OnSignal(signal)
    });

处理器 :

永无区块 操作线 - OnSignal 立即返回
信号分类 不同的遥测类型前缀
度量指标 监测健康(排队、处理、丢弃)
间距内存 - 如果队列填满,会减少最古老的信号

为何这样工作

信号是强大的因为它们是 短时间:

财产津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴津贴 |----------|---------| | 环形无法不受约束地成长 - 旧信号时代熄灭 - | 自我清洁 不需要清理代码 | 分离光荣者不知道听众 | 可观察 任何代码都能感知到目前的状况 | 私营没有用户数据 - 只是信号名 | 快速快速 O(1)探测短路

时间窗口已经存在用于调试。信号只是赋予它语义的含义。

结论结论结论结论结论

信号将孤立的处决原子变成 遥感网络每位协调员可以:

埃米特 有关其经历的信号
感知发自其本身历史或共有汇的信号
反应反应 自动通过 CancelOnSignals 和 DeferOnSignals

没有信息代理,没有共享状态,没有协调协议,只有使用自然衰减的元数据操作。

原子不会直接和对方交谈, 它们只是把痕迹留在窗口里, 其他人可以观察它们。它对于合成系统来说是一丝不苟的。

火信号感知忘记

短时间信号 - 将原子变成一个遥感网络

举起手来!

源源文件

问题:孤立的原子

解决方案:行动信号

时间信号法

法律1 - 信号从不暗中触发执行

第2号法律 - 明确、地方和可选的信号法

第3号法律 -- -- 局部到原子的信号

第4号法律 - 信号是仅附加事实

第5号法律 -- -- 信号表面断裂和时间限制

法律6 - 观察者阅读,他们不写

第7号法律 - 类似事件的行为是一个层,不是原始行为

第8号法律 -- -- 任何情况下不得发生跨原子变异

第9号法律 -- -- 全球观点是衍生的,不会变异

第10号法律 - 清除手持者

最佳分析:脚印,不是指示

此删除是什么

快速比较

《守则》差异

"啊哈!" 线条

真正信号是什么

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

应用程序 Insights / Open 遥测

反应扩展( Rx)

中继R 通知

Polly 电路断开器

比较比较表

升升和减缩信号

精密信号示例: HTTP 调用

正在压缩信号

递减事件

真实世界实例:回收率限制

遥感信号

信号反应处理

现实世界实例:适应率限制

现实世界实例:交叉协调员认识

真实世界实例:健康监测

真实世界实例:基于信号的电路断裂

信号限制:防止无限环圈

信号传动

信号辛克:全球信号空间

与字符串匹配模式

信号命名公约

Async 信号处理

发送信号

Async 信号处理器

遥测信号Handler 示例

为何这样工作

结论 结论 结论 结论 结论

链接链接链接链接

结论结论结论结论结论