内容
# redb.Route
**.NET 企业集成框架。仅需几行 C# 代码即可连接任意系统至任意系统。**
> **大型语言模型 / 编码代理：**整个 redb 生态系统的规范化训练合约 → <https://redbase.app/llms.txt>
redb.Route 是 .NET 中与 [Apache Camel](https://camel.apache.org/) 相似的框架 —— 一个流畅的 C# 领域特定语言（DSL），通过 `From → Process → To` 管道连接 Kafka、RabbitMQ、Redis、SQL、HTTP、gRPC、SFTP、MQTT、S3 等。无需 XML，无消息总线锁定，无每个端点许可费。
[](https://www.nuget.org/packages/redb.Route)
[](LICENSE)
[](https://dotnet.microsoft.com)
> **生产运行时：**路由被打包为 `.tpkg` 模块，并部署到 **[redb.Tsak](https://github.com/redbase-app/redb-tsak)** —— 具有热重载、集群、REST 管理 API、CLI 和 Blazor 仪表盘的运行时容器。redb.Route 处理 _what_（管道逻辑）；redb.Tsak 处理 _where_、_when_ 和 _how many_（部署生命周期）。参见 [在生产环境中运行路由](#running-routes-in-production--redbtsak)。
```csharp
From("kafka://orders?groupId=svc&brokers=localhost:9092")
.Filter(Header("type").isEqualTo("new"))
.Retry(3)
.To("rabbitmq://events?host=localhost");
```
---
## 亮点
- **27 种外部传输 + 5 种内置组件**（Direct、SEDA、Timer、Mock、Log）。其中之一 — `llm:` — 是一个一流的 LLM 连接器，具有通用 OpenAI 兼容提供程序（14 种 API）和本地 Anthropic Messages API 提供程序，以及 `.AsLlmTool()`，可以将其他 26 种传输中的任意一种暴露为 LLM 可调用工具，无需连接器更改。Token-by-token 流式传输在端到端运行：`llm://` 步骤上的 `?stream=true` 通过 `redb.Route.Http`（带有 `event: done` 摘要尾部）或通过 `redb.Route.WebSocket` 作为每个 Token 的一个 `Text` 帧。
- **MCP 客户端连接器 — `mcp:` URI。** `redb.Route.Llm.Mcp` 启动外部 **Model Context Protocol** 服务器（stdio 或 HTTP+SSE），在启动时运行 `initialize` + `tools/list`，并将每个远程工具注册到 `IToolDescriptorRegistry` 作为一流的 `ILlmToolDescriptor`。代理像任何本地工具一样拾取它们 - 完整的审计（`ToolSetHash`）、治理（`Safety`）、可观察性和审批流程自动继承。取消被贯彻到底：中止的代理迭代发出 JSON-RPC `notifications/cancelled`，并且正在进行的 `tools/call` 展开。
- **30+ EIP 模式** - Splitter、Aggregator、CBR、WireTap、Saga、Circuit Breaker、Idempotent Consumer、Claim Check、Throttle、Resequencer、Scatter-Gather 等。
- **编译表达式引擎** - `${header.x}`、`${header.x++}`、算术、JSONPath、XPath 通过 `System.Linq.Expressions` 编译为 `Func<IExchange, T>`。无解释器开销。
- **类型安全流畅构建器** - `Kafka.Topic("orders").GroupId("svc").Acks("All")` 而不是 URI 字符串。
- **内置 OpenTelemetry** - 每个路由和每个步骤的活动和度量，无需额外配置。
- **事务路由** - `.Transacted()` 将管道步骤包装在 `TransactionScope` 中；SQL 传输绑定 ADO.NET 事务每个语句。
- **生产运行时** - [redb.Tsak](#running-routes-in-production--redbtsak) 托管路由程序集，具有热重载、集群、REST API 和 Web UI。
- **Apache 2.0 许可** - 无商业限制，无每个端点定价。
---
## 目录
- [快速入门](#quick-start)
- [为什么选择 redb.Route](#why-redbroute) — [差异化](#what-sets-redbroute-apart) · [与其他 .NET 选项比较](#vs-other-net-options-quick-guide) · [手动代码比较](#manual-code-comparison)
- [用例](#use-cases)
- [配置](#configuration) — appsettings、IOptions、DI 参数化
- [比较](#how-it-compares) — 与 Apache Camel、MassTransit、NServiceBus、Wolverine 比较；[从 Camel 迁移](#migrating-from-apache-camel)
- [路由模式](#routing-patterns) — From、To、Filter、Choice、Split、Recipient List、Dynamic Router
- [错误处理](#error-handling) — [重试](#retry-per-step) · [死信通道](#dead-letter-channel) · [TryCatch](#trycatch-scoped) · [OnException 范围](#onexception--scope-form-chain) · [OnException 配置](#onexception--configure-form-action) · [重新传递选项](#redelivery-configuration) · [综合示例](#combined-example)
- [消息转换](#message-transformation) — SetBody、SetHeader、Marshal、Unmarshal
- [表达式语言](#expression-language) — 帮助器、谓词、字符串模板
- [其他 EIP 模式](#other-eip-patterns) — WireTap、Multicast、Throttle、Circuit Breaker、Saga、Idempotent Consumer、Loop、Delay、Resequencer、Enrich
- [请求-响应（InOut）](#request-response-inout)
- [并发性和并行性](#concurrency--parallelism) — `.Threads(N)` EIP、消费者级旋钮、[完整指南](CONCURRENCY.md)
- [可靠性](#reliability) — RabbitMQ 确认、Kafka EOS、持久性 IdempotentConsumer、通过 Sql 轮询的 Outbox
- [验证](#validation)
- [使用 Mock 进行测试](#testing-with-mock)
- [遥测](#telemetry)
- [日志记录](#logging)
- [流畅传输构建器](#fluent-transport-builders)
- [架构](#architecture)
- [DSL 参考](#dsl-reference)
- [URI 方案参考](#uri-scheme-reference)
- [传输能力矩阵](#transport-capability-matrix)
- [软件包](#packages)
- [在生产环境中运行路由 — redb.Tsak](#running-routes-in-production--redbtsak)
- [贡献](#contributing)
- [许可证](#license)
---
## 快速入门
每个 redb.Route 管道都遵循相同的形状：单个源（`From`）、中间的零个或多个处理器和一个或多个接收器（`To`）。消息作为携带 `Body`、`Headers` 和 `Properties` 的 `IExchange` 实例流动。
```mermaid
flowchart LR
From([From\nkafka://orders]) --> Filter[Filter]
Filter --> Transform[SetBody / Transform]
Transform --> Choice{Choice}
Choice -- high --> ToA([To\ndirect://fast-lane])
Choice -- low --> ToB([To\nseda://batch])
Choice -- else --> ToC([To\ndirect://standard])
```
```csharp
// Program.cs — 最小集成管道
var builder = Host.CreateApplicationBuilder(args);
builder.Services.AddRedbRoute(route =>
{
route.AddRoutes(r =>
{
// Kafka → 过滤器 → RabbitMQ，3 行代码
r.From("kafka://orders?groupId=svc&brokers=localhost:9092")
.Filter(e => e.Message.GetHeader<string>("type") == "new")
.To("rabbitmq://events?host=localhost");
});
});
builder.Services.AddRedbRouteKafka();
builder.Services.AddRedbRouteRabbitMQ();
var app = builder.Build();
app.Run();
```
对于复杂的路由逻辑，将路由分组到一个 `RouteBuilder` 类中：
```csharp
public class OrderRoutes : RouteBuilder
{
protected override void Configure()
{
From("kafka://orders?groupId=svc&brokers=localhost:9092")
.RouteId("order-pipeline")
.Choice()
.When(Header("priority").isEqualTo("high"))
.Log("高优先级订单")
.To("direct://fast-lane")
.When(Header("priority").isEqualTo("low"))
.To("seda://batch-queue")
.Otherwise()
.To("direct://standard")
.EndChoice();
From("direct://fast-lane")
.Retry(3)
.SetHeader("processed-at", e => DateTimeOffset.UtcNow)
.Process(async (exchange, ct) =>
{
var body = exchange.Message.GetBody<string>();
exchange.Message.SetBody($"PROCESSED: {body}");
})
.To("rabbitmq://processed?host=localhost");
}
}
builder.Services.AddRedbRoute(route => route.AddRouteBuilder<OrderRoutes>());
```
---
## 为什么选择 redb.Route
### redb.Route 的优势
您在其他 .NET 集成框架中找不到的四件事：
- **27 种传输开箱即用，包括一个完整的 LLM 连接器。** Kafka、RabbitMQ、Redis、SQL、HTTP、gRPC、SFTP、MQTT、S3、Amazon SQS + SNS、IBM MQ、AMQP 1.0、Azure Service Bus、Elasticsearch、Firebase、LDAP、Mail、TCP、WebSocket、SignalR、Telegram、FTP、Quartz、File、**Exec**（本地进程执行）、 **`llm:`**（OpenAI 兼容的通用提供程序，适用于 14 家供应商 + 本地 Anthropic Messages API）和 **`mcp:`**（Model Context Protocol 客户端 - 整个社区 MCP 生态系统作为可调用工具）- **以及 `.AsLlmTool()`，将其他 26 种传输中的任何一种转换为 LLM 工具，无需连接器更改**。MassTransit 提供了 5 种，Wolverine 提供了 4 种，NServiceBus 提供了 7 种。竞争对手期望您只使用消息代理；redb.Route 将文件、邮箱、FTP 服务器、SQL 轮询、本地进程和 LLM 端点视为一流传输 - 并且其中任何一种都可以成为 LLM 可调用工具。
- **编译表达式引擎。** `${header.x}`、`${header.x++}`、算术、JSONPath、XPath 通过 `System.Linq.Expressions` 在路由构建时编译为 `Func<IExchange, T>`。无解释器，无每消息解析，结果缓存每个路由。Apache Camel 的 Simple Language 被解释；MassTransit / Wolverine / NServiceBus 没有表达式引擎。
- **30+ EIP 模式作为一流的 DSL。** 过滤器、选择、分割器、聚合器、多播、WireTap、接收者列表、动态路由器、重新排序器、分散聚合器、声明检查、幂等消费者、Saga、电路中断器、限流器、重试、死信、循环、延迟、去抖、丰富、超时、TryCatch、事务、处理、验证。 .NET 竞争对手为您提供 Saga + 请求/响应 + 外箱；其他所有功能您都需要自己编写。
- **Apache 2.0 许可，无每个端点定价。** NServiceBus 在 2 个端点后是商业性的。redb.Route 对任何使用、任何规模都无限制。
以及：类型安全的流畅构建器（`Kafka.Topic(...).GroupId(...)` 而不是 URI 字符串），每个路由和每个步骤的内置 OpenTelemetry，通过 `.Transacted()` 进行事务管道，以及具有热重载和集群的运行时环境（[redb.Tsak](#running-routes-in-production--redbtsak)）。
### 与其他 .NET 选项的快速指南
| 如果您需要 | 选择 |
|-------------|------|
| 跨多种传输路由（Kafka + SQL + SFTP + HTTP 在一个管道中） | **redb.Route** |
| EIP 模式（Splitter、Aggregator、Content-Based Router、WireTap） | **redb.Route** |
| 协议桥接（Kafka ↔ RabbitMQ、MQTT ↔ HTTP、FTP ↔ SQL） | **redb.Route** |
| 具有补偿的 Choreography Saga（管道级，无外部状态） | **redb.Route**（`.Saga(s => s.Step(...).Compensate(...))`） |
| 通过 SQL 轮询的 Outbox（`SELECT … WHERE processed=0` → 进程 → `UPDATE`） | **redb.Route**（`Sql.Poll(...)` + `.Transacted()`） |
| 进程内调度到处理程序类（`RedbController` 背后是 `direct:`） | **redb.Route** + `redb.Route.Controllers` |
| 具有 `RouteBuilder` 消费者类的异步进程内发布/订阅 | **redb.Route**（`seda:` + `RouteBuilder`） |
| 跨进程重启持久化编排的 Saga 状态机 | MassTransit / NServiceBus / Wolverine |
| 事务管道（延迟代理提交/确认与路由结果绑定） | **redb.Route**（`.Transacted()` + `ITransactedAction`：Kafka EOS、RabbitMQ tx、IBM MQ、AMQP） |
| 托管事务 Outbox 容器（自动表 + 后台重试守护进程） | MassTransit / NServiceBus / Wolverine |
> redb.Route 和 MassTransit/NServiceBus 解决不同的问题。redb.Route 是一个 **管道和传输集成引擎**（.NET 的 Apache Camel）。MassTransit/NServiceBus 是 **消息总线框架**，以消费者注册和管理 Saga 状态为中心。您可以同时使用两者 - redb.Route 用于跨协议路由和转换，MassTransit 用于持久的长时间运行的工作流。
完整对比表：[比较](#how-it-compares)。
### 手动代码比较
| 问题 | 手动代码 | redb.Route |
|---------|-------------|------------|
| 连接 Kafka → RabbitMQ | 80+ 行：消费者、生产者、序列化、错误处理 | `From("kafka://orders").To("rabbitmq://events")` |
| 基于内容的路由 | 嵌套 `if/switch`，自定义调度表 | `.Choice().When(...).To(...).Otherwise().To(...)` |
| 重试 + 死信 | 自制的重试循环，DLQ 逻辑 | `.Retry(3).DeadLetterChannel("seda://failed")` |
| 文件轮询 → 数据库 | `FileSystemWatcher` + ADO.NET + 线程 | `From("file:///data?include=*.csv").To("sql://...")` |
| 可观察性 | 自定义指标、`Activity` 跨度、日志记录 | 内置 OpenTelemetry - 每个路由/步骤的跟踪和指标 |
| 协议桥接 | 每个协议对的自定义适配器 | 通过 URI 方案任意到任意：`kafka:`、`http:`、`grpc:`、`mqtt:` |
---
## 用例
redb.Route 适用于任何消息在系统之间移动并需要在途中进行转换、路由或缓冲的工作负载：
- **消息代理桥接。** Kafka 主题 → 过滤器 → 丰富 → RabbitMQ 交换。MQTT 传感器流 → 批处理 → SQL 插入。
- **文件摄取管道。** SFTP 文件夹轮询 → 解码 CSV → 验证 → SQL 更新 → 存档到 S3。
- **API 网关和 Webhook。** HTTP 端点（`Http.Listen`）→ 验证 → 分发到内部队列 → 响应。Stripe/GitHub Webhook → SignalR 广播。
- **定时作业。** Quartz Cron → SQL 提取 → 邮件报告。定时器 → 轮询外部 API → 推送到 Kafka。
- **协议转换。** SOAP → JSON，XML → CSV，gRPC → REST，MQTT → AMQP。任意到任意通过编解码 + URI 交换。
- **Saga 编排，无状态机框架。** 保留库存 → 充值付款 → 运送 - 出现故障时有补偿步骤。
- **事件驱动的微服务。** 每个服务的 `RouteBuilder` 类注册路由；`IHostedService` 启动它们；OpenTelemetry 跟踪整个链条。
- **遗留集成。** IBM MQ、FTP、LDAP、POP3/IMAP、原始 TCP - 一流支持，而不是事后才想到的。
- **实时仪表盘。** Kafka 流 → 限流 → SignalR Hub → Blazor/JS 客户端。
如果问题是 *"如何将消息从 A 移动到 B，并在中间应用规则？"*，redb.Route 可以在几行 C# 代码中回答。
---
## 配置
端点、传输和选项与标准的 .NET 配置堆栈一起工作 - `appsettings.json`、`IOptions<T>`、环境变量、机密 - 无需自定义配置层。
### 来自配置的端点 URI
在启动时解析代理 URL、主题名称和凭据，然后在代码中构建端点 URI：
```jsonc
// appsettings.json
{
"Kafka": {
"Brokers": "broker1:9092,broker2:9092",
"GroupId": "order-svc",
"Topic": "orders"
},
"RabbitMQ": {
"Host": "rabbit.local",
"Exchange": "events"
}
}
```
```csharp
builder.Services.AddRedbRoute(route =>
{
route.AddRoutes(r =>
{
var cfg = builder.Configuration;
var kafka = Kafka.Topic(cfg["Kafka:Topic"]!)
.Brokers(cfg["Kafka:Brokers"]!)
.GroupId(cfg["Kafka:GroupId"]!);
var rabbit = Rabbit.Queue("events")
.Host(cfg["RabbitMQ:Host"]!)
.Exchange(cfg["RabbitMQ:Exchange"]!, "topic");
r.From(kafka).To(rabbit);
});
});
```
### 将选项绑定到 RouteBuilder
类型化的 `IOptions<T>` 通过 DI 像其他服务一样流动。将其注入 `RouteBuilder` 构造函数，并在 `Configure()` 中使用这些值：
```csharp
public class KafkaOptions
{
public string Brokers { get; set; } = "";
public string GroupId { get; set; } = "";
public string Topic { get; set; } = "";
}
public class OrderRoutes : RouteBuilder
{
private readonly KafkaOptions _kafka;
public OrderRoutes(IOptions<KafkaOptions> kafka) => _kafka = kafka.Value;
protected override void Configure()
{
From(Kafka.Topic(_kafka.Topic).Brokers(_kafka.Brokers).GroupId(_kafka.GroupId))
.RouteId("order-pipeline")
.To("seda://orders-internal");
}
}
// Program.cs
builder.Services.Configure<KafkaOptions>(builder.Configuration.GetSection("Kafka"));
builder.Services.AddRedbRoute(route => route.AddRouteBuilder<OrderRoutes>());
```
### 命名 SQL 数据源
SQL 传输使用 **命名** 数据源注册在 DI 中，因此连接字符串保留在配置中，而不在路由代码中：
```csharp
builder.Services.AddRedbRouteSql(sql =>
{
sql.AddDataSource("main", builder.Configuration.GetConnectionString("Main")!, "SqlServer");
sql.AddDataSource("audit", builder.Configuration.GetConnectionString("Audit")!, "PostgreSql");
});
// 路由仅按名称引用数据源 - 凭据从不出现在 URI 中
From(Sql.Poll("SELECT * FROM Orders WHERE Status = 'New'")
.DataSource("main")
.Delay(5000))
.To(Sql.Execute("INSERT INTO AuditLog (Body) VALUES (@body)")
.DataSource("audit")
.Transacted());
```
### 引擎选项
```csharp
builder.Services.Configure<RouteEngineOptions>(o =>
{
o.EnableTelemetry = true; // OpenTelemetry 活动
o.EnableMetrics = true; // OpenTelemetry 度量
o.DefaultRetryAttempts = 3;
o.DefaultErrorMode = ErrorMode.DeadLetter;
});
```
---
## 比较
... (rest of the document remains the same)
### 概要
| | Apache Camel | MassTransit | NServiceBus | Wolverine | **redb.Route** |
|--|--|--|--|--|--|
| 平台 | JVM | .NET | .NET | .NET | **.NET 8/9/10** |
| 许可证 | Apache 2.0 | Apache 2.0 | **商业** | MIT | **Apache 2.0** |
| 关注点 | EIP路由 | 发布/订阅 + 消息总线 | 消息总线 | 中介 + 消息总线 | **企业服务总线 (.NET)** |
| 传输 | 300+ | 5 | 7 | 4 | **23 + 5 内置 + LLM** |
| EIP模式 | 80+ | Saga, 请求/响应, 事务框 | Saga, 请求/响应 | Saga, 请求/响应, 事务框 | **30+** |
| 表达式引擎 | 简单语言（解释执行） | 无 | 无 | 无 | **编译（Linq.Expressions）** |
| 配置 | XML 或 Java DSL | C# 流畅接口 | C# 流畅接口 | C# 流畅接口 | **仅 C# 流畅接口** |
| Saga持久化 | 是 | **是（基于数据库）** | **是（基于数据库）** | **是（Marten）** | 补偿步骤 + 持久`IdempotentConsumer`（redb RTTI存储/SQL）用于去重；无内置状态机 |
| 事务框 | 插件 | **是** | **是** | **是** | 通过`Sql.Poll(...).OnSuccess(...).Transacted()`轮询事务框 + 持久`IdempotentConsumer` |
| 发布者确认/EOS | 是 | 是 | 是 | 是 | **是** — RabbitMQ发布者确认（`MaxOutstandingConfirms`），Kafka事务生产者 + `Acks=All` |
### 与 Apache Camel 比较
如果你熟悉 Camel，你就会了解 redb.Route —— 相同的 EIP模式，相同的`from → process → to`模型，相同的 URI 端点方案。
| 概念 | Apache Camel (Java) | redb.Route (.NET) |
|---------|--------------------|-----------------|
| 路由定义 | `from().process().to()` | `From().Process().To()` |
| 路由类 | `RouteBuilder.configure()` | `RouteBuilder.Configure()` |
| 进程内同步 | `direct:` | `direct:` |
| 内存异步队列 | `seda:` | `seda:` |
| 周期性触发 | `timer:` | `timer:` |
| 消息对象 | `Exchange` + `Message` | `IExchange` + `IMessage` |
| 表达式语言 | 简单语言（解释执行） | 字符串表达式（编译执行） |
| EIP模式 | 80+ | 30+（过滤器，选择器，分裂器，聚合器，多播，WireTap，收件人列表，动态路由器，重新排序器，散播-收集，声明检查，幂等消费者，Saga，电路中断器，限流器，重试，死信，循环，延迟，消抖，丰富，超时，TryCatch，事务，…） |
| 组件 | 300+ | 27种传输 + 5种内置 |
| 配置 | XML DSL 或 Java DSL | 仅 C# 流畅接口 |
| 运行时/编排 | Camel K / JBoss Fuse | [redb.Tsak](#在生产环境中运行路由——redbtsak) |
| 平台 | JVM | .NET 8 / 9 / 10 |
redb.Route 与 Camel的不同之处：
- **无 XML。** C# 流畅接口，具有完整的 IntelliSense 支持。
- **类型安全构建器。** `Kafka.Topic("orders").GroupId("svc")` 而不是 URI 字符串。
- **原生 .NET 栈。** `IHostedService`，`IServiceCollection`，`ILogger`，`CancellationToken` 无处不在。
- **内置 OpenTelemetry。** 每个路由和每个步骤的跟踪和指标。
- **编译表达式。** `${header.x++}`，算术，JSONPath，XPath 编译为 `Func<IExchange, T>` —— 不是解释执行的。
- **事务路由。** `.Transacted()` 通过 `TransactionScope`。
- **.NET 特定传输。** SignalR，ASP.NET Kestrel 服务器端消费者。
#### 从 Apache Camel 迁移
Camel 路由几乎可以逐行转换为 redb.Route。方法名称使用 PascalCase，谓词是类型的，但结构是相同的。
| Apache Camel (Java) | redb.Route (C#) |
|---------------------|-----------------|
| `from("kafka:orders?brokers=...")` | `From("kafka://orders?brokers=...")` |
| `.routeId("order-pipeline")` | `.RouteId("order-pipeline")` |
| `.choice()` | `.Choice()` |
| `.when(header("type").isEqualTo("new"))` | `.When(Header("type").isEqualTo("new"))` |
| `.otherwise()` | `.Otherwise()` |
| `.end()` | `.EndChoice()` |
| `.split(body())` | `.Split(Body())` |
| `.aggregate(header("orderId"), strategy)` | `.Aggregate(Header("orderId"), strategy)` |
| `.wireTap("seda:audit")` | `.WireTap("seda://audit")` |
| `.recipientList(header("dest"))` | `.RecipientList(e => ParseDest(e))` |
| `.errorHandler(deadLetterChannel("seda:dlq"))` | `.DeadLetterChannel("seda://dlq")` |
| `.onException(IOException.class).maximumRedeliveries(3)` | `OnException<IOException>().MaximumRedeliveries(3)` |
| `.transacted()` | `.Transacted()` |
| `.setBody(simple("${header.x}"))` | `.SetBody(Expr("${header.x}"))` |
| `.log("got ${body}")` | `.Log("got ${body}")` |
| `.to("rabbitmq://events")` | `.To("rabbitmq://events")` |
一个200行的Camel `RouteBuilder.configure()`通常变成一个200行的C# `RouteBuilder.Configure()`，形状相同 —— 只有宿主（Spring Boot vs `IHost`）和构建系统发生变化。
### 与 MassTransit 比较
| 方面 | MassTransit | redb.Route |
|--------|-------------|------------|
| 焦点 | 可靠的消息总线 + Saga | 企业服务总线 —— 27种传输，30+ EIP（企业集成模式），事务管道 |
| 路由模型 | 消费者类 | 路由管道（`From → Process → To`） |
| EIP模式 | Saga，请求/响应，事务框 | 30+ EIP模式 |
| 传输 | 5（RabbitMQ，Kafka，SQS，Azure SB，gRPC） | 22（+ 文件，SFTP，SQL，MQTT，TCP，HTTP，FTP，LDAP，IBM MQ，S3，…） |
| Saga | 状态机 + **持久状态**（基于数据库） | 补偿步骤（正向 + 反向）与持久去重通过`IdempotentConsumer`（redb RTTI存储/SQL后端，两阶段提交）—— 无内置状态机 |
| 事务框 | 是（事务事务框模式） | 通过`Sql.Poll(...).OnSuccess(...).Transacted()`轮询事务框 + 持久`IdempotentConsumer`（事实上的事务框） |
| 发布者可靠性 | 是（RabbitMQ确认，Kafka EOS） | **是** —— RabbitMQ发布者确认，强制标志，事务通道，自动恢复；Kafka事务生产者 + `Acks=All` + `IsolationLevel=ReadCommitted` |
| 表达式 | 无 | 编译引擎：9种类型，17种谓词，`${header.x++}`，算术 |
| 事务路由 | 无 | 是 —— `.Transacted()` 通过 `TransactionScope` |
| 许可证 | Apache 2.0 | Apache 2.0 |
在具有DB持久性的**状态机Saga**和每个消费者具有内置事务框表时使用MassTransit。在企业集成中，使用redb.Route进行端到端：30+ EIP模式，27种传输，事务管道（`ITransactedAction`），协议桥接，文件处理和多传输管道 —— 可靠性基元（发布者确认，EOS，持久去重，轮询事务框）由DSL构建块组成。
### 与 NServiceBus 比较
| 方面 | NServiceBus | redb.Route |
|--------|-------------|------------|
| 焦点 | 消息总线（可靠交付 + Saga） | 企业服务总线 —— 27种传输，30+ EIP（企业集成模式），事务管道 |
| 路由模型 | 消息处理类 | 路由管道 |
| EIP模式 | Saga（持久），请求/响应 | 30+ EIP模式 |
| 传输 | 7（RabbitMQ，Kafka，SQS，Azure SB，SQL，MSMQ，Learning） | 22（+ 文件，SFTP，MQTT，TCP，HTTP，gRPC，FTP，LDAP，IBM MQ，S3，…） |
| 表达式 | 无 | 编译表达式引擎 |
| 事务路由 | 无 | 是 —— `.Transacted()` 通过 `TransactionScope` |
| 监控 | ServicePulse / ServiceInsight（商业） | OpenTelemetry + redb.Tsak Web UI |
| 许可证 | **商业**（免费至多2个端点/应用） | Apache 2.0 |
NServiceBus成熟且专注于保证交付和持久Saga。redb.Route具有Apache-2.0许可，并涵盖企业集成端到端：30+ EIP模式，27种传输，事务管道（`ITransactedAction`）和完整运行时（redb.Tsak），具有集群，热重载和OTel。
### 与 Wolverine 比较
| 方面 | Wolverine | redb.Route |
|--------|-----------|------------|
| 焦点 | 中介 + 消息总线 + Saga | 企业服务总线 —— 27种传输，30+ EIP（企业集成模式），事务管道 |
| 路由模型 | 处理程序发现 + 中间件 | 路由管道（`From → Process → To`） |
| EIP模式 | Saga，事务框，请求/响应 | 30+ EIP模式 |
| 传输 | 4（RabbitMQ，Kafka，Azure SB，SQS） | 22（+ 文件，SFTP，SQL，MQTT，HTTP，gRPC，FTP，LDAP，IBM MQ，S3，…） |
| Saga存储 | Marten / EF Core | 内存补偿步骤 |
| 表达式 | 无 | 编译表达式引擎 |
| 事务路由 | 事务框为基础 | `.Transacted()` 通过 `TransactionScope` |
| 许可证 | MIT | Apache 2.0 |
Wolverine在具有Marten支持的Saga的处理器样式消息传递方面表现出色。redb.Route涵盖企业集成端到端：30+ EIP模式，27种传输，进程内中介（`RedbController`），发布/订阅（SEDA），事务管道和完整生产运行时（redb.Tsak）。
---
## 路由模式
```csharp
// 基于内容的路由器
.Choice()
.When(Header("type").isEqualTo("order")).To("direct://orders")
.When(Header("type").isEqualTo("payment")).To("direct://payments")
.Otherwise().To("seda://unknown")
.EndChoice()
// 收件人列表 — 发送到多个动态目的地
.RecipientList(e => new[] { "direct://audit", "seda://analytics" })
// 动态路由器 — 在运行时计算目的地
.DynamicRouter(e => ComputeNextEndpoint(e))
// 分割器 — 子路由样式
.Split(e => SplitOrderIntoItems(e),
sub => sub.Process(async (e, ct) => await ProcessItem(e, ct)))
// 分割器 — 范围样式
.Split(e => SplitOrderIntoItems(e))
.Process(async (e, ct) => await ProcessItem(e, ct))
.EndSplit()
```
`When(...)` 接受三种形式 —— lambda，`IPredicate` 或字符串表达式：
```csharp
.When(e => e.Message.GetHeader<string>("type") == "order") // lambda
.When(Header("type").isEqualTo("order")) // IPredicate
.When("header.type == 'order'") // 字符串表达式
```
---
## 错误处理
redb.Route具有四层错误处理，设计用于组合：每步骤重试，范围`DoTry/DoCatch`，路由本地`OnException`和全局`OnException`在`RouteBuilder`中声明。`DeadLetterChannel`可以将失败的消息路由到任何URI，当重试次数耗尽时。
### 重试（每步骤）
单个处理步骤周围的本地重试。在外部调用可能暂时失败时很有用。
```csharp
.Retry(maxRetries: 5, initialDelay: TimeSpan.FromSeconds(1))
.To("http://flaky-service/submit")
```
### 死信通道
在重试次数耗尽后，将失败的交换路由到接收端点。异常保留在交换上（`exchange.GetException()`）并且可以从死信子路由访问。
```csharp
// 路由级别：任何未处理的失败都进入seda://dlq
From("kafka://orders")
.DeadLetterChannel("seda://dlq")
.Process(...)
.To("sql://orders");
// DLQ路由检查异常并决定下一步
From("seda://dlq")
.Log("DLQ: ${header.JMSCorrelationID} — ${exception.message}")
.Choice()
.When(e => e.GetException() is TimeoutException).To("seda://retry-later")
.Otherwise().To("sftp://archive/failed/")
.EndChoice();
```
### TryCatch（范围）
路由内的`try/catch/finally`块。选择此选项时，只有管道的特定部分需要特殊处理。
```csharp
.DoTry()
.To("http://external-api/submit")
.Process(async (e, ct) => await PostProcess(e, ct))
.DoCatch<HttpRequestException>()
.Log("HTTP失败：${exception.message}")
.To("seda://retry-queue")
.DoCatch<TimeoutException>()
.Log("超时，存档")
.To("sftp://archive/timeouts/")
.DoFinally()
.Log("尝试完成（成功或失败）")
.End()
```
### OnException — 范围形式（链式）
为特定异常类型定义处理程序，并具有完整的重新传递配置。有两种放置选项：
**在路由内** — 处理程序仅适用于单个路由：
```csharp
From("kafka://orders")
.OnException<TimeoutException>()
.MaximumRedeliveries(3)
.RedeliveryDelay(TimeSpan.FromSeconds(1))
.UseExponentialBackOff()
.BackOffMultiplier(2.0)
.Handled() // 标记异常已处理，交换继续
.To("seda://timeout-handler")
.EndOnException()
.Process(...)
.To("rabbitmq://processed");
```
**在RouteBuilder级别** — 全局，适用于构建器中的每个路由：
```csharp
public class OrderRoutes : RouteBuilder
{
protected override void Configure()
{
// 全局处理程序 — 适用于以下所有From(...)路由
OnException<HttpRequestException>()
.MaximumRedeliveries(5)
.RedeliveryDelay(TimeSpan.FromSeconds(1))
.UseExponentialBackOff()
.BackOffMultiplier(2.0)
.RetryAttemptedLogLevel(LogLevel.Warning)
.OnRedelivery(e => Metrics.Inc("redelivery", 1))
.Handled()
.To("seda://transient-failures")
.EndOnException();
OnException<DbException>()
.MaximumRedeliveries(2)
.UseOriginalMessage() // 在发送到处理程序之前恢复原始正文
.To("seda://db-failures")
.EndOnException();
// 一个块中的多个异常类型
OnException(typeof(TimeoutException), typeof(SocketException))
.MaximumRedeliveries(3)
.To("seda://network-failures")
.EndOnException();
From("kafka://orders")
.To("http://payments-svc/charge");
From("kafka://shipments")
.To("http://logistics-svc/dispatch");
}
}
```
### OnException — 配置表单（操作）
用于一个块中的多个处理程序。每个`Handle<T>`都接受一个子路由和每个异常的重新传递选项。
```csharp
.OnException(cfg => cfg
.Handle<TimeoutException>(
handler: h => h.Log("超时，重试").To("seda://retry"),
maxRedeliveries: 3,
redeliveryDelay: TimeSpan.FromSeconds(2),
useExponentialBackoff: true)
.Handle<HttpRequestException>(
handler: h => h.To("seda://http-fail"),
maxRedeliveries: 5)
.Handle<DbException>(
handler: h => h.To("seda://db-fail")));
```
### 重传配置
以下所有修改都可以在`OnException(...).EndOnException()`范围内有效：
| 方法 | 目的 |
|--------|---------|
| `MaximumRedeliveries(n)` | 重试尝试次数上限，处理程序触发前 |
| `RedeliveryDelay(TimeSpan)` | 尝试之间基本延迟 |
| `BackOffMultiplier(d)` | 乘数应用于尝试之间的延迟 |
| `UseExponentialBackOff()` | 从固定后退切换到指数后退 |
| `Handled()` / `Handled(false)` | 标记异常已处理（交换继续正常） |
| `Continued()` | 在处理程序后继续路由原始交换 |
| `OnWhen(predicate)` | 仅当谓词匹配时触发处理程序 |
| `RetryWhile(predicate)` | 仅当谓词为真时继续重试 |
| `UseOriginalMessage()` | 恢复原始消息（处理前）以便处理程序使用 |
| `UseOriginalBody()` | 仅恢复原始正文 |
| `OnRedelivery(action)` | 每次重试尝试前调用回调 |
| `OnPrepareFailure(action)` | 调用回调前准备失败 |
| `OnExceptionOccurred(action)` | 异常被捕获的那一刻调用回调 |
| `RetryAttemptedLogLevel(level)` | 重试尝试消息的日志级别 |
| `RetriesExhaustedLogLevel(level)` | 重试次数耗尽消息的日志级别 |
| `LogStackTrace(bool)` | 是否记录完整堆栈跟踪 |
| `LogExhausted(bool)` | 重试次数耗尽时是否记录 |
| `RedeliveryPolicy(policy)` | 应用预先构建的`RedeliveryPolicy`实例 |
### 组合示例
全局策略 + 路由本地覆盖 + DLC + 每步重试，全部组合：
```csharp
public class PaymentRoutes : RouteBuilder
{
protected override void Configure()
{
// 1. 全局策略 — 5次瞬时HTTP错误重试，指数退避
OnException<HttpRequestException>()
.MaximumRedeliveries(5)
.RedeliveryDelay(TimeSpan.FromSeconds(1))
.UseExponentialBackOff()
.BackOffMultiplier(2.0)
.RetryAttemptedLogLevel(LogLevel.Warning)
.OnRedelivery(e => Metrics.Inc("redelivery", 1))
.Handled()
.To("seda://transient-failures")
.EndOnException();
From("kafka://payments?groupId=svc")
.RouteId("payment-pipeline")
// 2. 路由级别DLC — 任何超出处理程序的错误都进入此处
.DeadLetterChannel("seda://payments-dlq")
// 3. 路由本地覆盖 — 此路由特定的超时策略
.OnException<TimeoutException>()
.MaximumRedeliveries(2)
.UseOriginalMessage()
.To("seda://payment-timeouts")
.EndOnException()
.Validate(e => e.Message.GetBody<Payment>().Amount > 0)
.Process(async (e, ct) => await EnrichWithCustomer(e, ct))
// 4. 每步重试 — 无策略，仅需重试
.Retry(maxRetries: 3, initialDelay: TimeSpan.FromMilliseconds(500))
.To("http://payments-gateway/charge");
From("seda://payments-dlq")
.Log("支付DLQ — ${exception.message}")
.To("sftp://archive/payments/failed/");
}
}
```
解析顺序（存在多个匹配项时）：路由本地`OnException`→ RouteBuilder全局`OnException`→ `DeadLetterChannel`→ 未处理（记录并传播）。每个级别中异常类型获胜。
## 消息转换
```csharp
// 设置正文 - 静态、lambda 或表达式
.SetBody("静态值")
.SetBody(e => 转换有效载荷(e.Message.GetBody<Order>()))
.SetBody(JPath("$.order.items"))
.SetBody(Expr("${header.orderId}-${body}"))
// 转换正文
.Transform(e => JsonSerializer.Serialize(e.Message.GetBody<object>()))
.Transform(JPath<string>("$.order.summary"))
// 头部
.SetHeader("来源", "订单服务")
.SetHeader("来源", e => e.Message.GetHeader<string>("原始来源") ?? "未知")
.SetHeader("总计", JPath<decimal>("$.order.total"))
.RemoveHeader("内部跟踪ID")
// 交换属性
.SetProperty("重试次数", 0)
.SetProperty("开始时间", e => DateTimeOffset.UtcNow)
.RemoveProperty("临时标志")
// 序列化/反序列化
.Marshal(typeof(JsonMessageSerializer))
.Unmarshal(typeof(XmlMessageSerializer), typeof(Order))
```
---
## 表达式语言
redb.Route具有编译的表达式引擎。模板和谓词编译为`Func<IExchange, T>`，通过`System.Linq.Expressions` - 没有解释器开销，结果缓存在每个路由中。
### 帮助程序
所有帮助程序都是`RouteBuilder`上的`protected static`方法 - 直接在`Configure()`中使用：
```csharp
// 消息访问器
Body() // 读取消息正文
Header("名称") // 读取头部值
Property("键") // 读取交换属性
Constant(42) // 常数值
Exchange(e => e.RouteId) // 委托 IExchange
// 结构化数据
JPath("$.order.total") // JSONPath 表达式
JPath<decimal>("$.order.total") // 类型化的 JSONPath
XPath("/order/status/text()") // XPath 表达式
XPath<string>("/order/status") // 类型化的 XPath
// 字符串模板（StringExpression - 将 ${...} 桥接到 IExpression）
Expr("${header.orderId}") // 单个头部插值
Expr("${header.prefix}-${body}") // 多值模板
Expr("header.amount") // 原始值表达式（无 ${}）
```
### 谓词
每个表达式都支持17种谓词方法，返回`IPredicate`，用于`Filter()`、`When()`、`Validate()`：
```csharp
// 比较
Header("状态").isEqualTo("活动")
Header("状态").isNotEqualTo("已取消")
Header("金额").isGreaterThan(1000)
Header("金额").isLessThan(500)
Header("金额").isGreaterThanOrEqualTo(100)
Header("金额").isLessThanOrEqualTo(9999)
Header("金额").isBetween(100, 5000)
// 字符串
Header("名称").contains("Corp")
Header("名称").startsWith("Order-")
Header("名称").endsWith(".pdf")
Header("电子邮件").regex(@"^[\w.]+@[\w.]+\.\w+$")
// 集合
Header("类型").In("订单", "支付", "退款")
// 空值
Header("可选").isNull()
Header("必填").isNotNull()
```
### 字符串表达式
`Expr()`创建一个`StringExpression`，自动检测`${...}`模板与原始值表达式。除了简单的插值，引擎支持算术、增量/减量和比较运算符，这些运算符编译为表达式树：
```csharp
// 模板模式 - 在运行时插值 ${...} 占位符
.SetBody(Expr("${header.orderId}-${body}"))
.SetHeader("跟踪", Expr("${header.source}-${header.correlationId}"))
// 增量/减量（修改底层头部/属性）
.SetHeader("尝试次数", Expr("${header.attempt++}")) // 后增量，返回旧值
.SetHeader("尝试次数", Expr("${++header.attempt}")) // 前增量，返回新值
// 算术
.SetHeader("总计", Expr("${header.qty * header.price}"))
.SetHeader("净值", Expr("${header.gross - header.tax}"))
// 值模式 - 评估为表达式，结果为对象
.Filter(Expr("header.amount").isGreaterThan(1000))
.SetBody(Expr("header.total"))
// 字符串表达式在 Filter/When 中（简写，无需 Expr）
.Filter("header.priority == '高'")
.When("header.amount > 1000")
```
---
## 其他 EIP 模式
```csharp
// Wire Tap - 发送副本到审计服务，无需影响主流程
.WireTap("seda://audit")
// 多播 - 并行发送到多个终点
.Multicast("direct://service-a", "direct://service-b")
// 限流 - 速率限制
.Throttle(100) // 最大 100 消息/秒
.Throttle(500, TimeSpan.FromMinutes(1)) // 每分钟最大 500
// 熔断器
.CircuitBreaker(cb => cb
.Threshold(5)
.ResetTimeout(TimeSpan.FromSeconds(30)))
// 幂等消费者 - 去重（作为具有持久后端的 outbox-style 去重）
.IdempotentConsumer(e => e.Message.GetHeader<string>("messageId"), repository)
// InMemoryIdempotentRepository - 进程本地，默认
// SqlIdempotentRepository - ADO.NET，两阶段提交，生存重启
// RedbIdempotentRepository - redb.Core RTTI 存储，两阶段提交，集群范围
// 循环 - 固定次数或条件
.Loop(3, sub => sub.Process(e => Retry(e)))
.Loop(e => !e.Message.GetHeader<bool>("完成"), sub => sub.Process(e => DoWork(e)))
// 延迟
.Delay(TimeSpan.FromSeconds(5))
.Delay(e => TimeSpan.FromSeconds(e.Message.GetHeader<int>("回退")))
// 重新排序 - 重新排序消息
.Resequence(e => e.Message.GetHeader<long>("序列号"))
// 丰富 - 内容丰富
.Enrich("http://api/details", (original, enrichment) => Merge(original, enrichment))
// Saga - 补偿步骤
.Saga(s => s
.Step(
action: async (e, ct) => await ReserveInventory(e, ct),
compensate: async (e, ct) => await ReleaseInventory(e, ct))
.Step(
action: async (e, ct) => await ChargePayment(e, ct),
compensate: async (e, ct) => await RefundPayment(e, ct))
.OnCompletion(async (e, ct) => await PublishOrderCompleted(e, ct)))
// 事务 - 将管道步骤包装在 TransactionScope 中
.Transacted()
// 进程 - 内联异步，带有 CancellationToken
.Process(async (exchange, ct) =>
{
var order = exchange.Message.GetBody<Order>();
var result = await ProcessOrder(order, ct);
exchange.Message.SetBody(result);
})
```
---
## 请求-响应 (InOut)
对于通过 HTTP/gRPC/WebSocket 的同步请求-响应，请将侦听器标记为 `InOut()`，并使用 `.Respond()` 完成路由：
```csharp
public class OrderApi : RouteBuilder
{
protected override void Configure()
{
From(Http.Listen("/api/orders").Port(8080).InOut())
.Unmarshal(typeof(JsonMessageSerializer), typeof(CreateOrderRequest))
.Process(async (e, ct) =>
{
var req = e.Message.GetBody<CreateOrderRequest>();
var resp = await orderService.CreateAsync(req, ct);
e.Message.SetBody(resp);
})
.Marshal(typeof(JsonMessageSerializer))
.Respond(); // 将正文返回给 HTTP 客户端
}
}
```
相同的模式适用于 `Grpc.Listen(...)` 和 `Ws.Listen(...)`。
---
## 并发性和并行性
redb.Route 将并发性分为两个轴：**消费者级别**（源拉取多少消息 - 像 `.ConcurrentConsumers(N)` / `.MaxConcurrentCalls(N)` / Kafka 分区这样的传输选项）和**处理级别**（管道一次处理多少个 - 通过 **`Threads(N)`** EIP）。
```csharp
// 串行轮询源（文件/SQL/S3/IMAP/...）→ 将处理扩展到 10 个线程池：
From("file://in").Threads(10).Process(HeavyTransform).To("http://api").End();
// 代理源 → 使用其本机竞争消费者旋钮（保持交付/确认语义）：
From(Mqtt.Subscribe("telemetry/#").Qos(1).ConcurrentConsumers(5)).Process(Ingest).End();
```
`.Threads(N)` 是一个异步边界（像 `.To("seda://")` 一样）：带回压的有界队列，每个交换的 DI 作用域，停止时优雅地排干，错误路由到 `OnException` - 当 `N > 1` 时不保留顺序。完整的指南，每个传输的表格和 Camel 映射： **[CONCURRENCY.md](CONCURRENCY.md)**。
---
## 可靠性
redb.Route 提供了四种用于至少一次和恰好一次交付的基元。它们可以组合：启用生产者的确认，消费者上的持久化去重，在服务中轮询 outbox - 您具有与框架捆绑在一起的 `UseOutbox()` 相同的保证，只是拼写出来。
### RabbitMQ 发布者确认 + 延迟消费者确认
RabbitMQ 生产者默认使用发布者确认，并带有代理端的流控：
```csharp
.To(Rabbit.Queue("orders")
.Host("rabbit.local")
.MaxOutstandingConfirms(2048) // 待处理发布受 ThrottlingRateLimiter 限制
.Mandatory(true) // 不可路由的消息作为异常出现（BasicReturn）
.Transacted(false) // 或使用 AMQP 事务代替确认
.AutomaticRecovery(true) // 重新连接 + 在代理重启时重新声明拓扑
.TopologyRecoveryEnabled(true))
```
每个 `BasicPublishAsync` 调用都会等待代理确认；当达到 `MaxOutstandingConfirms` 时，生产者背压启动。`Mandatory(true)` 将 `BasicReturn` 事件作为交换异常出现 - 路由失败会大声失败。
### 延迟确认 - `ITransactedAction`（RabbitMQ，AMQP，IBM MQ，Kafka）
延迟确认模式在所有四种代理上统一实现。当 `.Transacted()` 激活时，消费者确认和生产者发送都不立即触发。两者都注册为交换上的 `ITransactedAction` 实例（`exchange.Properties["TRANSACT_ACTION"]`)，并在 `TransactionScope` 关闭时一起执行。
| 传输 | 提交 | 回滚 |
|-----------|--------|----------|
| **RabbitMQ** 消费者 | `TxCommit` + `BasicAck(deliveryTag)` | `TxRollback` + `BasicNack(requeue=true)` |
| **RabbitMQ** 生产者 | `BasicPublishAsync` 触发 | 消息丢弃 |
| **AMQP 1.0** 消费者 | `receiver.Accept(msg)` | `receiver.Release(msg)` |
| **AMQP 1.0** 生产者 | `sender.Send(msg)` 触发 | 消息丢弃 |
| **IBM MQ** 消费者 | `QueueManager.Commit()`（MQCMIT） | `QueueManager.Backout()`（MQBACK） |
| **IBM MQ** 生产者 | MQPUT + MQCMIT | 消息丢弃 |
| **Kafka** 消费者 | `consumer.Commit(result)` - 提交偏移量 | 偏移量未提交（消息将重新传递） |
| **Kafka** 生产者 | `ProduceAsync` 触发（如果启用 EOS） | 消息丢弃 |
所有操作适用于同一个交换，共享同一个 `ConcurrentDictionary<string, ITransactedAction>` - 消费者确认和生产者发送原子提交或回滚。IBM MQ 还跟踪 `BackoutCount`；如果达到 `BackoutThreshold`，消息会移动到一个 backout 队列，而不是无限期回滚到主队列（毒消息处理）。
```csharp
// 与代理无关的.Transacted() 语法
From(Rabbit.Queue("orders").Transacted(true))
.Transacted()
.Process(async (e, ct) => await SaveToDb(e, ct))
.To(Rabbit.Queue("processed").Transacted(true));
// 提交：BasicAck(orders) + BasicPublish(processed) - 原子
// 回滚：BasicNack(orders, requeue=true) + 发送丢弃
From("wmq://ORDERS.IN?transacted=true")
.Transacted()
.Process(async (e, ct) => await SaveToDb(e, ct))
.To("wmq://ORDERS.OUT?transacted=true");
// 提交：MQCMIT on both consumer and producer
// 回滚：MQBACK - 消息返回 ORDERS.IN
From(Kafka.Topic("payments").GroupId("svc"))
.Transacted()
.Process(async (e, ct) => await SaveToDb(e, ct))
.To(Kafka.Topic("processed").EnableTransactionalProducer(true).Acks("All"));
// 提交：偏移量提交 + 内部 Kafka 事务
// 回滚：偏移量未提交 - 消息在下一次轮询时重新传递
```
### Kafka 恰好一次语义（EOS）
```csharp
.To(Kafka.Topic("orders")
.Brokers("broker1:9092,broker2:9092")
.Acks("All") // 等待所有同步副本
.EnableTransactionalProducer(true) // transactional.id 每个生产者分配
.TransactionIdPrefix("order-svc")
.EnableIdempotence(true))
```
消费者端配对使用 `IsolationLevel=ReadCommitted` 和手动提交以仅读取已提交的消息。传输支持完整的 Confluent.Kafka 事务 API。
### 持久化幂等消费者（实际上是去重）
`IdempotentConsumer` 按键拒绝重复消息。与持久化后端一起，它在重启和崩溃中幸存 - 与 MassTransit / NServiceBus 扮演的 outbox `processed_messages` 表相同的角色。
今天有三个后端可用：
| 后端 | 存储 | 两阶段提交 |
|---------|---------|:--:|
| `InMemoryIdempotentRepository` | 进程内存 | - |
| `SqlIdempotentRepository` | ADO.NET（任何提供者）- DDL 随包提供 | `Confirmed` 列 |
| `RedbIdempotentRepository` | redb.Core RTTI 存储（通过 redb 后端集群范围） | `Confirmed` 标志 |
两阶段提交意味着键被 **声明** 之前处理并且 **确认** 之后。崩溃处理中间，键未确认；重试时，消费者重新获取并重新处理。 与事务性 outbox 的语义相同。
```csharp
// 在 DI 中注册持久化后端
builder.Services.AddSingleton<IIdempotentRepository, SqlIdempotentRepository>();
// 在任意路由中使用
From("kafka://payments")
.IdempotentConsumer(e => e.Message.GetHeader<string>("messageId")) // 声明 + 确认
.Process(...)
.To("sql://payments");
// 注册持久化后端
```
## 日志记录
```csharp
// 使用模板记录简单日志
.Log("处理订单：${header.orderId}");
// 使用工厂方法记录日志
.Log(e => $"订单 {e.Message.GetHeader<string>("orderId")} 正文大小：{e.Message.Body?.ToString()?.Length}");
// 结构化日志作用域
.Log(LogLevel.Information)
.Message("订单已处理")
.Header("orderId")
.Header("status")
.ShowRouteId()
.EndLog()
```
---
## 流畅传输构建器
每种传输方式都提供类型安全的流畅构建器。构建器实现了`implicit operator string`，因此可以直接传递给`From()`或`To()`方法。
### 内置（无需额外包）
```csharp
.To(Direct.Endpoint("my-service")) // direct:my-service
.To(Seda.Send("orders").ConcurrentConsumers(5).Size(1000)) // seda:orders?...
.From(Timer.Every("poll").Period(5000).Delay(1000)) // timer:poll?period=5000&delay=1000
.From(Timer.Once("init")) // 触发一次
.To(LogDsl.Info("audit").ShowHeaders().ShowBody()) // log:audit?...
.To(MockDsl.Endpoint("result").ExpectedMessageCount(3)) // mock:result?...
```
### 部分传输构建器
```csharp
// Kafka
Kafka.Topic("orders")
.Brokers("broker1:9092,broker2:9092")
.GroupId("order-svc")
.AutoOffsetReset("Earliest")
.Sasl("PLAIN", "user", "pass")
.Acks("All")
.Compression("zstd")
// RabbitMQ
Rabbit.Queue("orders")
.Host("rabbit.local").Port(5672)
.Exchange("orders-exchange", "topic")
.RoutingKey("order.new")
.Durable().PrefetchCount(10)
.DeadLetterExchange("dlx")
.QueueType("quorum")
// Redis
Redis.Subscribe("events")
Redis.XRead("orders-stream").ConsumerGroup("svc").ConsumerName("worker-1")
Redis.Set("cache-key").ConnectionString("localhost:6379").Ttl(3600)
Redis.LPush("work-queue")
// SQL
Sql.Poll("SELECT * FROM orders WHERE status = 'new'")
.DataSource("main").Delay(5000).OnSuccess("UPDATE orders SET status = 'done'")
Sql.Execute("INSERT INTO audit (msg) VALUES (@body)").DataSource("main").Transacted()
Sql.Procedure("usp_ProcessOrder").DataSource("main").CommandTimeout(30)
// HTTP / HTTPS
Http.Post("api.example.com/orders").BearerAuth().Timeout(5000).ContentType("application/json")
Http.Listen("/webhooks/stripe").Host("0.0.0.0").Port(8080).Cors().InOut()
// gRPC
Grpc.Call("localhost:50051")
Grpc.Listen("0.0.0.0:50051")
// 文件系统
FileDsl.Read("/data/inbox").Include("*.csv").Recursive().ReadLock("Changed").Delete()
FileDsl.Write("/data/outbox").AutoCreate(true).TempPrefix(".tmp").FileExist("Override")
// SFTP
Sftp.Directory("/remote/inbox")
.Host("sftp.example.com").Port(22)
.Username("user").PrivateKeyFile("/keys/id_rsa")
.Include("*.xml").Delete()
// MQTT 5.0
Mqtt.Subscribe("sensors/+/temperature").Broker("mqtt.local").Port(1883).Qos(1).CleanSession()
Mqtt.Publish("alerts/temperature").Broker("mqtt.local").Retain()
// AMQP 1.0
Amqp.Address("orders").Host("amqp.local").Port(5672).Username("admin").Password("admin")
// 电子邮件
Smtp.Send("smtp.gmail.com").Port(587).Username("user").Password("pass").UseSsl()
.From("noreply@example.com").To("admin@example.com")
Imap.Read("imap.gmail.com").Port(993).Username("user").Password("pass").UseSsl()
.Folder("INBOX").Unseen().Delete()
// TCP
Tcp.Listen("0.0.0.0:9000").Codec("textline").Delimiter("\n")
Tcp.Connect("server:9000").Codec("lengthprefixed").Tls()
// WebSocket
Ws.Listen("0.0.0.0:8080/ws")
Ws.Connect("ws.example.com:8080/stream")
// Quartz
Cron.Schedule("daily-report", "0 0 9 * * ?")
QTimer.Every("heartbeat").Period(5000)
```
AzureServiceBus、Elasticsearch、Firebase、FTP、IBM MQ、LDAP、S3 和 SignalR 的构建器遵循相同的模式 - 请参阅每个软件包的 README 以获取更多信息。
---
## 架构
路由分为两个阶段编译。定义阶段记录`RouteStep`实例；编译阶段将它们转换为处理程序链。这种分离使得在执行前可以进行路由自省、验证和可视化。
```mermaid
flowchart TB
subgraph 定义["1. 定义阶段"]
RB[路由构建器] --> Steps[路由步骤记录]
end
subgraph 编译["2. 编译阶段"]
Steps --> Compiler[路由编译器]
Compiler --> Pipeline[处理程序链]
end
subgraph 运行["3. 运行时"]
Source[来源：消费者] --> Pipeline
Pipeline --> Sink[目标：生产者]
Pipeline --> Telemetry[遥测：OpenTelemetry 活动 + 仪表]
end
```
### 核心引擎 - `redb.Route`
| 子系统 | 责任 |
|-----------|----------------|
| `Definitions/` | `RouteStep` 记录，`RouteCompiler`，`PipelineProcessor` |
| `Processors/` | 30+ EIP 处理器：过滤器、选择器、分割器、聚合器、WireTap、多播、收件人列表、动态路由器、重新排序器、分散聚合器、声明检查、幂等消费者、Saga、断路器、限流器、循环、延迟、去抖动、丰富、超时、TryCatch、OnException |
| `Expressions/` | `Body`、`Header`、`Property`、`Constant`、`Exchange` 加 `JPath`、`XPath`、`StringExpression`（通过 `System.Linq.Expressions` 编译） |
| `Predicates/` | 17 种谓词类型：`isEqualTo`、`isGreaterThan`、`contains`、`regex`、`In`、`isBetween`、`isNull` 等 |
| `Components/` | 内置传输：直接、SEDA、定时器、模拟、日志 |
| `Telemetry/` | OpenTelemetry `ActivitySource` + 每个路由和每个步骤的 `Meter` |
| `ErrorHandling/` | 重试、`DeadLetterChannel`、`TryCatch`、`OnException`（作用域和配置形式） |
| `Validation/` | JSON 模式、XSD、谓词验证 |
| `Serialization/` | JSON、XML 序列化/反序列化 |
| `Transactions/` | 通过 `.Transacted()` 集成 `TransactionScope` |
| `Fluent/` | 所有传输的类型安全端点构建器 |
### 组件层
```mermaid
flowchart TB
subgraph 核心["redb.Route - 核心引擎"]
Defs[定义]
Procs["处理器\n30+ EIP"]
Expr["表达式\n编译的 IL"]
Pred["谓词\n17 种类型"]
Built["内置组件\n直接 · SEDA · 定时器 · 模拟 · 日志"]
Tele[OpenTelemetry]
Err[错误处理]
Val[验证]
Ser[序列化]
Tx[事务范围]
end
subgraph 传输["redb.Route.* - 27 种传输"]
Msg["消息\nKafka · RabbitMQ · Redis · AzureSB · IbmMq · Amqp"]
Data["数据存储\nSql · Elasticsearch · Firebase · S3 · Ldap"]
Net["网络\nHttp · Grpc · WebSocket · SignalR · Tcp"]
Files["文件\nFile · Ftp · Sftp"]
Proto["协议 / 调度\nMail · MqttNet · Quartz"]
AI["AI / 进程\nLlm · Exec"]
Sup["支持\n控制器 · 通用文件 · 验证适配器"]
end
核心 --> 传输
```
### 传输 - 底层客户端和功能
| 传输 | 底层客户端 | 备注 |
|-----------|-------------------|-------|
| `Kafka` | `Confluent.Kafka` | 主题、消费者组、事务、SASL、压缩 |
| `RabbitMQ` | `RabbitMQ.Client` 7.x | 队列、交换机、死信队列、法定人数队列、预取 |
| `Redis` | `StackExchange.Redis` | 发布/订阅 + 流 + 键值存储操作 |
| `AzureServiceBus` | `Azure.Messaging.ServiceBus` | 队列、主题、会话、计划 |
| `IbmMq` | `IBMMQDotnetClient` | 队列 + 主题 + 事务 + RPC + SSL |
| `Amqp` | `AMQPNetLite` | AMQP 1.0 - Artemis、Azure SB、Amazon MQ；通过 `ReplyTo`/`CorrelationId` 进行 RPC |
| `Sql` | ADO.NET | 轮询、批量执行、存储过程、事务 |
| `Elasticsearch` | `Elastic.Clients.Elasticsearch` 8.x | 搜索、CRUD、9 种操作 |
| `Firebase` | Firebase 管理 SDK | Firestore、云存储、FCM |
| `S3` | `AWSSDK.S3` | AWS S3 + MinIO、多部分、SSE |
| `Ldap` | `Novell.Directory.Ldap` | 搜索、CRUD、身份验证 |
| `Http` | `HttpClient` + Kestrel | REST 客户端、Webhook、CORS、InOut |
| `Grpc` | `Grpc.Net` | 客户端 + Kestrel 服务器、二进制消息 |
| `WebSocket` | `ClientWebSocket` + Kestrel | 文本/二进制帧 |
| `SignalR` | `Microsoft.AspNetCore.SignalR` | 中心消费者 + 生产者/广播 |
| `Tcp` | `System.Net.Sockets` | 文本行和长度前缀 |
| `File` | 文件系统 | 轮询、原子写入、通配符、锁定 |
| `Ftp` | `FluentFTP` | FTP/FTPS、被动/主动 |
| `Sftp` | `SSH.NET` | 远程文件、密钥身份验证、代理 |
| `Mail` | `MailKit` | SMTP、IMAP、POP3、IDLE 推送 |
| `MqttNet` | `MQTTnet` | MQTT 5.0、共享订阅 |
| `Quartz` | `Quartz.NET` | Cron 调度、间隔定时器、持久作业 |
| `Exec` | `System.Diagnostics.Process` | 本地进程执行；允许列表、工作目录固定、超时、输出上限 |
| `Llm` | `HttpClient`（兼容 OpenAI + Anthropic 消息） | LLM 连接器 - 具有通用 OpenAI 兼容提供商（14 种供应商）和本地 Anthropic 提供商的工具循环代理引擎；**令牌逐个流式传输**（`?stream=true` → `IAsyncEnumerable<string>` 作为 SSE 在 `redb.Route.Http` 上公开，具有 `event: done` 摘要尾随 / 一个 `Text` 帧在 `redb.Route.WebSocket` 上每个令牌） |
| `Mcp` | JSON-RPC 2.0 via stdio / HTTP+SSE | 模型上下文协议客户端。生成外部 MCP 服务器，运行 `initialize` + `tools/list`，将每个远程工具投影到 `IToolDescriptorRegistry` 作为 `McpToolDescriptor`。生产者专用（`mcp://server/tool`）。取消转发为 `notifications/cancelled`。 |
### Saga 回滚流程
```mermaid
sequenceDiagram
participant R as 路由
participant S1 as 步骤 1：预留
participant S2 as 步骤 2：收费
participant S3 as 步骤 3：发货
R->>S1: action()
S1-->>R: ok
R->>S2: action()
S2-->>R: ok
R->>S3: action()
S3--xR: throw
Note over R: 回滚逆序
R->>S2: compensate()
R->>S1: compensate()
```
---
## DSL 参考
`IRouteDefinition` 方法的快速参考。所有方法都返回 `IRouteDefinition` 以进行链式调用，除非另有说明。
**转换/丰富**: `SetBody`（×4：静态、lambda、`IExpression`、`StringExpression`）、`Transform`（×3：lambda、`IExpression`、字符串）、`SetHeader`（×4）、`SetProperty`（×4）、`RemoveHeader`、`RemoveProperty`、`RemoveBody`、`Marshal`、`Unmarshal`
**过滤**: `Filter`（×3：lambda、`IPredicate`、字符串表达式）
**处理**: `Process`（×4：异步、异步 + `CancellationToken`、同步、`IProcessor`）
**基于内容的路由**: `Choice()`、`When`（×3：lambda、`IPredicate`、字符串）、`Otherwise()`、`EndChoice()`
**多播/ WireTap**: `Multicast`（简单 + 完整参数）、`WireTap`、`RecipientList`、`DynamicRouter`
**分割/聚合**: `Split`（lambda 或 `IExpression`，作用域或子路由）、`EndSplit()`、`Aggregate`
**循环/延迟**: `Loop`（计数或谓词，作用域或子路由）、`Delay`（`TimeSpan` 或 lambda）、`DelayExpression`
**错误处理**: `DoTry()`、`DoCatch<T>()`、`DoFinally()`、`OnException<T>()`、`EndOnException()`、`Retry`、`DeadLetterChannel`、`ThrowException`（×5）
**红elivery 配置**: `MaximumRedeliveries`、`RedeliveryDelay`、`BackOffMultiplier`、`UseExponentialBackOff`、`Handled`、`ExceptionHandled`
**日志记录**: `Log`（模板/ lambda / 作用域）、`EndLog()`
**验证**: `Validate`、`ValidateJsonSchema`（×2）、`ValidateXsd`（×3）、`ValidateFluent`、`ValidateAnnotations`
**生命周期**: `Transacted`、`RollbackAll`、`Respond`、`SetPattern`、`Stop`
**遥测**: `Traced`（作用域或内联）、`EndTraced()`、`Metered`（作用域或内联）、`EndMetered()`
**高级 EIP**: `Saga`、`IdempotentConsumer`、`Throttle`、`CircuitBreaker`、`Resequence`、`Enrich`、`PollEnrich`
**作用域**: `End()` - 通用作用域关闭器
---
## URI 方案参考
| 方案 | 包 | 入口点 |
|--------|---------|-------------|
| `direct:` | redb.Route | `Direct.Endpoint("name")` |
| `seda:` | redb.Route | `Seda.Consume("name")` / `Seda.Send("name")` |
| `timer:` | redb.Route | `Timer.Every("name")` / `Timer.Once("name")` |
| `log:` | redb.Route | `LogDsl.Info("name")` |
| `mock:` | redb.Route | `MockDsl.Endpoint("name")` |
| `kafka:` | redb.Route.Kafka | `Kafka.Topic("name")` |
| `rabbitmq:` | redb.Route.RabbitMQ | `Rabbit.Queue("name")` |
| `redis:` | redb.Route.Redis | `Redis.Subscribe("ch")` / `Redis.Set("key")` / … |
| `sql:` | redb.Route.Sql | `Sql.Poll("query")` / `Sql.Execute("stmt")` / `Sql.Procedure("sp")` |
| `http:` / `https:` | redb.Route.Http | `Http.Get("path")` / `Http.Listen("path")` / … |
| `grpc:` | redb.Route.Grpc | `Grpc.Call("host:port")` / `Grpc.Listen("host:port")` |
| `file:` | redb.Route.File | `FileDsl.Read("dir")` / `FileDsl.Write("dir")` |
| `sftp:` | redb.Route.Sftp | `Sftp.Directory("path")` |
| `mqtt:` | redb.Route.MqttNet | `Mqtt.Subscribe("topic")` / `Mqtt.Publish("topic")` |
| `amqp:` | redb.Route.Amqp | `Amqp.Address("name")` |
| `mail:` | redb.Route.Mail | `Smtp.Send("host")` / `Imap.Read("host")` / `Pop3.Read("host")` |
| `tcp:` | redb.Route.Tcp | `Tcp.Listen("host:port")` / `Tcp.Connect("host:port")` |
| `websocket:` | redb.Route.WebSocket | `Ws.Listen("host:port/path")` / `Ws.Connect("host:port/path")` |
| `cron:` / `qtimer:` | redb.Route.Quartz | `Cron.Schedule("name", "expr")` / `QTimer.Every("name")` |
| `exec:` | redb.Route.Exec | `ExecDsl.Run()`（生产者）/ `ExecDsl.Schedule(...)`（定时消费者） |
| `llm:` | redb.Route.Llm | `Llm.Factory("haiku")` / `Llm.Factory("openai")` / `Llm.Factory("groq")` / … |
| `mcp:` | redb.Route.Llm.Mcp | `mcp://serverName/toolName` - 通过 MCP `tools/list` 自动注册 |
| `asb:` | redb.Route.AzureServiceBus | `Asb.Queue("name")` / `Asb.Topic("name", "sub")` |
| `es:` | redb.Route.Elasticsearch | `Es.Index("name")` |
| `fstore:` / `fbstorage:` / `fcm:` | redb.Route.Firebase | `Firestore.Collection("name")` / `FbStorage.Bucket("b")` / `Fcm.Send()` |
| `ftp:` / `ftps:` | redb.Route.Ftp | `Ftp.Directory("/path")` |
| `wmq:` | redb.Route.IbmMq | `IbmMq.Queue("name")` / `IbmMq.Topic("name")` |
| `ldap:` / `ldaps:` | redb.Route.Ldap | `Ldap.Search("baseDn")` |
| `s3:` | redb.Route.S3 | `S3Dsl.Bucket("name")` |
| `sqs:` / `sns:` | redb.Route.Sqs | `Sqs.Queue("name")` / `Sns.Topic("name")` |
| `telegram:` | redb.Route.Telegram | `Tg.Receive(token)` / `Tg.Send(token)` |
| `signalr:` | redb.Route.SignalR | `SignalR.Hub("/hub")` |
> **注意。** 由于设计原因，没有 `quartz:` URI 方案。Quartz 调度通过工作主机（`ConfigureQuartz`）集成，并通过 `cron:`（Cron 表达式）或 `qtimer:`（间隔定时器）触发路由。`quartz:` URI 方案会将调度配置与路由寻址混淆。
---
## 传输能力矩阵
每个传输方式作为源（`From`）和目标（`To`）的能力，以及支持的高级功能。
| 传输 | 消费者（`From`） | 生产者（`To`） | 请求-响应 (InOut) | 事务 | 流式传输 | 备注 |
|-----------|:--:|:--:|:--:|:--:|:--:|-------|
| Direct | ✅ | ✅ | ✅ | — | — | 进程内同步 |
| SEDA | ✅ | ✅ | — | — | ✅ | 内存中有界队列，多消费者 |
| Timer | ✅ | — | — | — | — | 周期性/一次性触发 |
| Mock | ✅ | ✅ | — | — | — | 仅用于测试，记录消息 |
| Log | — | ✅ | — | — | — | 日志接收器 |
| Kafka | ✅ | ✅ | — | ✅ | ✅ | EOS：事务性生产者，`Acks=All`，`IsolationLevel=ReadCommitted`，手动提交 |
| RabbitMQ | ✅ | ✅ | ✅ | ✅ | — | 发布者确认 (`MaxOutstandingConfirms`)，`Mandatory` 标志，事务通道，自动恢复，RPC |
| Redis | ✅ | ✅ | — | — | ✅ | 发布/订阅 + 流 + 键值存储操作 |
| AzureServiceBus | ✅ | ✅ | ✅ | ✅ | — | 会话、计划、RPC |
| IbmMq | ✅ | ✅ | ✅ | ✅ | — | 队列 + 主题 + 事务 + RPC + SSL |
| Amqp | ✅ | ✅ | ✅ | — | — | AMQP 1.0；通过 `ReplyTo`/`CorrelationId` 进行 RPC |
| Sql | ✅ (轮询) | ✅ | — | ✅ | — | 轮询源，批量接收，存储过程 |
| Elasticsearch | — | ✅ | — | — | — | 索引，搜索，CRUD |
| Firebase | ✅ (Firestore) | ✅ | — | — | ✅ | Firestore，云存储，FCM |
| S3 | ✅ (轮询) | ✅ | — | — | ✅ | 分段上传/下载 |
| SQS | ✅ (轮询) | ✅ | — | ✅ | — | Amazon SQS - 长轮询，`ConcurrentConsumers(N)`，FIFO，可见性超时 + 心跳，事务确认 |
| SNS | — | ✅ | — | — | — | Amazon SNS - 主题发布，FIFO，SNS→SQS 风扇 |
| Telegram | ✅ (长轮询) | ✅ | — | — | — | 机器人 API - 发送/文档/照片/编辑/删除/答案，429 `retry_after`，内联/回复键盘，Webhook 解包；最多一次 |
| Ldap | — | ✅ | — | — | — | 搜索和 CRUD |
| Http | ✅ (监听) | ✅ | ✅ | — | ✅ | Kestrel 服务器 + HttpClient |
| Grpc | ✅ (监听) | ✅ | ✅ | — | ✅ | 单向 + 流式 |
| WebSocket | ✅ | ✅ | — | — | ✅ | 文本/二进制帧 |
| SignalR | ✅ (中心) | ✅ | — | — | ✅ | 中心服务器 + 客户端广播 |
| Tcp | ✅ | ✅ | — | — | ✅ | 文本行和长度前缀 |
| File | ✅ (轮询) | ✅ | — | — | — | 原子写入，文件锁定 |
| Ftp | ✅ (轮询) | ✅ | — | — | — | FTP/FTPS |
| Sftp | ✅ (轮询) | ✅ | — | — | — | SSH 密钥身份验证，代理 |
| Mail | ✅ (IMAP/POP3) | ✅ (SMTP) | — | — | — | IDLE 推送用于 IMAP |
| MqttNet | ✅ | ✅ | — | — | ✅ | MQTT 5.0，共享订阅 |
| Quartz | ✅ | — | — | — | — | Cron / 间隔触发器 |
| Exec | ✅ (定时) | ✅ | ✅ | — | — | 本地进程；`AllowedCommands` 允许列表，`WorkingDirectory` 固定，`TimeoutMs`，字节上限；主要用作 `.AsLlmTool("shell")` |
| Llm | — | ✅ | ✅ | — | ✅ | LLM 连接器 - 通用 OpenAI 兼容（14 种供应商）+ 本地 Anthropic 消息 API；工具循环代理引擎；通过 `AddRedbLlmStorage()` 持久化对话/审批/预算/幂等存储 |
| Mcp | — | ✅ | — | — | — | MCP 客户端连接器 - 生成外部 MCP 服务器（stdio + HTTP/SSE），通过 `tools/list` 自动发现工具并将它们注册为 `ILlmToolDescriptor`。取消转发为 `notifications/cancelled`；在传输失败时自动重启 |
## 软件包
| 软件包 | NuGet | 描述 |
|---------|-------|-------------|
| `redb.Route` | [](https://www.nuget.org/packages/redb.Route) | 核心引擎：DSL、处理器、表达式、遥测 |
| `redb.Route.Kafka` | [](https://www.nuget.org/packages/redb.Route.Kafka) | Apache Kafka（Confluent） |
| `redb.Route.RabbitMQ` | [](https://www.nuget.org/packages/redb.Route.RabbitMQ) | RabbitMQ（官方客户端 7.x） |
| `redb.Route.Redis` | [](https://www.nuget.org/packages/redb.Route.Redis) | Redis（StackExchange）— 发布/订阅、流、键值存储、列表 |
| `redb.Route.Sql` | [](https://www.nuget.org/packages/redb.Route.Sql) | SQL 数据库（ADO.NET）— 轮询、执行、存储过程 |
| `redb.Route.Http` | [](https://www.nuget.org/packages/redb.Route.Http) | HTTP/HTTPS — 客户端和 Kestrel 服务器 |
| `redb.Route.Grpc` | [](https://www.nuget.org/packages/redb.Route.Grpc) | gRPC — 客户端和 Kestrel 服务器 |
| `redb.Route.File` | [](https://www.nuget.org/packages/redb.Route.File) | 文件系统 — 轮询、原子写入、锁定 |
| `redb.Route.Sftp` | [](https://www.nuget.org/packages/redb.Route.Sftp) | SFTP（SSH.NET） |
| `redb.Route.MqttNet` | [](https://www.nuget.org/packages/redb.Route.MqttNet) | MQTT 5.0（MQTTnet） |
| `redb.Route.Amqp` | [](https://www.nuget.org/packages/redb.Route.Amqp) | AMQP 1.0（Artemis、Azure SB、Amazon MQ） |
| `redb.Route.Mail` | [](https://www.nuget.org/packages/redb.Route.Mail) | 电子邮件 — SMTP、IMAP、POP3（MailKit） |
| `redb.Route.Tcp` | [](https://www.nuget.org/packages/redb.Route.Tcp) | 原始 TCP — 文本行、长度前缀、TLS |
| `redb.Route.WebSocket` | [](https://www.nuget.org/packages/redb.Route.WebSocket) | WebSocket — 客户端和 Kestrel 服务器 |
| `redb.Route.Quartz` | [](https://www.nuget.org/packages/redb.Route.Quartz) | Quartz.NET 调度（Cron + 计时器） |
| `redb.Route.Core` | [](https://www.nuget.org/packages/redb.Route.Core) | 与 [redb](https://github.com/redbase-app/redb) 类型的对象存储的桥接 — `IRedbService` 访问、幂等存储库 |
| `redb.Route.AzureServiceBus` | [](https://www.nuget.org/packages/redb.Route.AzureServiceBus) | Azure 服务总线 — 队列、主题、会话 |
| `redb.Route.Elasticsearch` | [](https://www.nuget.org/packages/redb.Route.Elasticsearch) | Elasticsearch 8.x — 搜索、CRUD、9 种操作 |
| `redb.Route.Firebase` | [](https://www.nuget.org/packages/redb.Route.Firebase) | Firebase — Firestore、云存储、FCM |
| `redb.Route.Ftp` | [](https://www.nuget.org/packages/redb.Route.Ftp) | FTP/FTPS（FluentFTP） |
| `redb.Route.GenericFile` | [](https://www.nuget.org/packages/redb.Route.GenericFile) | 文件、FTP、SFTP 传输的基础库 |
| `redb.Route.IbmMq` | [](https://www.nuget.org/packages/redb.Route.IbmMq) | IBM MQ — 队列、主题、事务、RPC |
| `redb.Route.Ldap` | [](https://www.nuget.org/packages/redb.Route.Ldap) | LDAP / 活动目录 — 搜索、CRUD、身份验证 |
| `redb.Route.S3` | [](https://www.nuget.org/packages/redb.Route.S3) | AWS S3 / MinIO — 上传、下载、多部分 |
| `redb.Route.Sqs` | [](https://www.nuget.org/packages/redb.Route.Sqs) | Amazon SQS + SNS — 队列消费/生产、FIFO、SNS 发布 + SNS→SQS |
| `redb.Route.Telegram` | [](https://www.nuget.org/packages/redb.Route.Telegram) | Telegram Bot API — 长轮询消费者 + 发送/文档/照片/编辑/删除/答案生产者、429 处理、内联/回复键盘、网络钩子解包 |
| `redb.Route.SignalR` | [](https://www.nuget.org/packages/redb.Route.SignalR) | SignalR — 中心服务器和客户端/广播生产者 |
| `redb.Route.Llm` | [](https://www.nuget.org/packages/redb.Route.Llm) | LLM 连接器 — `llm:` URI、流畅的 DSL、代理循环、通用 OpenAI 兼容提供程序（14 个 API）、本地 Anthropic Messages API 提供程序、`AddRedbLlmStorage()`（5 个存储库 / 9 个模式） |
| `redb.Route.Llm.Abstractions` | [](https://www.nuget.org/packages/redb.Route.Llm.Abstractions) | LLM 工具能力契约 — `ILlmToolDescriptor`、`LlmToolCapability`、`RouteToolBridge`、`[ExposeAsLlmTool]`、`.AsLlmTool()` DSL 方面 |
| `redb.Route.Llm.Tools` | [](https://www.nuget.org/packages/redb.Route.Llm.Tools) | 实用 LLM 工具 — HttpFetch、JsonPath、XPath、MathEval、RegexExtract、Tavily 网络搜索 |
| `redb.Route.Llm.Mcp` | [](https://www.nuget.org/packages/redb.Route.Llm.Mcp) | MCP 客户端连接器 — `mcp:` URI、stdio + HTTP/SSE 传输、通过 `tools/list` 自动发现、取消转发为 `notifications/cancelled`、传输失败时自动重启 |
| `redb.Route.Exec` | [](https://www.nuget.org/packages/redb.Route.Exec) | 本地进程执行 — `exec://run` 生产者 + 调度消费者、允许列表、工作目录固定、超时、输出上限 |
| `redb.Route.Controllers` | [](https://www.nuget.org/packages/redb.Route.Controllers) | 控制器调度 — 属性路由、DI、InOut |
| `redb.Route.Validation.Adapters` | [](https://www.nuget.org/packages/redb.Route.Validation.Adapters) | 验证适配器 — FluentValidation + DataAnnotations |
共 34 个软件包：核心引擎 + 27 个传输（包括 `llm:` 和 `mcp:`）+ 6 个支持库（Core、Controllers、GenericFile、Validation.Adapters、Llm.Abstractions、Llm.Tools）。
---
## 在生产环境中运行路由 - redb.Tsak
编写路由是一回事；在生产环境中运行路由是另一回事。 **[redb.Tsak](../redb.Tsak)** 是 redb.Route 的运行时容器 — 一个 .NET 等效于 [Apache Karaf](https://karaf.apache.org/) / [Camel K](https://camel.apache.org/camel-k/)。它托管路由上下文、管理它们的生命周期、在集群中分发它们、调度作业、捕获指标和日志，并通过 REST API、CLI 和 Blazor 仪表板公开所有内容。
```mermaid
flowchart TB
subgraph Tsak["redb.Tsak 运行时"]
Worker[Worker / IHostedService]
Modules[(模块程序集加载上下文\n热重载)]
Ctx[上下文管理器\n启动 / 停止 / 重启]
Cluster[集群协调器\n领导者选举 + 再平衡]
Quartz[Quartz 调度器\nAdoJobStore]
Metrics[指标 + 监视器\n环形缓冲区日志]
Sec[API 密钥身份验证\n角色 + 审计]
end
subgraph Surfaces["操作员界面"]
REST[REST API\n13 个控制器 / 45+ 端点]
CLI[tsak CLI\n11 个命令组 / 30+ 命令]
UI[Blazor 仪表板\n11 个页面]
end
subgraph Storage["持久状态"]
Redb[(redb\nPostgres / MSSql)]
end
Modules --> Worker
Worker --> Ctx
Worker --> Cluster
Worker --> Quartz
Worker --> Metrics
Worker --> Sec
Tsak --> Surfaces
Cluster --> Redb
Sec --> Redb
Quartz --> Redb
```
### 项目布局
| 项目 | 目的 |
|---------|---------|
| `redb.Tsak.Core` | 控制器、服务、安全、热重载、监视 |
| `redb.Tsak.Worker` | 主机（`IHostedService`）、DI、Serilog、Quartz 模式初始化 |
| `redb.Tsak.Core.Pro` | 集群（领导者选举 + 节点注册）、监视器、许可 |
| `redb.Tsak.Contracts` | 线DTO（System.Text.Json），由 Core / Client / Web / CLI 共享 |
| `redb.Tsak.Client` | REST API 的类型化 HTTP 客户端 |
| `redb.Tsak.CLI` | `tsak` 命令 — System.CommandLine、配置文件、监视模式 |
| `redb.Tsak.Web` | Blazor Server 仪表板 |
| `redb.Tsak.Web.Pro` | PRO 网络（扩展身份验证） |
### 模块加载和热重载
Tsak 区分 **两个** 加载位置，它们不可互换：
| 路径 | 默认值 | ALC | 加载时 | 目的 |
|------|---------|-----|-------------|---------|
| `Libs/shared/` | `HotReloadOptions.SharedPath` | **默认** ALC | 启动前，模块发现前 | 连接器、公共 DTO、契约、原生依赖项 — 必须在所有模块中具有 **单个类型标识** |
| `Libs/modules/` | `Tsak:Modules:AssemblyPaths` | **隔离** `ModuleAssemblyLoadContext` 每个包（`pkg:{Name}`） | 启动 + `.tpkg` 更改时 | 应用程序路由模块，隔离以防止版本冲突 |
- **版本控制** — `ModuleVersion` 格式 `yyyy.MM.dd.HHmm`；`KeepVersions` 保留以前的版本以进行回滚。
- **热重载** — `HotReloadService` 监视目录；新 DLL → 加载到新的 ALC → 平滑交换 → 失败时回滚。
- **可收集的 ALC** — 按需选择（`Collectible=true`）；默认关闭，因为 XmlSerializer / Reflection.Emit 固定程序集。启用后，GC 回收旧版本。
- **`.tpkg` 包** — 带有 `manifest.json`（`Name`、`Version`、`EntryPoints[]`、`Dependencies[]`）、模块 DLL 和每个模块的 `{ModuleName}.config.json` 的 ZIP 存档。原子部署 / 卸载。
- **`SharedAssemblyLoader`** — 扫描 `Libs/shared/`，通过 `AssemblyLoadContext.Default.LoadFromStream` 加载，使用程序集特定的 **本机库解析器**（探测本机子目录），跟踪文件时间戳进行更改检测，容忍 `FileLoadException`（版本冲突记录，非致命）。
- **`LoadedAssemblyTracker`** — 中央注册表，防止 **类型标识分裂脑**：相同程序集名称 → 单个 `Assembly` 实例跨所有 ALC。注册一次性 `Default.Resolving` 处理程序。`LoadOrReuse(name, bytes)` 在线程安全的一个加载过程中使用 `Lazy<T>` 进行每个名称加载；`Replace` 启用热重载强制替换（旧版本保留在内存中 - 默认 ALC 不变）。当打开 `.tpkg` 时，入口点进入隔离的 ALC，但配套 DLL 通过跟踪器进入默认 ALC，以便跨模块类型匹配。
#### 模块清单 (`manifest.json`)
```jsonc
{
"Name": "OrdersModule", // 必需
"Version": "2026.05.06.1130", // 信息性
"EntryPoints": [ "OrdersModule.dll" ], // 加载到 pkg:Orders ALC
"Dependencies": [ "SharedContracts" ] // 通过跟踪器加载到默认
}
```
### 集群模式（PRO）
通过 `Tsak:Cluster:Enabled=true` 启用。状态保存在 redb 中 - 无需额外的协调服务（无 ZooKeeper，无 etcd，无 Consul）。集群形成基于树的拓扑结构：
```
cluster:default (_tsak_clusters)
└── group:default (_tsak_groups)
└── node:NODE_ID (_tsak_nodes)
```
| 组件 | 角色 |
|-----------|------|
| `RedbLeaderElection` | 通过 redb 的分布式锁，带有 TTL 过期 + **纪元隔离**（防止分裂脑） |
| `RedbNodeRegistry` | 心跳跟踪，死节点清理 |
| `ClusterCoordinator` | 背景循环：心跳 → 领导者选举 →（如果领导者）检测死节点 + 重新平衡任务分配 |
| `AssignmentManager` | 跨节点分发上下文/模块 |
| `IContextCustomizer` | 插件点 - 集群模式注入 `ClusteredRoutePolicy` 以协调路由所有权 |
协调器在模块发现之前启动，以避免引导期间的竞争。Quartz `AdoJobStore` 在所有节点中共享已调度的作业 - 只有选定的领导者为作业触发。
### 生命周期和优雅关闭
```mermaid
stateDiagram-v2
[*] --> Created
Created --> Started: Start
Started --> Running: routes ready
Running --> Stopping: Stop
Stopping --> Stopped: drain in-flight
Stopped --> Started: Restart
Stopped --> [*]: Remove
```
**托管服务启动顺序** （Worker DI）：
1. `QuartzSchemaInitializer` — 通过原始 ADO.NET（无 redb 依赖，无引导死锁）幂等地应用 Quartz DDL（Postgres/MSSql）
2. `QuartzHostedService` — 启动调度器（RAM 或 AdoJobStore）
3. `MetricsCollectionService` — 将定期进程 / GC 采样到循环缓冲区
4. `TsakHostedService`（BackgroundService）— 主协调器：共享程序集加载器 → 模块发现 → 上下文启动 → 集群注册
5. `RouteWatchdogService`（BackgroundService）— 挂起交换检测器
**关闭序列** (`TsakHostedService.StopAsync`) :
1. 记录 `"Tsak 正常关闭，停止所有上下文..."`
2. 以 **反向顺序** 运行 `ITsakLifecycleHook.OnStoppingAsync()` — 启动的最后一个钩子是停止的第一个钩子。集群注销是一个钩子 → 在上下文停止之前释放领导权。
3. 对于每个上下文：使用 **单独的 `CancellationToken.None`** 调用 `ITsakContextManager.StopContextAsync()` — 故意不传播主机关闭令牌，以便一个缓慢的上下文不会取消其他上下文的排空。每个上下文拥有其内部的排空超时以应对未完成的交换。
4. `base.StopAsync()` 释放托管服务资源。
5. Quartz 调度器停止，日志接收器刷新。
停止/重启 REST 响应包括 `DrainTimeMs` 和 `InflightAfter` — 操作员可以准确看到排空需要多长时间以及是否有任何内容被强制终止。SIGTERM 处理委托给标准 .NET 主机（`IHostApplicationLifetime`）。
- **正常停止** — `context:graceful-stop-timeout` 排空未完成的交换；超时后强制停止。
- **生命周期钩子** — `ITsakLifecycleHook` 在模块发现之前和关闭后运行（停止时反向顺序）。
- **生命周期审计** — `LifecycleAuditService` 记录每个启动/停止/错误事件，可以通过 `/api/lifecycle` 查询。
- **监视器** — `RouteWatchdogService` 标记长时间运行和挂起的未完成交换；可配置的阈值，可选的自动重启。
### REST API — 13 个控制器，45+ 端点
| 区域 | 端点 |
|------|-----------|
| **身份验证** | `POST /api/auth/keys`，`GET /api/auth/keys`，`DELETE /api/auth/keys/{id}` |
| **模块** | `GET /api/modules`，`GET /api/modules/{name}`，`DELETE /api/modules/{name}` |
| **上下文** | `GET`，`POST start/stop/restart`，`DELETE`，`POST reset-route-states`，`GET endpoints` |
| **路由** | `GET`，`POST start/stop/force-stop`，`GET inflight`，`GET metrics`（每个上下文） |
| **系统** | `GET /api/system/health`，`/metrics`，`/metrics/history`，`/info` |
| **调度器** | `GET status/jobs/running`，`POST start/standby/pause/resume` |
| **日志** | `GET /api/logs`（环形缓冲区），`GET /api/logs/files`，`GET /api/logs/files/{name}` |
| **健康检查** | `GET /api/system/health/startup` + `/live` + `/ready`（适用于 Kubernetes） |
| **集群**（PRO） | `GET /api/cluster status/nodes`，`POST rebalance` |
| **生命周期** | `GET /api/lifecycle`（审计事件） |
| **仪表板** | `GET /api/dashboard`（摘要） |
| **诊断** | `GET /api/diagnostics`（完整转储，每个路由转储） |
| **监视器**（PRO） | `GET /api/watchdog status/alerts`，`POST enable/disable` |
| **用户**（PRO） | `GET / POST / PUT / DELETE /api/users` |
### CLI — `tsak`
基于 System.CommandLine。11 个命令组，30+ 命令，配置文件管理，三个输出格式。
| 组 | 命令 |
|-------|----------|
| `system` | `health`，`metrics`，`info` |
| `context` | `list`，`get`，`start`，`stop`，`restart`，`remove` |
| `route` | `list`，`get`，`start`，`stop`，`force-stop`，`inflight` |
| `module` | `list`，`get`，`remove` |
| `scheduler` | `status`，`jobs`，`running`，`start`，`standby`，`pause`，`resume` |
| `cluster` | `status`，`nodes`，`rebalance` |
| `log` | `get`（带 `--limit`， `--level`） |
| `auth` | `list`，`create`，`revoke` |
| `watchdog` | `status`，`alerts`，`enable`，`disable` |
| `diagnostics` | `lifecycle`，`dump`，`route`，`dashboard` |
| `profile` | `list`，`add`，`remove`，`use`（本地 - 无 API 调用） |
全局选项： `--profile/-p`， `--url/-u`， `--key/-k`， `--output table|json|quiet`， `--watch` + `--interval`（实时轮询）， `--yes/-y`（跳过确认）， `--no-color`， `--timeout`。配置文件保存在 `~/.tsak/profiles.json` 中；环境变量 `TSAK_URL` / `TSAK_KEY` 覆盖。
```bash
# 切换配置文件，实时监视路由
tsak profile use prod
tsak route list --watch --interval 2
# 集群操作
tsak cluster status
tsak cluster rebalance --yes
# 通过 REST + 调度操作进行热部署模块
tsak module list -o json
tsak scheduler jobs
tsak scheduler pause --job orders.daily-report
```
### Web UI — Blazor Server 控制面板
| 页面 | 路由 | 用途 |
|------|-------|---------|
| **控制面板** | `/` | 概述：上下文、路由、指标、警报 |
| **集群** | `/cluster` | 拓扑结构、领导者、节点注册、心跳 |
| **节点详情** | `/node/{id}` | 节点选项卡：上下文、模块、健康状态 |
| **路由** | `/node/{id}/routes` | 上下文路由列表、状态、正在进行中 |
| **路由视图** | `/node/{id}/routes/{ctx}/{routeId}` | 处理器、指标、正在进行中的交换 |
| **端点** | `/node/{id}/endpoints` | HTTP 端点，从/到每个路由 |
| **日志** | `/node/{id}/logs` | 环形缓冲区日志查看器，带有级别过滤器 |
| **监视器** | `/node/{id}/watchdog` | 活动警报、卡住的交换 |
| **身份验证** | `/auth` | API 密钥管理（创建/撤销） |
| **登录** (专业版) | `/login` | 身份验证页面 |
使用 Chart.js 进行指标显示，表格组件，周期性刷新（SignalR 准备就绪，适用于推送）。
### Quartz 调度器
| 模式 | 作业存储 | 用途 |
|------|-----------|-----|
| 独立 | `RAMJobStore`（内存中） | 开发，单节点，无数据库 |
| 单节点 + redb / 集群 | `JobStoreTX` (`AdoJobStore`) | 作业在重启后存活，并在集群中共享 |
`QuartzSchemaInitializer` 是一个 `IHostedService`，在 Quartz 调度器启动前运行 - 它为 Postgres 和 MSSql 应用嵌入式 DDL（可重复执行，无 redb 依赖，以避免启动死锁）。每个 `IRouteContext` 通过 DI 接收一个 `IScheduler`；Quartz 传输（`cron:` / `qtimer:`）是触发器的入口。
### 身份验证和安全性
- **API 密钥** — `Authorization: Bearer tsak_XXX` 或 `X-Api-Key: tsak_XXX`。
- **HMAC-SHA256** 验证，具有恒定时间比较（防止时间攻击）。
- **角色** — 以逗号分隔，由 `AuthorizeProcessor` 强制执行。
- **过期 + 撤销** — 密钥可以具有 `ExpiresAt`；撤销是立即的，缓存（5 分钟 TTL）被失效。
- **用户链接** — 如果注册了 `IUserProvider`，则密钥链接到用户并继承 `Disabled` 标志。
- **存储** — `ConfigApiKeyStore`（appsettings，开发）或 `RedbApiKeyStore`（RTTI 存储，集群复制）。
- **审计** — 每个创建/撤销/验证事件都会被记录（`LogAdminAuditService`）；`[AuditAdminAction]` 属性标记值得审计的端点。
### 可观察性
#### OpenTelemetry 和 Prometheus
通过 NuGet 包实现一流的 OTel 集成：
- `OpenTelemetry.Extensions.Hosting`
- `OpenTelemetry.Exporter.Prometheus.HttpListener`
- `OpenTelemetry.Instrumentation.Process`
- `OpenTelemetry.Instrumentation.Runtime`
当 `Tsak:Metrics:Prometheus:Enabled=true` 时，Tsak 调用 `.AddOpenTelemetry().WithMetrics(b => b.AddPrometheusHttpListener(...))` 并在 **`http://*:9464/`** （端口可通过 `Tsak:Metrics:Prometheus:Port` 配置）上公开一个 Prometheus 格式的端点。OTel 管道注册了 redb.Route `ActivitySource`（`RouteActivitySource.SourceName`），因此由路由处理器发出的任何跟踪都会自动收集 - 将 OTLP 收集器/Jaeger/Tempo 指向活动源，或从 Prometheus 抓取 `:9464/metrics`。
```jsonc
"Tsak": {
"Metrics": {
"Prometheus": { "Enabled": true, "Port": 9464 }
}
}
```
#### 健康探测 - Kubernetes 就绪
| 端点 | 探测类型 | 返回 200 时 |
|----------|-----------|------------------|
| `GET /api/system/health/startup` | 启动 | 进程启动 |
| `GET /api/system/health/live` | 存活 | 进程未死锁 |
| `GET /api/system/health/ready` | **就绪** | 上下文正在运行 **且** 健康（比存活更严格） |
`HealthCheckService` 聚合探测（最差状态赢得：不健康 > 降级 > 健康），并且永远不会抛出 - 探测内部的异常成为不健康。模块可以通过实现 `IModuleHealthContributor` 贡献自定义探测。专业版提供 `ClusterHealthContributor`，它报告领导者/成员健康状况。
#### 指标和日志
- **进程指标** — CPU，工作集，线程，GC Gen0/1/2，总内存；具有可配置历史记录的循环缓冲区（默认为 24 小时）。
- **每个上下文的指标** — 处理的消息，错误率，吞吐量。
- **日志捕获** — Serilog `RingBufferSink`（内存中的循环缓冲区，大小通过 `Tsak:Logs:BufferSize`，默认为 2000）以及 `appsettings.json` 中声明的可选文件/控制台接收器。`MemoryUsageEnricher` 为每个日志事件添加了一个 `MemoryUsage` 字段（KB）。
- **生命周期审计** — 通过 `/api/lifecycle` 可查询每个上下文/路由启动/停止/错误。
- **管理员审计** — `[AuditAdminAction]` 端点默认发出结构化的 WRN 日志，专业版可以将接收器替换为基于 redb 的存储。
### 部署模式
| 模式 | 状态存储 | Quartz | API 密钥 | 集群 |
|------|-------------|--------|----------|---------|
| **独立** | `InMemoryTsakStateStore` | `RAMJobStore` | `ConfigApiKeyStore`（appsettings） | 关闭 |
| **单节点 + redb** | `RedbTsakStateStore`（Postgres / MSSql） | `AdoJobStore` | `RedbApiKeyStore`（RTTI 存储） | 关闭 |
| **集群** | 共享 `RedbTsakStateStore` | 共享 `AdoJobStore` | 共享 `RedbApiKeyStore` | `RedbLeaderElection` + `RedbNodeRegistry` |
最小配置：
```jsonc
{
"Tsak": {
"Redb": { "Provider": "postgres", "ConnectionString": "..." },
"Modules": {
"AssemblyPaths": [ "Libs/modules", "Libs/shared" ]
},
"HotReload": {
"Collectible": false, // true 启用 GC 回收旧版本
"KeepVersions": 3
},
"Cluster": {
"Enabled": true,
"ClusterName": "default",
"GroupName": "default",
"NodeId": null, // null = 自动生成/从环境变量中选择
"HeartbeatInterval": 5000,
"LeaderLockTtlSeconds": 10
}
},
"Quartz": {
"quartz.jobStore.type": "Quartz.Impl.AdoJobStore.JobStoreTX, Quartz"
}
}
```
### Kubernetes 部署
Tsak 提供了一个现成的 `Dockerfile`（多阶段，`mcr.microsoft.com/dotnet/runtime:9.0`，暴露 9090 端口），从头到尾设计用于容器部署。
#### K8s 探测
健康端点默认情况下是身份验证豁免的（`Tsak:Api:AuthExempt`）：
```yaml
startupProbe:
httpGet: { path: /api/system/health/startup, port: 9090 }
failureThreshold: 30
periodSeconds: 4 # 最多 120 秒启动
livenessProbe:
httpGet: { path: /api/system/health/live, port: 9090 }
periodSeconds: 10
failureThreshold: 3 # 30 秒后重启 Pod
readinessProbe:
httpGet: { path: /api/system/health/ready, port: 9090 }
periodSeconds: 5
failureThreshold: 2 # 10 秒后从负载均衡器中移除
```
**存活** 不检查模块健康状况 - 这可以避免在滚动更新期间重启循环。**就绪** 更严格：降级状态 → Pod 从服务端点中移除（不会重启），允许集群重新平衡。
#### 优雅终止
K8s 发送 SIGTERM，然后在 `terminationGracePeriodSeconds` 后发送 SIGKILL。设置 `Tsak:Shutdown:TimeoutSeconds`（推荐：`terminationGracePeriodSeconds − 5` 用于集群注销缓冲区）。然后关闭序列变为：
```
SIGTERM → 集群注销（钩子，逆序）→
上下文排空（隔离 CT，最多 TimeoutSeconds）→
Quartz 关闭 → 日志刷新 → SIGKILL 永不需
```
#### 集群模式下的 Pod 身份
当以 StatefulSet 或无头服务运行，注入 Pod 元数据通过向下 API：
```yaml
env:
- name: POD_NAME
valueFrom: { fieldRef: { fieldPath: metadata.name } }
- name: POD_NAMESPACE
valueFrom: { fieldRef: { fieldPath: metadata.namespace } }
- name: POD_IP
valueFrom: { fieldRef: { fieldPath: status.podIP } }
- name: Tsak__Cluster__NodeId
valueFrom: { fieldRef: { fieldPath: metadata.name } }
```
Tsak 使用 `Tsak:Cluster:NodeId` 作为其集群身份。将其映射到 `metadata.name` 可在 Pod 重启时提供稳定、可预测的节点 ID（对于分配重新平衡非常重要）。
#### Prometheus 抓取
```yaml
annotations:
prometheus.io/scrape: "true"
prometheus.io/port: "9464"
prometheus.io/path: "/metrics"
```
设置 `Tsak:Metrics:Prometheus:Enabled=true` 以激活 OTel Prometheus 导出器，监听端口 9464。
#### 可插拔的集群后端
所有集群协调都位于抽象接口后面（`ILeaderElection`、`IDistributedLock`、`INodeRegistry`、`IClusterCoordinator`、`IClusterBootstrap`、`IAssignmentManager`）位于 `redb.Tsak.Core.Pro` 中。目前的实现使用 redb（Postgres / MSSql）。设计允许在不更改任何其他代码的情况下，插入 Kubernetes Lease API 实现（`KubernetesLeaderElection`、`KubernetesNodeRegistry`）- 只需在调用 `AddTsakCluster()` 之前注册替代实现。
### 为什么选择 redb.Tsak（而非 Apache Karaf / Camel K）
- **没有 JVM，没有 OSGi。** 纯 .NET 8/9/10，具有本地 `AssemblyLoadContext` 隔离，而不是 OSGi 包。
- **集群无需外部协调。** 领导者选举 + 节点注册，在现有的 redb（Postgres / MSSql）中持久化。无需 ZooKeeper，无需 etcd，无需 Consul。
- **内置 Blazor 控制面板。** 完整的运行时管理 UI 与 Karaf 需要 Hawtio，Camel K 需要 Kamel CLI + 外部控制面板。
- **CLI + REST + Web 对称。** 相同的 DTO（`redb.Tsak.Contracts`）跨所有三个接口；CLI 只是一个 API 的薄客户端。
- **类型标识安全模块加载。** `LoadedAssemblyTracker` 防止经典的 OSGi 样式“同一类型的两个版本”错误，而无需对模块作者进行清单体操。
- **开箱即用的 OpenTelemetry + Prometheus。** `:9464/metrics` Prometheus 端点，redb.Route 的 OTel `ActivitySource` 已注册 - 直接插入收集器/Jaeger/Tempo。
- **Kubernetes 原生探测。** `/api/system/health/*` 下的独立启动/存活/就绪端点。
- **幂等架构初始化。** Quartz 表和 redb 架构在启动时引导 - 无需手动 DDL 步骤。
- **热重载和回滚。** 原子 `.tpkg` 包，版本保留，失败时自动回滚。
- **确定性关闭排序。** 钩子按逆序触发；上下文排空由 `CancellationToken.None` 隔离，因此一个缓慢的上下文不会中止其他上下文。
有关更深入的功能参考和部署指南，请参阅 [redb.Tsak/STATUS.md](../redb.Tsak/STATUS.md)。
---
## 贡献
欢迎贡献。帮助最有价值的领域：
- 带有最小重现的错误报告
- 文档改进
- 新的传输连接器
- 示例和教程
---
## 许可证
Apache License 2.0 - 请参阅 [LICENSE](LICENSE)。
[RedBase](https://github.com/redbase-app) 生态系统的一部分。
连接信息
猜你喜欢
暂无推荐服务器