简介：这套 ActiveMQ Demo（C#）面向希望在 .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这些抽象；具体实现包决定你走哪种线上协议。选型错了，代码编译能过，连上去就是握手失败。

判断方法很简单：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.queuePrefetch=10"; 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 到 9，4 是默认值 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.queuePrefetch=1"); 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.queuePrefetch=1这个参数值得单独说。默认预取值较大，消费者一次会从 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 表示不接收本连接自己发的消息，避免自产自消打转。

最后一行是真事故高发区。有人用主题做任务派发，本地起了两个实例做压测，结果每个任务被执行两次，还以为是 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是最容易被误用的一个。看名字像「安全」，实际语义是「消费者拿到就算送到了」。回调里数据库连接超时、反序列化失败、业务校验不过，这些异常在库看来都是「回调正常结束」，消息照确认不误，然后永久消失。有过一次线上对账少数据，翻了半天日志才定位到消费端的异常被外层 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 不会自动重连，进程还活着，日志里也不吭声。改成故障转移格式：

// randomize=false 让客户端按顺序尝试，避免每次都撞到挂掉的那个 const string uri = "failover:(tcp://10.0.0.11:61616,tcp://10.0.0.12:61616)" + "?randomize=false&initialReconnectDelay=1000&maxReconnectDelay=30000";

另一个隐蔽原因是先建消费者后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 上留垃圾——这点比手工建固定回执队列干净得多。

自检脚本我一般配成一组固定断言，跑完看表格就行：

这套断言跑一次几十秒，比人工点控制台靠谱。血泪经验是：任何一次 Broker 配置变更、包版本升级、连接串调整之后，都强制走一遍自检，别信「就改了个端口不会有影响」。我吃过一次亏——升级 NMS 包后CreateTextMessage的签名变了，编译提示很弱，跑自检才发现消息体是空的，生产者端全部静默失败，而那批消息是账务流水。从那以后我每次动连接层，都先跑这五条断言再谈别的。希望帮到你。

本文还有配套的精品资源，点击获取