简介：这是一份面向 C# 开发者与消息中间件初学者的 ActiveMQ 实践示例包，围绕消息发送与接收流程展开，适合在 WinForm 桌面程序中验证队列通信，并借此理解连接工厂、会话、生产者、消费者等基础概念。压缩包共 36 个文件，约 326KB，以 19 个 cs 源码为主体，配合 resx 窗体资源、dll 依赖、exe 示例程序及 csproj、sln 等工程文件，包含生产端和消费端两个可参照项目，便于对照调试。已有 1293 人学习，读者可从中了解 ActiveMQ 客户端接入方式、消息收发封装与界面触发逻辑，并基于现成工程结构进行改写、排错和功能扩展，适合作为入门练习与小型桌面应用的原型参考。
1. 先跑通一条消息，再谈架构
某个 .NET 老系统要和另一侧基于 JMS 的服务对接，对面只肯暴露队列，不提供 HTTP 接口，团队里没人碰过 ActiveMQ，最后是从一份 ActiveMQ Demo（C#）开始把链路打通的。这份 Demo 的价值不在代码量，而在于它把 C# 侧访问 ActiveMQ 的完整路径摆了出来：连接工厂怎么建、Queue 和 Topic 分别怎么发、确认模式怎么选、资源什么时候释放。适合三类人：第一次在 .NET 里接消息队列的后端、要从 Java 侧迁移消费逻辑的开发者、以及需要给现有系统补一条异步链路但不想引入新中间件的团队。下面按“能跑起来 → 能跑对 → 不踩坑”的顺序拆。
2. 环境与工程骨架：先把 61616 这条连接坐实
很多人卡在第一天，不是代码写错，而是 broker 根本没在监听。ActiveMQ 的 C# 客户端走的是 OpenWire 协议，默认端口 61616，管理控制台在 8161，这两个端口任何一个不通，后面的代码再漂亮也没用。这一章先把服务端和客户端两条线各自确认清楚，再落到 Demo 的工程结构上，避免一边调代码一边怀疑环境。
2.1 Broker 侧：61616 与 8161 各自负责什么
拿到 ActiveMQ 发行包后，解压目录里bin/activemq start是启动脚本，conf/activemq.xml是主配置，data/activemq.log是日志。启动后先别急着写 C#，用端口和日志两件事确认服务真的起来了。
# 启动 broker（后台方式） cd /opt/apache-activemq ./bin/activemq start # 确认两个端口都在监听 ss -lntp | grep -E '61616|8161' # 看启动日志里有没有 transport connector 的绑定记录 tail -n 50 data/activemq.log第一段是启动，第二段确认监听，第三段看日志。日志里出现Listening for connections at: tcp://0.0.0.0:61616这类行才说明 OpenWire 端口真的对外开了；如果只看到tcp://127.0.0.1:61616，那本机测试没问题，容器外或另一台机器就连不上。这一点在容器化部署里特别容易翻车——镜像默认绑定回环地址，映射端口出来依然拒绝连接。管理控制台的默认账号常见是admin/admin，这只是演示环境的口令，任何对外的环境都必须改掉，配置文件在conf/jetty-realm.properties。
浏览器打开http://<host>:8161/admin，用账号登进去，Queues 页面能看到队列名和堆积数，这是后面排查“消息到底发出去没有”的第一现场。C# 侧的连接串就写tcp://<host>:61616，两个端口别混，8161 是 HTTP 控制台，客户端连它是连不上的。
2.2 客户端侧：NuGet 依赖与连接工厂的构造参数
C# 访问 ActiveMQ 用的是 Apache NMS 抽象加 ActiveMQ 实现，工程里需要Apache.NMS和Apache.NMS.ActiveMQ两个包。装包时选稳定版即可，不要图新去拉预览版，NMS 的 API 在小版本之间偶有行为差异，线上项目跟着预览版走会很被动。
dotnet new console -n AmqDemo cd AmqDemo dotnet add package Apache.NMS dotnet add package Apache.NMS.ActiveMQ两行装包命令之后，先写一个最小连接工厂。连接对象IConnection是重量级的，一个进程内按 broker 维度复用即可，不要每条消息 new 一次；ISession是轻量的，可以按业务模块各自创建。下面的骨架把 URI、账号、连接超时都收在配置里。
using Apache.NMS; using Apache.NMS.ActiveMQ; // 连接串、账号从配置读，别硬编码在业务类里 var brokerUri = "tcp://127.0.0.1:61616"; var userName = "admin"; var password = "admin123"; IConnectionFactory factory = new ConnectionFactory(brokerUri) { UserName = userName, Password = password, // 建连超时，网络抖动时避免线程被长时间挂住 RequestTimeout = TimeSpan.FromSeconds(15) }; using IConnection connection = factory.CreateConnection(); connection.Start(); // 这一行不写，消费者永远收不到消息 Console.WriteLine($"connected: {connection.IsStarted}");ConnectionFactory的构造函数接收 URI，UserName/Password是属性而不是构造参数，这点和不少消息中间件客户端不一样。RequestTimeout控制同步请求的等待上限，取值太小会在 broker 稍慢时抛超时，取值太大则故障时线程堆积，一般给 10 到 30 秒。最后那行connection.Start()是 NMS 的设计：连接创建后处于停止态，必须显式启动才会真正开始投递，新手最容易漏。
2.3 Demo 工程骨架：把连接、会话、收发拆开
一个能长期用的 Demo 不该把所有逻辑塞进Main。我一般拆成三块：连接管理、消息生产、消息消费。配置文件用appsettings.json，把 URI、队列名、订阅名、确认模式全部外置，换环境只改配置。
public sealed class MqConnection : IDisposable { private readonly IConnection _connection; public ISession Session { get; } public MqConnection(string uri, string user, string pwd, AcknowledgementMode mode = AcknowledgementMode.AutoAcknowledge) { var factory = new ConnectionFactory(uri) { UserName = user, Password = pwd }; _connection = factory.CreateConnection(); _connection.Start(); // 会话由连接创建，确认模式在这一层就定死，后面所有消费者共用 Session = _connection.CreateSession(mode); } public void Dispose() { Session?.Close(); _connection?.Close(); // Close 内部会先停投递再断链 } }MqConnection把“建连 + 启动 + 建会话”三步封装，Dispose里按会话到连接的顺序关闭，避免连接已断而会话还在提交事务的尴尬。AcknowledgementMode作为构造参数传进来，是因为同一个进程里经常有两种消费者：日志类的可以自动确认，订单类的必须手工确认甚至走事务，模式不能一刀切。
配置项建议至少包含这几个键：BrokerUri、UserName、Password、QueueName、TopicName、DurableSubscriptionName、ClientId、PrefetchSize。前四个是必填，后面几个在用 Topic 或调优时才有意义。把这套骨架搭起来之后再写业务，改动面会小很多。
3. Queue 与 Topic：两套模型在代码里的真实差别
ActiveMQ 支持点对点和发布订阅两种语义，落到 NMS 上分别是GetQueue和GetTopic。选错模型不会报错，但会出现“消息发出去了却没人消费”或者“十个实例同时处理同一条订单”的诡异现象。这一章把两种模型的发送、接收、过滤都写一遍，重点在参数上，因为参数错了模型再对也没用。
3.1 Queue：点对点下的发送与两种接收姿势
Queue 的语义是每条消息只被一个消费者拿到，多实例部署天然就是竞争消费，适合订单、任务分发这类场景。生产端要注意投递模式：MsgDeliveryMode.Persistent会让 broker 落盘，broker 重启消息还在；NonPersistent快但重启即丢。Demo 默认应该是 Persistent，除非你明确接受丢失。
using Apache.NMS; // queue 由会话创建，同名即同一个逻辑队列 IDestination queue = session.GetQueue("demo.order.queue"); using IMessageProducer producer = session.CreateProducer(queue); producer.DeliveryMode = MsgDeliveryMode.Persistent; ITextMessage msg = session.CreateTextMessage("{\"orderId\":\"A1001\",\"amount\":99.5}"); msg.Properties["orderType"] = "PAY"; // 自定义属性，供选择器过滤 msg.NMSCorrelationID = "A1001"; // 便于链路追踪 // 单条消息也可覆盖投递模式与优先级、过期时间 producer.Send(msg, MsgDeliveryMode.Persistent, MsgPriority.Normal, TimeSpan.FromMinutes(10));GetQueue返回的IDestination只是名字的封装，同名即同队列，不需要提前在 broker 上建。NMSCorrelationID是我强烈建议每条消息都带上的字段，排查时用它可以一路追到下游。Send的重载里第四个参数是过期时间，超过这个时间还没被消费，broker 会把消息丢进死信队列，别把它当成重试超时用。
接收有两种：同步Receive适合批处理或定时拉取，异步Listener适合常驻消费。
// 姿势一：同步拉取，超时返回 null using IMessageConsumer consumer = session.CreateConsumer(queue); IMessage received = consumer.Receive(TimeSpan.FromSeconds(5)); if (received is ITextMessage text) { Console.WriteLine($"got: {text.Text}, corr={text.NMSCorrelationID}"); } // 姿势二：异步回调，连接 Start 之后持续投递 consumer.Listener += m => { if (m is ITextMessage t) { try { Handle(t.Text); // AutoAcknowledge 下，回调正常返回即视为确认 } catch (Exception ex) { // 回调里抛异常，消息是否重投取决于确认模式和 broker 配置 Console.WriteLine($"handle failed: {ex.Message}"); } } };Receive传TimeSpan是阻塞等待，返回null代表这段时间内没有消息，不代表出错；传TimeSpan.Zero是非阻塞探测。异步回调默认在客户端内部线程上执行，不要在回调里做长耗时同步操作，否则预取的消息会全部堵在这一个回调上，表现就是“消费速率上不去”。
3.2 Topic：发布订阅与持久订阅的两个必填项
Topic 的语义是广播，每个订阅者都收到一份。普通订阅只在订阅者在线时有效，进程一停，这期间的消息就没了。要保证离线期间的补投，必须用持久订阅，而持久订阅有两个绕不开的参数：连接上的ClientId和订阅名。
// ClientId 必须在 Start 之前设置，且全局唯一 connection.ClientId = "demo-consumer-01"; connection.Start(); IDestination topic = session.GetTopic("demo.event.topic"); // 持久订阅：名字固定，进程重启后 broker 会把离线消息补上 using IMessageConsumer durable = session.CreateDurableConsumer( (ITopic)topic, "sub-order-event", // 订阅名，同一 ClientId 下唯一 "orderType = 'PAY'", // 选择器，可为 null false); // noLocal durable.Listener += m => Console.WriteLine(((ITextMessage)m).Text);CreateDurableConsumer的四个参数依次是主题、订阅名、选择器、是否屏蔽自己发的消息。订阅名一旦上线就不能随便改，改了相当于换了一个新订阅，历史消息全部丢失。发布端用CreateProducer(topic)正常发，但投递模式必须是 Persistent，否则 broker 不会为离线订阅者保留。
注意：持久订阅的 ClientId 重复会导致后连上的那个把前一个顶掉，表现是“两个实例只有一个能收到”。容器化环境里千万别用固定字符串当 ClientId，带上 Pod 名或机器名。
3.3 消息体与自定义属性：把选择器用起来
NMS 支持几种消息体：ITextMessage、IBytesMessage、IMapMessage、IObjectMessage。跨语言场景下我只推荐前两种，IObjectMessage依赖 .NET 序列化，Java 侧解不开；IMapMessage类似字典，简单结构可以用。业务结构统一用 JSON 塞进ITextMessage，两边都省事。
属性（msg.Properties[...]）和消息头是两回事：消息头是NMSDeliveryMode、NMSPriority、NMSExpiration这类协议字段，属性是你自己加的键值对，专门给选择器用。选择器语法接近 SQL 的 WHERE，支持=、<>、>、<、LIKE、IN、AND、OR，属性名区分大小写。
// 发送端：属性值只用基础类型，别塞对象 msg.Properties["orderType"] = "PAY"; msg.Properties["retryCount"] = 0; msg.Properties["region"] = "north"; // 接收端：选择器写在创建消费者时，broker 侧过滤，不走网络 using IMessageConsumer filtered = session.CreateConsumer(queue, "orderType = 'PAY' AND retryCount < 3");选择器在 broker 端执行，不匹配的消息根本不会下发到客户端。这有个副作用：如果某个队列的所有消费者都带选择器，不匹配的消息会一直堆在队列里，只能靠过期时间或人工清理。所以队列上要么留一个不带选择器的兜底消费者，要么给消息设置NMSExpiration。
属性值的类型也要注意。发int收int没问题，但发int用字符串比较就会永远匹配不上，而且不报错。血泪经验是：所有参与过滤的属性都约定成字符串，能省掉一大半这类玄学问题。
4. 会话、确认与事务：三种确认模式的实际边界
消息“丢没丢”这个问题，九成出在确认模式上。NMS 的AcknowledgementMode有几种取值，语义差别很大，而且和事务会话是互斥的两条路。这一章把自动确认、客户端确认、事务会话三条路径的代码和边界讲清楚，再说资源释放顺序，因为释放顺序错了同样会丢消息。
4.1 AutoAcknowledge 与 ClientAcknowledge 的差别
AutoAcknowledge是最省事的模式，客户端在回调返回或者Receive返回之后自动回确认。它的坑在于“确认”发生得太早：如果你的业务处理在回调里抛了异常，消息已经被确认，broker 那边认为投递成功，不会重投。日志类、可容忍丢失的场景用它没问题，订单类要慎重。
// 自动确认：回调正常返回即确认 using ISession autoSession = connection.CreateSession(AcknowledgementMode.AutoAcknowledge); using IMessageConsumer autoConsumer = autoSession.CreateConsumer(queue); autoConsumer.Listener += m => { var text = ((ITextMessage)m).Text; Process(text); // 这里抛异常，消息也已经确认过了 }; // 客户端确认：处理成功后再手工回执 using ISession clientSession = connection.CreateSession(AcknowledgementMode.ClientAcknowledge); using IMessageConsumer clientConsumer = clientSession.CreateConsumer(queue); clientConsumer.Listener += m => { try { Process(((ITextMessage)m).Text); m.Acknowledge(); // 只有走到这里，broker 才认为消费成功 } catch (Exception ex) { // 不确认，连接断开后 broker 会重投 Console.WriteLine($"nack: {ex.Message}"); } };ClientAcknowledge的确认是累积的：确认第 N 条消息，等于把同一会话上此前所有未确认的消息一并确认。这个语义经常被误解成“逐条确认”，结果就是前面处理失败的消息被后面成功的那条顺手确认掉。要严格逐条，只能一条消息一个会话，代价是连接开销变大，实际项目里更常见的做法是走事务会话。
Acknowledge()调用失败（比如连接已断）会抛异常，这个异常必须捕获，否则回调线程会挂掉，消费者静默失效。
4.2 Transacted 会话：提交与回滚的边界
事务会话用AcknowledgementMode.Transacted创建，消息的确认不再由Acknowledge控制，而是由Commit和Rollback决定。一批消息处理完统一提交，中途任何一条失败就整体回滚，broker 会重投这一批。这是最接近“要么全成功要么全失败”的语义。
using ISession txSession = connection.CreateSession(AcknowledgementMode.Transacted); using IMessageConsumer txConsumer = txSession.CreateConsumer(queue); for (int i = 0; i < 20; i++) { IMessage m = txConsumer.Receive(TimeSpan.FromSeconds(2)); if (m == null) break; try { Process(((ITextMessage)m).Text); } catch (Exception ex) { txSession.Rollback(); // 整批回滚，消息重回队列 Console.WriteLine($"rollback: {ex.Message}"); break; } } txSession.Commit(); // 不提交，消息不会被确认事务会话的边界很关键：一个ISession上如果既做消费又做生产，Commit会同时提交发送和确认，这在“消费一条转发一条”的链路里很方便，但也意味着转发失败会把消费一起回滚。我一般让消费和生产用不同的会话，避免耦合。
回滚不是无限的。broker 对同一条消息的重投次数有上限，默认是 6 次，超过之后消息进死信队列ActiveMQ.DLQ。很多团队发现“消息莫名其妙不见了”，其实是进了 DLQ 而没人看。控制台里 DLQ 的堆积数要纳入监控。
另一个边界是Commit本身失败。网络在提交瞬间断开，客户端不知道 broker 收没收到，这就是典型的不确定状态。业务侧必须能接受重复消费，也就是消费逻辑要幂等——用消息里的业务单号做去重，是最省事的做法。
4.3 资源释放顺序与连接复用
释放顺序错了会丢消息，这个坑很隐蔽。正确顺序是：先关消费者（停止投递），再关生产者，再关会话（此时未提交的事务会回滚），最后关连接。
// 正确顺序：consumer -> producer -> session -> connection consumer?.Close(); producer?.Close(); session?.Close(); connection?.Close(); // 反例：先关连接，会话上的事务提交会抛 NMSException连接是重量级资源，一个进程按 broker 建一到两个就够，不要每次发消息都CreateConnection。但也不要一个连接上挂几百个会话，会话数上去之后客户端内部的锁竞争会很明显。常见的折中是：连接池按进程单例，会话按业务模块各自持有，消费者在模块启动时创建、在模块停止时关闭。
5. 避坑与排查：几类反复出现的现场
Demo 跑通是一回事，放到真实环境里能不能稳住是另一回事。下面这几类是反复出现的现场，每一条都按“现象 → 原因 → 解决”写，遇到问题时对着现象找会快很多。
5.1 连不上：61616 拒绝连接与超时
现象：ConnectionFactory.CreateConnection()抛NMSConnectionException，消息是No I/O has been performed或者直接连接超时。
原因：三种最常见。一是 broker 的 transport connector 绑的是127.0.0.1，跨机器访问必然失败；二是防火墙或安全组没放行 61616；三是 URI 写错，把 8161 控制台端口当成消息端口。
解决：先在客户端机器上用telnet <host> 61616或nc -vz <host> 61616验证端口连通性，不要只在本机测。通了再改conf/activemq.xml里<transportConnector name="openwire" uri="tcp://0.0.0.0:61616"/>的绑定地址，重启 broker 后看日志确认。容器部署时注意端口映射和绑定地址是两件事，映射对了绑定错了照样连不上。
5.2 收不到：Start、订阅与预取三个开关
现象：生产端Send不报错，控制台也能看到队列里有堆积，但消费者的回调一次都没进。
原因：第一是忘了connection.Start()，连接建了但没启动，消费者处于挂起状态；第二是 Topic 用了普通订阅，消息发送时消费者不在线；第三是消费者带的选择器把所有消息都过滤掉了。
解决：先看控制台里该队列的 Consumer Count 是否为 1，为 0 说明消费者压根没注册上；为 1 但堆积不降，看是不是选择器问题，临时去掉选择器再测一次。Topic 场景检查ClientId和订阅名是否设置、投递模式是否为 Persistent。还有一个隐蔽的原因：PrefetchSize设得过大，一批消息全被某个消费者预取走，其他消费者看起来“收不到”，其实是分配不均，把预取调小到 10 到 100 之间再观察。
5.3 重复与丢失：确认模式与异常路径
现象：同一笔订单被处理了两次；或者反过来，日志里明明发了消息，消费端完全没有记录。
原因：重复多半是事务回滚后重投造成的，加上业务处理不幂等；丢失多半是用了AutoAcknowledge而业务处理抛了异常，消息已被确认无法重投。还有一种是被投进了死信队列。
解决：消费逻辑统一用业务单号做幂等，别指望中间件只投一次——它本来就不保证。需要“处理成功才确认”的链路切到ClientAcknowledge或Transacted，并确认异常路径里不会误调用Acknowledge/Commit。丢失排查先看ActiveMQ.DLQ的堆积数，再翻data/activemq.log里有没有重投超限的记录。
5.4 死信队列：重投超限之后
现象：某个队列的堆积突然清零，业务侧不再有失败日志，但下游数据确实缺了一块。
原因：消息重投超过 broker 配置的上限，被自动转入ActiveMQ.DLQ，原队列里就不再有了。默认上限是 6 次，这个值很多团队并不知道。
解决：把 DLQ 的队列深度做成监控指标，一旦大于 0 就告警。处理 DLQ 里的消息一般有两种方式：写一个专门的消费者把消息取出来落库，人工核对后重新投递；或者在 broker 配置里给特定队列设置更宽松的重投策略。不要图省事把重投上限调到很大，那只会把问题推迟，还可能拖垮消费线程。
提示：
ActiveMQ.DLQ是所有队列共用的死信目的地，消息头里有originalDestination字段，用来判断它原本该去哪，写补偿程序时要按这个字段分流。
6. 进阶：failover、预取与可观测性
单机 broker 在 Demo 阶段够用，真上线要面对断连重连。NMS 的 failover 传输把多个 broker 地址串成一个逻辑端点，断线后自动尝试下一个，连接恢复后消费者会重建。URI 里的几个参数决定重连行为，配错了会出现“主节点恢复后客户端还挂在备节点上”的情况。
// 多地址 failover，参数按网络质量调 var uri = "failover:(tcp://10.0.0.11:61616,tcp://10.0.0.12:61616)" + "?randomize=false" + "&initialReconnectDelay=1000" + "&maxReconnectDelay=30000" + "&maxReconnectAttempts=10" + "&timeout=3000"; IConnectionFactory factory = new ConnectionFactory(uri) { UserName = "admin", Password = "admin123" };几个参数的取舍：randomize=false让客户端按顺序尝试，避免两个节点同时被一半客户端打满；initialReconnectDelay是首次重连等待，给 1 秒比较合适；maxReconnectAttempts设成有限值，配合业务侧的降级逻辑，比无限重试可控；timeout是连接建立的超时，太小会在网络抖动时频繁误判。这张表是我一般会写在配置注释里的对照：
| 参数 | 常见取值 | 作用 |
|---|---|---|
| randomize | false | 按地址顺序重连，避免节点负载倾斜 |
| initialReconnectDelay | 1000 ms | 首次重连间隔 |
| maxReconnectDelay | 30000 ms | 重连间隔上限，指数退避封顶 |
| maxReconnectAttempts | 10 | 超过后抛异常，交给业务降级 |
| timeout | 3000 ms | 单次建连超时 |
预取是另一个容易被忽略的旋钮。默认预取会让消费者一次拿走一批消息，好处是吞吐高，坏处是分配不均和内存占用。消费者处理慢、单条耗时长时，把预取降到 10 以内能让负载更均匀；批量计算型消费者可以调大。改法有两个：会话级的consumer.PrefetchSize = 10，或者 URI 上加nms.PrefetchPolicy.queuePrefetch=10，前者优先级更高。
可观测性这块，控制台的 Queues 页面只适合人眼看，自动化要靠 JMX 或 broker 的统计插件把队列深度、消费者数、入队出队速率导出来。我一般至少接三个指标：队列深度、DLQ 深度、消费者数量。前两个看堆积，第三个看消费端是不是静默掉线了——消费者数从 4 掉到 1，队列深度却还在缓慢上涨，这种曲线比任何日志都直观。
从那以后我每次接手带消息中间件的 .NET 项目，都会先把连接串、确认模式、DLQ 深度这三样东西抄在纸上核对一遍再动业务代码，希望帮到你。
本文还有配套的精品资源，点击获取
