☰
RabbitMQ 官方教程 C / .NET 8 实战指南:从 Hello World 到 Publisher Confirms 的完整落地
2026/10/15 2:43:15 网站建设 项目流程
  • 示例工程

【免费下载链接】rabbitmq-tutorials

Tutorials for using RabbitMQ in various ways

项目地址:https://gitcode.com/gh_mirrors/ra/rabbitmq-tutorials
点击查看免费下载

本指南基于当前仓库 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、Receive教程一:Hello World生产者 / 消费者
NewTask、Worker教程二:工作队列任务发布者 / 多个工作者
EmitLog、ReceiveLogs教程三:发布/订阅日志生产者 / 日志消费者
EmitLogDirect、ReceiveLogsDirect教程四:路由带严重级别路由的日志
EmitLogTopic、ReceiveLogsTopic教程五:主题通配符绑定日志
RPCServer、RPCClient教程六:RPC斐波那契计算服务
PublisherConfirms教程七:发布者确认三种确认策略基准

从 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。

  • 示例工程

【免费下载链接】rabbitmq-tutorials

Tutorials for using RabbitMQ in various ways

项目地址:https://gitcode.com/gh_mirrors/ra/rabbitmq-tutorials
点击查看免费下载
上一篇:5分钟快速上手Umi-OCR:免费离线文字识别终极指南
下一篇:OBS Studio日志系统完全指南:从新手到专家的故障排查手册

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询