• 示例工程

本指南基于当前仓库 dotnet/README.md 中收录的 RabbitMQ 官方教程 C# 移植版本，系统讲解在 .NET 8 与 RabbitMQ.Client 7.2.x 环境下，如何依次跑通"Hello World、工作队列、发布/订阅、路由、主题、RPC、发布者确认"七套官方教程。文中不仅完整保留原文档的全部运行命令，还逐份对照仓库源码（dotnet/目录下的 13 个可执行项目）分析其关键 API 调用与设计要点，读完你既能照单运行每个示例，也能理解 RabbitMQ 异步客户端、quorum 队列、消息确认、交换机绑定等核心机制的源码级实现。

环境准备与项目结构

原文档明确要求：使用示例前必须先有一个正在运行的 RabbitMQ 服务（推荐监听在localhost的默认节点）。无论 Windows 还是 Linux，唯一的前置开发依赖是 .NET（dotnet core），本仓库的示例面向 .NET 8；仓库还额外附带了一份面向 Visual Studio 2022 的解决方案文件。官方教程强调命令行方式编译运行，因此本文所有示例均以dotnet run --project <路径>形式给出，每个命令建议在独立的控制台/终端标签页中执行。

仓库 dotnet/ 目录下的项目组织结构如下：

从 Send.csproj 可以看出工程基线：TargetFramework为net8.0，OutputType为Exe（控制台应用），并启用了ImplicitUsings与Nullable；核心依赖为RabbitMQ.Client7.2.1（同一大版本内的客户端 API 均基于 async/await）。全部 13 个项目统一登记在 dotnet.sln 中，因此也可以直接用 Visual Studio 2022 打开解决方案，以 F5 方式分别启动生产者和消费者项目。

教程一：Hello World——最小可运行的消息收发

原文档给出的运行方式是在两个终端分别启动接收端与发送端：

# terminal tab 1 dotnet run --project Receive/Receive.csproj # terminal tab 2 dotnet run --project Send/Send.csproj

（以上命令均在dotnet/目录根下执行；下文所有命令同理。）

先启动 Receive 会打印[*] Waiting for messages.，随后 Send 发送一条Hello World!消息，Receive 输出[x] Received Hello World!，最后按[Enter]退出。查看源码，发送端 Send.cs 的核心调用链为：

var factory = new ConnectionFactory { HostName = "localhost" }; using var connection = await factory.CreateConnectionAsync(); using var channel = await connection.CreateChannelAsync(); await channel.QueueDeclareAsync(queue: "hello", durable: true, exclusive: false, autoDelete: false, arguments: new Dictionary<string, object?> { { "x-queue-type", "quorum" } }); const string message = "Hello World!"; var body = Encoding.UTF8.GetBytes(message); await channel.BasicPublishAsync(exchange: string.Empty, routingKey: "hello", body: body);

接收端 Receive.cs 则通过AsyncEventingBasicConsumer订阅队列，并在ReceivedAsync回调中打印消息体，最后以BasicConsumeAsync("hello", autoAck: true, consumer: consumer)消费。

值得注意的源码级细节有两个：

• 全程异步 API：7.2.x 客户端把连接、建 channel、声明队列、发布、消费全部升级为xxxAsync方法（CreateConnectionAsync、CreateChannelAsync、QueueDeclareAsync、BasicPublishAsync、BasicConsumeAsync），这是与旧版IConnection/IModel同步 API 的最大差异。
• quorum 队列：声明队列时通过arguments传入"x-queue-type": "quorum"且durable: true。quorum 队列是 RabbitMQ 基于 Raft 的复制队列类型，相比经典队列提供更强的数据安全与故障恢复能力，这正是教程一展示的"即使只发一条消息，也要保证持久化"的官方推荐做法。

教程二：工作队列——任务分发、公平调度与消息确认

原文档在三个终端中模拟"多个 Worker 抢任务、一个 NewTask 批量发任务"的场景：

# terminal tab 1 dotnet run --project Worker/Worker.csproj # terminal tab 2 dotnet run --project Worker/Worker.csproj # terminal tab 3 dotnet run --project NewTask/NewTask.csproj "First Message" dotnet run --project NewTask/NewTask.csproj "Second Message" dotnet run --project NewTask/NewTask.csproj "Third Message" dotnet run --project NewTask/NewTask.csproj "Fourth Message" dotnet run --project NewTask/NewTask.csproj "Fifth Message"

任务内容以参数形式传入，例如"First Message"；若要模拟耗时任务，可在消息中加入点号.——Worker.cs 中int dots = message.Split('.').Length - 1; await Task.Delay(dots * 1000);会按点号数量模拟秒级耗时，方便观察负载均衡。

生产者 NewTask.cs 的关键实现：

await channel.QueueDeclareAsync(queue: "task_queue", durable: true, exclusive: false, autoDelete: false, arguments: new Dictionary<string, object?> { { "x-queue-type", "quorum" } }); var properties = new BasicProperties { Persistent = true }; await channel.BasicPublishAsync(exchange: string.Empty, routingKey: "task_queue", mandatory: true, basicProperties: properties, body: body);

这里Persistent = true将消息标记为持久化（配合持久队列，保证 RabbitMQ 重启后任务不丢），mandatory: true则要求消息无法路由时返回给生产者而非静默丢弃。

消费者 Worker.cs 演示了工作队列的三大要点：

• 公平调度（fair dispatch）：BasicQosAsync(prefetchSize: 0, prefetchCount: 1, global: false)将每个消费者的预取数限制为 1，避免"手快的消费者把队列掏空"，让 RabbitMQ 在消费者处理完并确认后才派发下一条；
• 手动确认：BasicConsumeAsync("task_queue", autoAck: false, ...)关闭自动确认，处理完成后调用BasicAckAsync(deliveryTag: ea.DeliveryTag, multiple: false)显式确认；
• 消息重投：未确认的消息在消费者断开时会重新入队，交给其他 Worker，这正是任务不丢失的保障。

教程三：发布/订阅——fanout 交换机与临时匿名队列

原文档演示了一个日志广播场景：两个订阅者同时接收，一个发布者广播：

# terminal tab 1 dotnet run --project ReceiveLogs/ReceiveLogs.csproj # terminal tab 2 dotnet run --project ReceiveLogs/ReceiveLogs.csproj # terminal tab 3 dotnet run --project EmitLog/EmitLog.csproj

发布端 EmitLog.cs 声明ExchangeType.Fanout类型的logs交换机，并把消息发布到该交换机（routingKey为空字符串）：

await channel.ExchangeDeclareAsync(exchange: "logs", type: ExchangeType.Fanout); var message = GetMessage(args); var body = Encoding.UTF8.GetBytes(message); await channel.BasicPublishAsync(exchange: "logs", routingKey: string.Empty, body: body);

接收端 ReceiveLogs.cs 展示了 fanout 订阅的标准三连：

1. 同样声明logs交换机；
2. 调用无参channel.QueueDeclareAsync()让服务端生成一个随机临时队列名，并通过返回的QueueDeclareOk.QueueName获取（队列名类似amq.gen-...，进程退出即自动删除）；
3. 用QueueBindAsync(queue: queueName, exchange: "logs", routingKey: string.Empty)将匿名队列绑定到 fanout 交换机。

由于 fanout 交换机忽略 routing key 向所有绑定队列广播，因此启动多少个 ReceiveLogs，每个都能收到同一条日志——这正是"发布/订阅"区别于工作队列"竞争消费"的核心。

教程四：路由——direct 交换机与按严重级别过滤

原文档以info / warning / error三种严重级别演示按需订阅：

# terminal tab 1 dotnet run --project ReceiveLogsDirect/ReceiveLogsDirect.csproj warning error # terminal tab 2 dotnet run --project ReceiveLogsDirect/ReceiveLogsDirect.csproj info warning error # terminal tab 3 dotnet run --project EmitLogDirect/EmitLogDirect.csproj info "Run. Run. Or it will explode." dotnet run --project EmitLogDirect/EmitLogDirect.csproj warning "Run. Run. Or it will explode." dotnet run --project EmitLogDirect/EmitLogDirect.csproj error "Run. Run. Or it will explode."

命令中第一个参数是 routing key（严重级别），其余参数拼成消息体。终端 1 只关心warning和error，终端 2 关心全部三种级别。

生产者 EmitLogDirect.cs 声明ExchangeType.Direct的direct_logs交换机，并把severity作为 routing key 发布；接收端 ReceiveLogsDirect.cs 的核心是按需建立多条绑定：

await channel.ExchangeDeclareAsync(exchange: "direct_logs", type: ExchangeType.Direct); var queueDeclareResult = await channel.QueueDeclareAsync(); string queueName = queueDeclareResult.QueueName; foreach (string? severity in args) { await channel.QueueBindAsync(queue: queueName, exchange: "direct_logs", routingKey: severity); }

direct 交换机要求消息的 routing key 与绑定 key完全相等才投递。注意该程序开头有参数校验（ReceiveLogsDirect.cs）：无参数时向标准错误输出Usage: ... [info] [warning] [error]并退出，这是值得在生产脚本中借鉴的参数防御写法。

教程五：主题——topic 交换机与通配符绑定

原文档演示了 topic 交换机最灵活的绑定语法，支持*（匹配一个单词）与#（匹配零个或多个单词）：

# terminal tab 1 # To receive all the logs: dotnet run --project ReceiveLogsTopic/ReceiveLogsTopic.csproj "#" # To receive all logs from the facility "kern": dotnet run --project ReceiveLogsTopic/ReceiveLogsTopic.csproj "kern.*" # Or if you want to hear only about "critical" logs: dotnet run --project ReceiveLogsTopic/ReceiveLogsTopic.csproj "*.critical" # You can create multiple bindings: dotnet run --project ReceiveLogsTopic/ReceiveLogsTopic.csproj "kern.*" "*.critical" # terminal tab 2 # And to emit a log with a routing key "kern.critical" type: dotnet run --project EmitLogTopic/EmitLogTopic.csproj kern.critical "A critical kernel error"

绑定语义与官方教程一致："#"接收全部日志，"kern.*"只收kern设施的所有级别，"*.critical"只收所有设施的 critical 级别，多个绑定 key 可以一次传入实现并集订阅。

生产者 EmitLogTopic.cs 声明ExchangeType.Topic的topic_logs交换机，默认 routing key 为anonymous.info；接收端 ReceiveLogsTopic.cs 与 direct 版结构一致，只是把direct_logs换成topic_logs，并同样用foreach为每个绑定 key 调用QueueBindAsync。消费回调中通过ea.RoutingKey打印实际命中消息的 routing key（如kern.critical），便于直观理解通配符匹配结果。例如在上述命令组合下，kern.critical "A critical kernel error"会被终端 1、终端 2（kern.*）、终端 3（*.critical）、终端 4（双绑定）同时收到。

教程六：RPC——回调队列、CorrelationId 与异步响应映射

原文档按"先起服务、再发请求"的顺序运行：

# terminal tab 1 # Our RPC service is now ready. We can start the server: dotnet run --project RPCServer/RPCServer.csproj # terminal tab 2 # To request a fibonacci number run the client: dotnet run --project RPCClient/RPCClient.csproj

服务端 RPCServer.cs 实现了经典 RPC 模式：声明持久化 quorum 队列rpc_queue（同样带"x-queue-type": "quorum"参数），BasicQosAsync(prefetchCount: 1)保证一次只处理一个请求；收到请求后构造BasicProperties { CorrelationId = props.CorrelationId }作为响应属性，将斐波那契计算结果Fib(n)发布回请求方指定的props.ReplyTo队列，最后BasicAckAsync确认请求消息。其中的Fib函数（RPCServer.cs）只接受合法正整数输入。

客户端 RPCClient.cs 的异步响应映射是全示例中最精巧的部分：

• StartAsync中声明服务端命名回调队列_replyQueueName，并注册消费者持续监听该队列；
• CallAsync中为每次请求生成唯一Guid作为CorrelationId，设置ReplyTo = _replyQueueName，同时把correlationId与TaskCompletionSource<string>的配对存入线程安全的ConcurrentDictionary回调映射表；
• 消费者回调收到响应时，按ea.BasicProperties.CorrelationId从映射表中取出对应的tcs，调用TrySetResult(response)唤醒等待中的调用方；
• 支持CancellationToken：取消时从映射表移除并SetCanceled()，DisposeAsync中关闭 channel 与 connection。

Main默认请求fib(30)（也可传入参数），服务端逐个打印[.] Fib(n)，客户端最终输出[.] Got '832040'。

教程七：发布者确认——三种吞吐策略的对比实现

原文档只给出一条运行命令：

# terminal tab 1 dotnet run --project PublisherConfirms/PublisherConfirms.csproj

该程序（PublisherConfirms.cs）是七套教程中逻辑最丰富的一个，它内置了三个参数常量并顺序执行三种确认策略：

• MAX_OUTSTANDING_CONFIRMS = 256：限制在途未确认消息数；
• MESSAGE_COUNT = 50_000：每个策略发布的消息总数；
• 可选命令行参数：dotnet run --project PublisherConfirms/PublisherConfirms.csproj <hostname>可指定非 localhost 的 RabbitMQ 地址。

三种策略在 PublisherConfirms.cs 中被依次调用：

1. 逐条确认（PublishMessagesIndividuallyAsync）：每次BasicPublishAsync后等待该消息的确认结果，最慢但语义最简单，异常（nack 或 basic.return）打印到标准错误；
2. 批量确认（PublishMessagesInBatchAsync）：batchSize = MAX_OUTSTANDING_CONFIRMS / 2，累积ValueTask列表达到批大小后统一await，兼顾吞吐与错误可见性；
3. 异步确认（HandlePublishConfirmsAsynchronously）：吞吐最高，通过 channel 事件驱动处理确认——channel.BasicAcksAsync/BasicNacksAsync处理确认与否定确认，BasicReturnAsync处理不可路由消息（每 1000 条故意用一个随机 routing key 触发basic.return，见 PublisherConfirms.cs），并用LinkedList<ulong>维护在途序号、GetNextPublishSequenceNumberAsync()获取发布序号、SemaphoreSlim保护并发访问。

启用确认通道的关键在 PublisherConfirms.cs：

var channelOpts = new CreateChannelOptions( publisherConfirmationsEnabled: true, publisherConfirmationTrackingEnabled: true, outstandingPublisherConfirmationsRateLimiter: new ThrottlingRateLimiter(MAX_OUTSTANDING_CONFIRMS) );

其中publisherConfirmationsEnabled开启发布者确认模式（channel 进入 confirm 模式），publisherConfirmationTrackingEnabled让客户端内部跟踪确认，ThrottlingRateLimiter(256)对在途确认数量做限流；异步策略中该选项被显式关闭（PublisherConfirms.cs），改由程序自行跟踪。程序最后会用Stopwatch打印每种策略发布 50,000 条消息的耗时，让你在同一环境下直接对比三种确认策略的吞吐差异。

非交互终端下的运行注意事项

仓库 dotnet/AGENTS.md 记录了一个自动化场景的坑：Receive、Worker、ReceiveLogs、ReceiveLogsDirect、ReceiveLogsTopic、RPCServer 这些消费程序都用Console.ReadLine()阻塞等待[Enter]退出。当标准输入不是 TTY（例如 CI、后台脚本）时，ReadLine()会立即返回null，进程在收到任何消息前就退出了。解决方式是保持 stdin 打开——用一个阻塞进程管道喂给程序：

(while true; do sleep 20; done) | dotnet run --project Receive & (while true; do sleep 20; done) | dotnet run --project ReceiveLogs &

同时该文档强调：交互场景下Console.ReadLine()与 "Press [enter] to exit." 提示是刻意保留的，不应作为"解决自动化问题"的手段删除。

小结：从示例到生产实践的要点提炼

通读这七套 C# 教程及其源码，可以提炼出几条直接可迁移到生产项目的实践结论：

• 优先使用异步 API 与 quorum 队列：7.2.x 客户端全面 async 化，配合"x-queue-type": "quorum"+durable: true的队列声明（如 Send.cs、NewTask.cs、RPCServer.cs），在保证消息可靠性的同时获得复制队列的容灾能力；
• 消息可靠性三件套：生产者Persistent = true+mandatory: true（NewTask.cs），消费者prefetchCount: 1+ 手动 ack（Worker.cs），必要时叠加发布者确认（PublisherConfirms.cs）；
• 交换机类型决定路由语义：fanout 广播、direct 精确匹配、topic 通配符匹配，配合服务端匿名队列可实现灵活的临时订阅（ReceiveLogs.cs）；
• RPC 模式四要素：请求队列、回调队列、CorrelationId、ReplyTo属性，缺一不可（RPCClient.cs）；
• 按吞吐需求选择确认策略：逐条确认最稳、批量确认折中、异步事件驱动确认吞吐最高（PublisherConfirms.cs）。

如果想深入理解每个例子的运行细节，可以对照 dotnet/README.md 逐条执行上述命令，并随时查阅 dotnet 目录下对应项目的源码；整个仓库的多语言移植版本（Python、Java、Go、Ruby 等）见根目录 README.md。

• 示例工程