示例工程【免费下载链接】rabbitmq-tutorialsTutorials 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唯一的前置开发依赖是 .NETdotnet 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.0OutputType为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 Dictionarystring, 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)消费。值得注意的源码级细节有两个全程异步 API7.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 Dictionarystring, 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 dispatchBasicQosAsync(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 订阅的标准三连同样声明logs交换机调用无参channel.QueueDeclareAsync()让服务端生成一个随机临时队列名并通过返回的QueueDeclareOk.QueueName获取队列名类似amq.gen-...进程退出即自动删除用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、终端 2kern.*、终端 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与TaskCompletionSourcestring的配对存入线程安全的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 中被依次调用逐条确认PublishMessagesIndividuallyAsync每次BasicPublishAsync后等待该消息的确认结果最慢但语义最简单异常nack 或 basic.return打印到标准错误批量确认PublishMessagesInBatchAsyncbatchSize MAX_OUTSTANDING_CONFIRMS / 2累积ValueTask列表达到批大小后统一await兼顾吞吐与错误可见性异步确认HandlePublishConfirmsAsynchronously吞吐最高通过 channel 事件驱动处理确认——channel.BasicAcksAsync/BasicNacksAsync处理确认与否定确认BasicReturnAsync处理不可路由消息每 1000 条故意用一个随机 routing key 触发basic.return见 PublisherConfirms.cs并用LinkedListulong维护在途序号、GetNextPublishSequenceNumberAsync()获取发布序号、SemaphoreSlim保护并发访问。启用确认通道的关键在 PublisherConfirms.csvar 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: quorumdurable: true的队列声明如 Send.cs、NewTask.cs、RPCServer.cs在保证消息可靠性的同时获得复制队列的容灾能力消息可靠性三件套生产者Persistent truemandatory: trueNewTask.cs消费者prefetchCount: 1 手动 ackWorker.cs必要时叠加发布者确认PublisherConfirms.cs交换机类型决定路由语义fanout 广播、direct 精确匹配、topic 通配符匹配配合服务端匿名队列可实现灵活的临时订阅ReceiveLogs.csRPC 模式四要素请求队列、回调队列、CorrelationId、ReplyTo属性缺一不可RPCClient.cs按吞吐需求选择确认策略逐条确认最稳、批量确认折中、异步事件驱动确认吞吐最高PublisherConfirms.cs。如果想深入理解每个例子的运行细节可以对照 dotnet/README.md 逐条执行上述命令并随时查阅 dotnet 目录下对应项目的源码整个仓库的多语言移植版本Python、Java、Go、Ruby 等见根目录 README.md。赞分享示例工程【免费下载链接】rabbitmq-tutorialsTutorials for using RabbitMQ in various ways项目地址https://gitcode.com/gh_mirrors/ra/rabbitmq-tutorials点击查看免费下载相关推荐PDF.js 官方 Examples 指南从 Hello World 到完整页面渲染的实战教程PDF.js 官方 Examples 指南从 Hello World 到完整页面渲染的实战教程 本篇技术指南以 PDF.js 官方文档 Examples 篇前端ReScript 官方项目模板入门从 npm install 到 Hello World 的完整实战指南ReScript 官方项目模板入门从 npm install 到 Hello World 的完整实战指南 本指南以 ReScript 官方 starter 模编译器编程语言开发工具OpenUSD 官方教程体系全解环境配置、教程地图与 Hello World 完整实战OpenUSD 官方教程体系全解环境配置、教程地图与 Hello World 完整实战 本文为 OpenUSD 官方教程入口文档 docs/tut_usd_图形学3D渲染上一篇5分钟快速上手Umi-OCR免费离线文字识别终极指南下一篇OBS Studio日志系统完全指南从新手到专家的故障排查手册创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考