简介这套 ActiveMQ DemoC#面向希望在 .NET 桌面应用中接入消息队列的开发者重点演示在 WinForm 环境下实现 ActiveMQ 消息发送与接收的完整链路适合具备一定 C# 基础、需要快速理解生产者/消费者模型及客户端调用方式的学习者参考。压缩包共 36 个文件约 326KB以 19 个 cs 源码文件为主配合 4 个 resx 界面资源、4 个 dll 依赖库、3 个 exe 可执行文件以及 2 个 csproj 项目文件和 1 个 sln 解决方案工程结构相对紧凑。内容包含生产者与消费者两个演示项目覆盖消息发送、接收处理及界面交互等关键环节可直接运行观察消息流转也可作为二次开发时的代码参照。已有 1293 人学习下载便于对照现有实现梳理 ActiveMQ 在 WinForm 项目中的集成思路与常见配置要点。1. 消息发出去却查不到ActiveMQ 的 C# Demo 先要把哪三件事固定下来拿到一份 ActiveMQ C# Demo先别急着运行 Hello World更该确认这份 Demo 到底验证了三件事中的哪几件连接断了能不能自己接上、消息丢了能不能发现、消费端确认是不是落在了自动确认的坑里。ActiveMQ 在 .NET 侧走的是 NMS 接口写法像 JMS 又处处不一样第一次接的人十有八九卡在 NuGet 包选错、连接串少个协议前缀或者connection.Start()忘了调。这份 Demo 的价值就在骨架——连接、会话、生产者、消费者四层外加投递模式、确认模式、重投递这几组参数全摆明面上适合正在做后台任务异步化、订单状态流转、日志集中转发的开发者照着改比从零翻文档快得多。2. 环境与最小链路从 NuGet 选包到一行连接串2.1 三个包的分工选错就白折腾NMS 不是一个包是三层接口层定义IConnection、ISession、IMessageProducer这些抽象具体实现包决定你走哪种线上协议。选型错了代码编译能过连上去就是握手失败。NuGet 包作用适用场景Apache.NMS纯接口不含实现只被引用不单独安装Apache.NMS.ActiveMQOpenWire 协议实现对接 ActiveMQ 经典版端口 61616Apache.NMS.AMQPAMQP 1.0 实现对接支持 AMQP 的 Broker端口 5672判断方法很简单Broker 控制台里看到的是 OpenWire 连接器就用Apache.NMS.ActiveMQ对方的接入点写的是 AMQP 1.0才换Apache.NMS.AMQP。两者 API 长得像但连接串前缀、参数名、甚至消息属性的大小写都不一样混用会出现「连接建好了、会话也有、就是发不出去」这种查不出原因的故障。我一般先把实现包钉死在项目文件里业务代码只using Apache.NMS;将来换协议只改工厂这一处。安装时注意框架目标.NET Framework 项目和 .NET 新版本项目对依赖的引用方式不同混装容易出现同一份程序集两个版本的情况症状是运行时抛类型加载异常而不是编译错误。遇到这种先清bin/obj再统一包版本。2.2 最小生产者连接串里每个参数都要说得清using Apache.NMS; using Apache.NMS.ActiveMQ; using System; // activemq: 是协议前缀后面才是 OpenWire 的 tcp 地址少写前缀会按默认协议猜 const string brokerUri activemq:tcp://127.0.0.1:61616?nms.PrefetchPolicy.queuePrefetch10; IConnectionFactory factory new NMSConnectionFactory(brokerUri); // 账号口令对应 Broker 的 users.properties本地环境常见是 admin/admin using IConnection connection factory.CreateConnection(admin, admin); connection.Start(); // 不调 Start生产者和消费者都不会真正开工 // 会话在这里定死确认模式建完之后改不了 using ISession session connection.CreateSession(AcknowledgementMode.ClientAcknowledge); IDestination queue session.GetQueue(demo.order.queue); using IMessageProducer producer session.CreateProducer(queue); producer.DeliveryMode MsgDeliveryMode.Persistent; // 落盘Broker 重启不丢 producer.Priority MsgPriority.Normal; // 0 到 94 是默认值 producer.TimeToLive TimeSpan.FromMinutes(10); // 超时未消费会被 Broker 清掉 ITextMessage msg producer.CreateTextMessage({\orderId\:1001,\amount\:99.5}); msg.Properties[orderType] PAY; // 供选择器在 Broker 端过滤 producer.Send(msg);这段代码有三个必须理解的参数。activemq:前缀决定用哪个实现包解析地址写成裸tcp://时部分版本会退回默认协议连不上但报错很含糊。connection.Start()是显式动作NMS 允许你先建消费者再启动连接顺序错了就是「消费者建好了但收不到」。MsgDeliveryMode.Persistent意味着消息先落 Broker 的存储再返回吞吐会掉一截但换来的是 Broker 重启不丢日志、埋点这类可丢数据用NonPersistent能明显快。TimeToLive和Priority是写在消息上的但 Broker 也参与过期消息由 Broker 定时扫描清理不会推给消费者优先级需要 Broker 端开启优先级队列才真正生效否则只是记录在消息头里。这一点常被忽略——本地测试看着没问题上生产才发现高优先级消息还是排在后面。还有个小细节CreateTextMessage这个方法在不同 NMS 版本上挂在IMessageProducer或ISession编译不过时换另一个对象调即可功能一致。我第一次升级包版本时被这个签名变更卡了半天后来养成习惯——消息对象的创建统一走一个私有方法包一层换版本只改一处。2.3 消费者同步 Receive 与异步 Listener 的边界using Apache.NMS; using Apache.NMS.ActiveMQ; using System; IConnectionFactory factory new NMSConnectionFactory( activemq:tcp://127.0.0.1:61616?nms.PrefetchPolicy.queuePrefetch1); using IConnection connection factory.CreateConnection(admin, admin); connection.Start(); using ISession session connection.CreateSession(AcknowledgementMode.ClientAcknowledge); IDestination queue session.GetQueue(demo.order.queue); // 第二个参数是选择器语法接近 SQL-92过滤发生在 Broker 端 using IMessageConsumer consumer session.CreateConsumer(queue, orderType PAY); consumer.Listener message { try { if (message is ITextMessage text) { Console.WriteLine($[{message.NMSMessageId}] {text.Text}); } // ClientAcknowledge 下这行会确认当前会话里此前收到的所有消息 message.Acknowledge(); } catch (Exception ex) { // 不确认连接关闭或会话 Recover 后 Broker 会重投 Console.WriteLine($处理失败留给重投递: {ex.Message}); } }; Console.ReadLine();Listener是异步模式回调跑在库内部的线程上主线程必须阻塞住否则进程退出连接就断了。Receive()是同步阻塞适合批处理脚本这类「拿一条处理一条就结束」的场景线上常驻服务基本都用 Listener。回调里做两件事的顺序不能反先处理业务再确认。确认放在前面业务抛异常消息也回不去了这是最常见的数据丢失来源。另外ClientAcknowledge的语义是「确认本会话到目前为止消费的所有消息」不是单条确认所以哪怕一次只处理一条也要保证处理完再调。nms.PrefetchPolicy.queuePrefetch1这个参数值得单独说。默认预取值较大消费者一次会从 Broker 拉一批到本地缓冲。消费者进程崩了缓冲区里那批消息要等连接超时被 Broker 判定失效才会重投中间这段时间消息处于「已投递未确认」状态看着像丢了。把预取压到 1牺牲的是吞吐换的是故障时消息回队更快。批量处理场景可以放到 20 到 50单条耗时长、失败率高的业务就老实设 1。3. 队列还是主题两种投递语义的代码与参数对照3.1 Queue 的点对点与 Topic 的广播session.GetQueue(x)和session.GetTopic(x)决定了消息的分发方式队列是竞争消费一条消息只给一个消费者主题是广播每个订阅者各拿一份。// 队列三个消费者分摊每条消息只被其中一个处理 IDestination queue session.GetQueue(demo.task.queue); IMessageConsumer worker session.CreateConsumer(queue); // 主题先订阅再生产否则这条消息对这个订阅者不存在 IDestination topic session.GetTopic(demo.event.topic); IMessageConsumer sub session.CreateConsumer(topic);主题有个容易翻车的点非持久订阅者只在连接存活期间存在Broker 不为你保留离线消息。消费者重启的那几秒里发的事件全部丢失且没有任何提示。要让离线期间的消息也补上得建持久订阅// clientId 必须在连接级别唯一同一 clientId 重复连接会把前一个顶掉 connection.ClientId svc-order-01; IDestination topic session.GetTopic(demo.event.topic); IMessageConsumer durable session.CreateDurableConsumer( topic, order-subscriber, orderType PAY, false);CreateDurableConsumer的第二个参数是订阅名Broker 按clientId 订阅名记录消费进度。第三个参数是选择器只保留匹配的消息第四个参数noLocal传 true 表示不接收本连接自己发的消息避免自产自消打转。维度QueueTopic分发方式一条消息一个消费者一条消息广播给所有订阅者离线消息默认保留仅持久订阅保留典型场景任务派发、订单处理配置变更通知、缓存失效广播扩容影响加消费者提升吞吐加订阅者增加扇出成本常见误用拿队列做广播结果只有一个服务收到拿主题做任务派发同一任务被重复执行最后一行是真事故高发区。有人用主题做任务派发本地起了两个实例做压测结果每个任务被执行两次还以为是 Broker 重复投递。判断标准很简单这条消息的语义是「谁处理都行」还是「所有人都得知道」前者队列后者主题。3.2 持久化、优先级、TTL 三件套一起设参数单独看都简单组合起来才有边界。生产者端三个属性决定消息的「生死」producer.DeliveryMode MsgDeliveryMode.Persistent; // 与 NonPersistent 的区别是是否落盘 producer.Priority MsgPriority.High; // 需 Broker 端启用优先级才生效 producer.TimeToLive TimeSpan.FromMinutes(30); // 从进 Broker 开始计时 // 也可以逐条覆盖比如补偿任务的消息给更长的存活时间 producer.Send(msg, MsgDeliveryMode.Persistent, MsgPriority.High, TimeSpan.FromHours(2));Send的重载允许单条覆盖生产者默认值这在混合业务里很有用主流程消息给 10 分钟 TTL走重试通道的补偿消息给 2 小时。但要注意 TTL 是从消息进入 Broker 开始算的不是从被消费开始算如果消息在队列里压了 40 分钟才被取走10 分钟 TTL 的那条早就被清理了消费者根本看不到。优先级需要 Broker 端配置支持。开启之后 Broker 会按优先级维护多个内部队列高优先级先出没开启时优先级只写在消息头里出队顺序仍是先进先出。这个差异在本地单机测试时几乎看不出来因为消息一进就出了队列里没有积压。我的验证办法是故意停掉消费者灌 200 条不同优先级的消息再启消费者看前 20 条的顺序——一次就能确认配置到底生效没有。持久化也不是没有成本的。开启后每条消息都要等存储写完才返回磁盘慢的时候发送端延迟会明显上升。日志、埋点、指标上报这类丢几条无所谓的场景用NonPersistent吞吐能翻几倍只有订单、支付、账务这种必须一条不丢的才用Persistent。判断标准不是「重要不重要」而是「丢了能不能靠别的手段补回来」。3.3 消息选择器把过滤放到 Broker 端两种过滤方式的差别很大// 方式一Broker 端过滤不匹配的消息不会下发 IMessageConsumer filtered session.CreateConsumer(queue, orderType PAY AND amount 100);只建一个消费者、收全量、在代码里if判断——这是最常见的写法也是最容易把 Broker 内存打满的写法。预取策略是按消费者算的一个消费者会一次拉一大批全量消息到本地其中大部分被if丢掉占用内存还阻塞了其它消费者拿到这些消息。选择器支持, , , , AND, OR, NOT, BETWEEN, IN, LIKE, IS NULL这些字段可以是消息属性或NMSMessageId、NMSPriority这类系统头。写法上三个坑字符串必须用单引号用双引号在某些版本上会被当成标识符解析直接连接报错。属性名大小写敏感ordertype和orderType是两个不同的键。浮点数比较精度不可靠金额建议用整数分存储再比较。提示选择器表达式在 Broker 端解析语法错误通常在创建消费者时就抛出而不是等消息来。上线前先用控制台手工发几条带属性的消息确认消费者能收到比在代码里猜快得多。4. 可靠性三条线确认模式、事务、重投递4.1 四种确认模式选错就是丢消息模式确认时机适用场景风险AutoAcknowledge回调返回后自动确认处理逻辑不会失败的通知类消息回调里抛异常也视为已确认ClientAcknowledge显式调用 Acknowledge绝大多数业务消息漏调 Acknowledge 会一直重投DupsOkAcknowledge延迟批量确认高吞吐、可容忍重复故障时可能重复投递TransactionalCommit 时确认多条消息或消息加数据库同批忘记 Commit 消息一直挂着AutoAcknowledge是最容易被误用的一个。看名字像「安全」实际语义是「消费者拿到就算送到了」。回调里数据库连接超时、反序列化失败、业务校验不过这些异常在库看来都是「回调正常结束」消息照确认不误然后永久消失。有过一次线上对账少数据翻了半天日志才定位到消费端的异常被外层 try 吞掉了消息已经确认没有后悔药。Transactional模式下消费者必须显式Commit()才算确认Rollback()则让消息回队。它的额外好处是能把「处理消息」和「写业务库」放进同一个本地事务边界里消息不是数据库事务的一部分但代码上的提交顺序可控先处理业务、再提交数据库、最后Commit会话中间任何一步失败就Rollback让消息重来。4.2 本地事务批量边界怎么划using ISession session connection.CreateSession(AcknowledgementMode.Transactional); IDestination queue session.GetQueue(demo.order.queue); using IMessageProducer producer session.CreateProducer(queue); try { // 批量发一批一次网络往返 for (int i 0; i 100; i) { ITextMessage m producer.CreateTextMessage(${{\seq\:{i}}}); m.Properties[batchId] B20240101; producer.Send(m); } session.Commit(); // 100 条一起落地之前一条都对外不可见 } catch (Exception ex) { session.Rollback(); // 全部丢弃重来 Console.WriteLine($批量失败: {ex.Message}); }生产端事务的收益是把 100 次网络往返压成 1 次提交吞吐提升明显。代价是Commit之前这 100 条对消费者完全不可见攒批越大延迟越高。批量大小我一般按「单批耗时不超 200 毫秒」倒推几百到几千条视消息体大小而定。消费端事务的写法不同注意别把两者混在一个会话里// 消费端处理成功才 Commit失败 Rollback 让消息重投 consumer.Listener message { try { Handle(message); // 业务处理可能写库 session.Commit(); // 确认 } catch { session.Rollback(); // 消息回队RedeliveryCount 加一 } };这里有个必须守住的纪律Commit一定放在业务之后且业务要幂等。因为Rollback之后消息会重新投递如果前一次处理其实已经部分生效比如发了外部通知但写库失败重投就会重复通知。4.3 重投递策略与死信队列默认情况下一条消息消费失败会被反复重投RedeliveryCount 一直涨直到超过上限才被送进死信队列。var cf new Apache.NMS.ActiveMQ.ConnectionFactory( activemq:tcp://127.0.0.1:61616); cf.RedeliveryPolicy new RedeliveryPolicy { MaximumRedeliveries 3, // 最多重投 3 次 InitialRedeliveryDelay TimeSpan.FromSeconds(1),// 首次延迟 RedeliveryDelay TimeSpan.FromSeconds(5),// 后续基础延迟 UseExponentialBackOff true, // 指数退避 BackOffMultiplier 2.0 // 1s, 2s, 4s }; IConnection connection cf.CreateConnection(admin, admin);指数退避的意义在于区分「瞬时故障」和「永久故障」。数据库连接抖动几百毫秒就恢复退避一两次就过了消息体格式错误这种无论重投多少次都失败的情况快速耗尽重投次数进死信队列比卡在队列里反复重试更健康。重投次数耗尽后消息默认进入ActiveMQ.DLQ可以在控制台里看到。这里的关键是有人看——死信队列不是垃圾桶是需要监控的告警源。我一般会写一个独立的监控消费者挂着这个队列一有消息进来就发告警而不是等业务方反馈数据不对再去翻。提示客户端重投递和 Broker 端的重投递插件是两套机制同时开启会叠加。排查「消息怎么重投了 12 次」这类问题时先确认两边各配了什么。5. 避坑与排查五条踩出来的记录5.1 现象控制台能连代码一直报连接失败原因通常有两个。一是连接串少了activemq:前缀工厂按默认协议去解析端口对不上二是包选错装的是Apache.NMS.AMQP却连 OpenWire 端口 61616握手阶段就被拒。排查顺序是先确认 Broker 上监听的是哪种协议控制台连接器页面能看到再核对连接串前缀和 NuGet 包是否一致。别在代码里加各种超时参数试方向错了怎么调都没用。5.2 现象消费者启动后收了几条就再也不动了先看线程。Listener回调跑在库的接收线程上回调里如果做了长耗时操作同步 HTTP 调用、大批量数据库写入接收线程被占住后续消息就堆在本地缓冲里下不来。表现是 CPU 不高、连接还在、就是不消费。解决办法是把业务处理丢到自己的线程池里回调只负责取消息和确认但要注意这时确认和处理的顺序得自己保证不能回调一进来就Acknowledge。另一种可能是预取值太大加上消费者崩溃消息卡在「已投递未确认」状态等超时压小queuePrefetch能缩短这个窗口。5.3 现象Broker 重启后消费者还在但收不到新消息failover:连接串没配或者配了但参数不对。单机tcp://连接在 Broker 重启后就是断了NMS 不会自动重连进程还活着日志里也不吭声。改成故障转移格式// randomizefalse 让客户端按顺序尝试避免每次都撞到挂掉的那个 const string uri failover:(tcp://10.0.0.11:61616,tcp://10.0.0.12:61616) ?randomizefalseinitialReconnectDelay1000maxReconnectDelay30000;另一个隐蔽原因是先建消费者后Start()。连接没启动时创建消费者部分版本上消费者处于未就绪状态启动后也不会补订阅。养成「建连接 → Start → 建会话 → 建消费者」的固定顺序能省掉这类玄学问题。5.4 现象消息发出去控制台能看到消费端就是匹配不上选择器写错的概率最高。属性名大小写、字符串引号、数值类型三者任一不对表达式就匹配不到任何消息而消费者创建是成功的不报错。排查办法是在控制台手工浏览队列里的消息逐条看属性名和值到底是什么类型数字属性写成字符串、布尔写成true都会导致比较失败。还有一类是属性根本没设上去——msg.Properties[x] null在某些版本上会直接抛异常或忽略值一定要给确定的类型。5.5 现象消息反复重投日志里堆满同一条两个方向查。一是业务处理确实每次都失败比如反序列化字段不兼容、依赖的下游接口挂了这时候重投是对的该修的是业务二是处理成功了但确认没生效ClientAcknowledge下忘了调Acknowledge()或者事务模式下Commit()被放在了一个永远不会走到的分支里。区分方法很简单看RedeliveryCount是不是每次都从 0 开始涨、业务日志里有没有成功记录。确认没生效的情况重投间隔会稳定等于配置的退避时间很好认。6. 用请求-响应把 Demo 变成可回归的自检单向往队列里塞消息只能证明「发得出去」证明不了「链路是通的」。我习惯给每个项目的 Demo 加一段请求-响应自检跑一次就知道生产、消费、回执三段有没有断。// 服务端收到请求按 NMSReplyTo 回一条带相同关联 ID 的响应 using IMessageConsumer server session.CreateConsumer(session.GetQueue(demo.rpc.queue)); server.Listener req { IDestination replyTo req.NMSReplyTo; // 客户端指定的回执队列 using IMessageProducer replier session.CreateProducer(replyTo); ITextMessage resp replier.CreateTextMessage(OK); resp.NMSCorrelationID req.NMSCorrelationID; // 原样带回供客户端配对 replier.Send(resp); req.Acknowledge(); }; // 客户端临时队列收响应按关联 ID 匹配 using IDestination tempQueue session.CreateTemporaryQueue(); ITextMessage request producer.CreateTextMessage(PING); request.NMSReplyTo tempQueue; request.NMSCorrelationID Guid.NewGuid().ToString(N); producer.Send(request); using IMessageConsumer client session.CreateConsumer(tempQueue); IMessage reply client.Receive(TimeSpan.FromSeconds(5)); // 超时返回 null Console.WriteLine(reply null ? 自检失败5 秒无回执 : $自检通过: {reply.NMSCorrelationID});两个参数是核心NMSReplyTo告诉服务端往哪回NMSCorrelationID让客户端在一堆响应里认出自己那条。CreateTemporaryQueue生成的临时队列随连接销毁自动删除不会在 Broker 上留垃圾——这点比手工建固定回执队列干净得多。自检脚本我一般配成一组固定断言跑完看表格就行检查项通过标准常见失败原因队列可写发送无异常控制台队列数 1连接未 Start、账号无写权限队列可读5 秒内收到回执选择器不匹配、预取过大属性透传响应里的 CorrelationID 与请求一致属性名大小写、类型不匹配重连可用重启 Broker 后 30 秒内恢复未配 failover、参数拼错死信可见DLQ 有消息时监控能捕获监控消费者没开持久订阅这套断言跑一次几十秒比人工点控制台靠谱。血泪经验是任何一次 Broker 配置变更、包版本升级、连接串调整之后都强制走一遍自检别信「就改了个端口不会有影响」。我吃过一次亏——升级 NMS 包后CreateTextMessage的签名变了编译提示很弱跑自检才发现消息体是空的生产者端全部静默失败而那批消息是账务流水。从那以后我每次动连接层都先跑这五条断言再谈别的。希望帮到你。本文还有配套的精品资源点击获取