企业官网建设流程全解析

简介：这份 ActiveMQ Demo 面向使用 C# 与 WinForm 的开发者，尤其是刚开始接触消息队列、需要快速理解点对点通信流程的学习者。它提供发送端与接收端两个可运行示例，可用于本地联调、课程实验或项目原型验证，帮助读者对照代码梳理 ActiveMQ 的连接建立、消息发送与接收等基础环节。

压缩包共 36 个文件，约 326KB，包含 C# 源码、WinForm 窗体资源、依赖 DLL、可执行程序，以及 Visual Studio 解决方案和项目配置，结构上便于分别查看生产端与消费端的实现。已有 1293 人学习或下载，具备一定的参考基础。

通过这份示例，读者可以观察消息队列在桌面应用中的基本用法，理解界面操作与消息处理逻辑的组织方式，并据此搭建自己的测试环境、验证通信流程，或作为后续功能扩展的起点。

1. 从一条丢掉的工单说起：C# 侧接 ActiveMQ 真正该盯的是哪几个参数

上周帮一个做产线采集的团队看消息链路，业务代码没毛病，问题出在他们把 ActiveMQ 当成"发出去就完事"的管道：生产者 Send 完就返回，消费者接到就处理，中间没有 ACK、没有重投、没有死信兜底。一次网络抖动，几十条工单状态没了，查日志查了一下午。

这份 ActiveMQ Demo（C#）的价值就在这里：它把连接、会话、ACK 模式、持久化、重投策略、死信队列这几层参数摊开成一个能直接跑的控制台程序，让你在写业务之前先把边界摸一遍。适合手上是 .NET Framework 或 .NET 6+ 的老项目、需要给系统补一层异步消息、又不想直接上重量级消息平台的开发者。十分钟能跑通第一条消息，真正花时间的是后面那几个参数。

2. 环境与依赖：61616、8161 两个端口和一对 NuGet 包

先把 Broker 起起来，再把客户端包装对。这两步任何一步错了，后面调试全是噪音——分不清是服务端没起来、端口不通，还是客户端版本打架。我一般习惯先把服务端跑通、控制台能登进去看到队列，再动 C# 代码。

2.1 本机起 Broker：61616 走消息，8161 走控制台

ActiveMQ 的发行包解压即用，不用装数据库。Windows 上直接双击activemq.bat也能跑，但那样窗口一关进程就没了，日志也不好翻，所以我一律用脚本启动。

# 进入解压后的 bin 目录。脚本启动的好处是进程后台化、日志落在 data 目录 cd apache-activemq-5.x/bin ./activemq start # Linux / macOS # activemq.bat start # Windows，用管理员命令行 # 确认 OpenWire 端口（NMS 客户端默认连的就是这个）已在监听 ss -lntp | grep 61616 # Linux netstat -ano | findstr 61616 # Windows

启动后有几个东西要记住。61616是 OpenWire 协议的默认端口，Apache.NMS.ActiveMQ 这个客户端走的就是它；8161是 Web 控制台，浏览器访问http://127.0.0.1:8161/admin，默认凭据是admin/admin，第一次登进去就该改掉。同一个 Broker 通常还会开着 5672（AMQP）、61613（STOMP）、1883（MQTT），如果只是给 C# 用，这些端口不用管，但要注意别和本机已占用的端口冲突。

日志在data/activemq.log，启动失败时先看这里而不是看控制台。常见的是端口被占（报Address already in use）和 Java 堆不够（报OutOfMemoryError）。消息持久化默认走 KahaDB，落地在data/kahadb目录，这个目录的体积会随着未消费消息增长，压测前先留意磁盘。

提示：改conf/activemq.xml里的transportConnector可以改端口和绑定地址。默认只绑本机时，别的机器连不上，会表现为连接超时而不是拒绝，排查时容易误判。

2.2 NuGet 包：NMS API 与 ActiveMQ 实现必须成对

.NET 这边有两层包：Apache.NMS是接口定义，Apache.NMS.ActiveMQ是具体的协议实现。两者主版本必须对齐，混用最典型的症状是运行时抛MissingMethodException或TypeLoadException——编译期一切正常，一跑就炸，属于典型的血泪经验。

# 建一个控制台项目，只需要装实现包，它会带上对应的 NMS 依赖 dotnet new console -n MqDemo cd MqDemo dotnet add package Apache.NMS.ActiveMQ # 查看实际解析到的版本，确认两个包在同一档 dotnet list package --include-transitive

参数上要注意两点。第一，别手动去装一个更高版本的Apache.NMS，让它跟着实现包走，否则 NuGet 会给你解析出一对不匹配的组合。第二，老 .NET Framework 项目用packages.config方式安装时，注意包的目标框架，netstandard 版本的包在 net40 项目里引不进来，得挑对目标框架的版本。

装完在obj/project.assets.json里能查到实际版本号。生产项目我建议把版本号在Directory.Packages.props里锁死，别用浮动版本，否则某次 CI 重新还原就可能拉到不兼容的组合。

2.3 第一条消息：工厂 → 连接 → 会话 → 目标 → 生产者

NMS 的 API 结构和 JMS 是一一对应的，五层对象：连接工厂、连接、会话、目标、生产者/消费者。写的时候按这个顺序摆，代码就不会乱。

using Apache.NMS; using Apache.NMS.ActiveMQ; // 1) 连接工厂：activemq: 前缀可省，写上更直观；URI 决定走 tcp 还是 failover var factory = new ConnectionFactory("activemq:tcp://127.0.0.1:61616") { UserName = "admin", Password = "admin", }; // 2) 连接：构造时不建 TCP，Start() 才真正握手，异常也在这里抛 using IConnection connection = factory.CreateConnection(); connection.Start(); // 3) 会话：ACK 模式在这一层定死，后面改不了，只能新建会话 using ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge); // 4) 目标：队列名大小写敏感，生产端和消费端必须一字不差 IDestination queue = session.GetQueue("order.status"); // 5) 生产者：持久化投递要显式设置，别指望默认值 using IMessageProducer producer = session.CreateProducer(queue); producer.DeliveryMode = MsgDeliveryMode.Persistent; ITextMessage msg = session.CreateTextMessage("{\"orderId\":\"A1001\",\"state\":\"paid\"}"); msg.Properties["traceId"] = Guid.NewGuid().ToString("N"); // 自定义属性，消费端可用选择器过滤 msg.NMSCorrelationID = "A1001"; // 请求响应模式下的关联 ID producer.Send(msg, MsgDeliveryMode.Persistent, MsgPriority.Normal, TimeSpan.FromMinutes(5)); Console.WriteLine($"sent {msg.NMSMessageId}");

几个参数的实际含义值得单独说。UserNam/Password是 Broker 侧配置的凭据，默认admin/admin，生产环境必然要改。AcknowledgementMode决定消费端什么时候向 Broker 回执，这里先给AutoAcknowledge，第 4 章会拆开讲四档的差别。MsgDeliveryMode.Persistent表示消息要落盘，Broker 重启后还在；不设的话默认是NonPersistent，只走内存。Send的第四个参数是过期时间，设了之后消息到期没被消费会被 Broker 丢弃，业务上"状态通知"这类消息建议设个上限，免得积压半年才发现。

跑通这一步，去 8161 控制台的 Queues 页面能看到order.status这个队列，Messages Enqueued 是 1，Messages Dequeued 是 0。这个数字对不上，八成是队列名写错了或者连到了别的 Broker。

3. Queue 与 Topic 的代码骨架：两种模型选错，后面全是返工

消息模型选错是返工成本最高的一类错误。Queue 是一对一投递，Topic 是一对多广播，两者在代码层面只差一个GetQueue和GetTopic，但在运维层面差得远——前者要盯积压，后者要盯订阅者在线状态。这里把两套骨架都给出来。

3.1 Queue：一条消息只被一个消费者拿走，均衡靠 prefetch 摊

Queue 的语义是竞争消费：同一个队列上挂 N 个消费者，每条消息只投给其中一个。写消费端的时候要注意，Receive是同步阻塞的，Listener是回调的，两者别混用。

using Apache.NMS; using Apache.NMS.ActiveMQ; var factory = new ConnectionFactory("activemq:tcp://127.0.0.1:61616") { UserName = "admin", Password = "admin", }; // prefetch 决定 Broker 一次推多少条给消费者，设大了会导致分配不均 factory.PrefetchPolicy.QueuePrefetch = 1; using IConnection connection = factory.CreateConnection(); connection.Start(); using ISession session = connection.CreateSession(AcknowledgementMode.ClientAcknowledge); IDestination queue = session.GetQueue("order.status"); using IMessageConsumer consumer = session.CreateConsumer(queue); // 回调方式：注意回调线程来自 NMS 内部，别在里面做长耗时阻塞 consumer.Listener += message => { if (message is ITextMessage text) { Console.WriteLine($"got {text.NMSMessageId} -> {text.Text}"); try { // 业务处理放这里；处理成功才回执，失败就抛出去让它重投 message.Acknowledge(); } catch (Exception ex) { Console.WriteLine($"handle failed: {ex.Message}"); } } }; Console.ReadLine(); // 撑住进程，别让 Main 直接退出

QueuePrefetch = 1是我在消费耗时差异大的场景下的默认选择。Broker 默认会给每个消费者预推一批消息，推出去的消息别的消费者就拿不到了，如果第一个消费者处理慢，消息全堆在它本地，后面几个消费者干看着——这就是所谓的消费不均。设成 1 相当于关掉预取，代价是多几次网络往返。如果每条消息处理很快、追求吞吐，把它调大（比如 100）更划算。

3.2 Topic：非持久订阅掉线就漏消息，持久订阅要靠 ClientId 和订阅名

Topic 的语义是广播，每个订阅者都收到一份。默认的订阅是非持久的：消费者一断开，Broker 就不再为它保留消息，重新连上期间的消息全丢。要保证不漏，得用持久订阅，而持久订阅有两个硬性前置条件。

var factory = new ConnectionFactory("activemq:tcp://127.0.0.1:61616") { UserName = "admin", Password = "admin", }; using IConnection connection = factory.CreateConnection(); // 关键点一：ClientId 必须在 Start() 之前设好，否则持久订阅建不起来 connection.ClientId = "station-collector-01"; connection.Start(); using ISession session = connection.CreateSession(AcknowledgementMode.ClientAcknowledge); ITopic topic = session.GetTopic("device.alarm"); // 关键点二：订阅名是持久订阅的身份，同一 ClientId 下不能重复 using IMessageConsumer consumer = session.CreateDurableConsumer(topic, "sub-alarm-01"); consumer.Listener += message => { if (message is ITextMessage text) { Console.WriteLine($"[alarm] {text.Text}"); message.Acknowledge(); } }; Console.ReadLine();

ClientId相当于这个客户端的身份证，sub-alarm-01是订阅名，两者组合唯一确定一个持久订阅。Broker 会在消费者离线期间把消息存下来，等它重连再补发。这套机制在 8161 控制台的 Subscribers 页面能看到，Offline 状态的持久订阅也会一直占着存储，长期不用的要手动清理，不然存储只涨不跌。

还有一个容易踩的点：同一个ClientId不允许两个连接同时在线，第二个连接会被拒绝。做多实例部署时ClientId要带机器标识，别写死。

3.3 消息体选型：TextMessage 优先，ObjectMessage 慎用

NMS 提供了几种消息体：ITextMessage、IBytesMessage、IObjectMessage、IMapMessage。Demo 里最常见的是IObjectMessage，因为可以直接塞一个 C# 对象进去，写起来最省事。但跨语言、跨版本、跨团队的时候，我不推荐它。

IObjectMessage走的是 .NET 的二进制序列化，消费端必须能加载同一个类型、同一个版本的程序集。换句话说，生产端升级了类的字段，消费端没同步升级，就会抛反序列化异常，而且这个异常发生在消息已经出队之后，消息直接进死信队列。跨语言更没法用，Java 端读不了 .NET 的序列化格式。

IBytesMessage更适合做长期方案：自己选序列化方式（JSON、Protobuf、MessagePack），字节流谁都能读，类型演进也可控。

// 生产端：自己序列化成字节，消息体只是容器 IBytesMessage bytes = session.CreateBytesMessage(); byte[] payload = JsonSerializer.SerializeToUtf8Bytes(new { OrderId = "A1001", State = "paid" }); bytes.WriteBytes(payload); bytes.Properties["contentType"] = "application/json"; // 属性带上格式标识，消费端才知道怎么解 bytes.Properties["schemaVersion"] = 2; producer.Send(bytes, MsgDeliveryMode.Persistent, MsgPriority.Normal, TimeSpan.FromMinutes(5)); // 消费端：读字节 + 按 contentType 分发解析 if (message is IBytesMessage bm) { var buffer = new byte[bm.BodyLength]; bm.ReadBytes(buffer); var json = Encoding.UTF8.GetString(buffer); var order = JsonSerializer.Deserialize<OrderState>(json); Console.WriteLine($"{order?.OrderId} {order?.State}"); }

Properties["contentType"]和schemaVersion不是 ActiveMQ 的内置字段，是我自己加的约定。有了这两个属性，消费端可以对不同版本做兼容分支，也可以做灰度。自定义属性还能用于消息选择器，session.CreateConsumer(queue, "schemaVersion = 2")就只收新格式的消息，老消费者不受影响。

4. 可靠性参数怎么落：ACK 四档、持久化、事务与重投策略

业务能跑起来和消息不丢，中间隔着四个参数。这四个参数彼此关联，改一个会影响另一个，所以放在一起讲。

4.1 AckMode 四档：回执时机决定了重复与丢失的边界

ACK 模式的本质是回答一个问题：消费端什么时候告诉 Broker"这条我处理完了，可以删"。四档的差别如下。

AutoAcknowledge是 Demo 里最常用的，也是误解最多的。它并不是"处理完才回执"，而是在消息交给应用代码之后就回执了，此时业务还没跑。进程在业务处理中崩掉，这条消息在 Broker 看来已经消费成功，永远不会重投。

真正符合"至少一次"语义的是ClientAcknowledge：业务成功再回执，失败就不回执，连接断开后 Broker 会重投。代价是必须自己保证幂等，因为重投意味着同一条消息可能被处理两次。

using ISession session = connection.CreateSession(AcknowledgementMode.ClientAcknowledge); using IMessageConsumer consumer = session.CreateConsumer(session.GetQueue("order.status")); consumer.Listener += message => { if (message is not ITextMessage text) return; // 幂等表：用消息 ID 或业务主键去重，这是 ClientAcknowledge 的必备配套 if (ProcessedStore.Exists(text.NMSMessageId)) { message.Acknowledge(); // 处理过也要回执，否则会一直被重投 return; } ProcessBusiness(text.Text); ProcessedStore.Mark(text.NMSMessageId); message.Acknowledge(); };

幂等这块我在项目里一律用消息 ID 建去重表，或者用业务主键做唯一约束。不要指望"消息不会重复"——在ClientAcknowledge和DupsOkAcknowledge下重复是设计的一部分，不是 bug。

4.2 持久化：DeliveryMode 与 Broker 侧的存储

MsgDeliveryMode.Persistent表示消息在返回生产端之前已经写入 Broker 的存储，Broker 重启后消息还在。NonPersistent只写内存，重启即丢，但因为少一次磁盘 IO，吞吐高好几倍。

选择标准很直接：这条消息丢了业务能不能接受。工单状态、支付结果这类不能丢，用 Persistent；实时曲线、心跳这类下一秒就有新值覆盖的，用 NonPersistent 更划算。

持久化有两个配套参数要一起看。Broker 侧conf/activemq.xml里的<persistenceAdapter>决定用 KahaDB 还是 JDBC 存储，Demo 用默认的 KahaDB 即可；生产端Send的过期时间决定消息在存储里待多久，不设就是永久，长期积压会把磁盘吃满。另外 Broker 还有个producerFlowControl开关，磁盘使用率超过阈值时会阻塞生产端而不是让消息写失败，这是保护机制但会表现为"生产端 Send 卡住不动"，第一次遇到容易误判成网络问题。

4.3 事务会话：Commit 才回执，Rollback 整批重投

事务会话把多条消息打包成一个原子单元：发多条消息，Commit一起生效，Rollback一条都不发；收多条消息，Commit一起回执，Rollback一起重投。

using ISession session = connection.CreateSession(AcknowledgementMode.Transactional); IDestination queue = session.GetQueue("order.status"); using IMessageProducer producer = session.CreateProducer(queue); try { for (int i = 0; i < 10; i++) { ITextMessage msg = session.CreateTextMessage($"{{\"seq\":{i}}}"); producer.Send(msg, MsgDeliveryMode.Persistent, MsgPriority.Normal, TimeSpan.FromMinutes(5)); } session.Commit(); // 到这里 10 条才真正可见 } catch (Exception ex) { session.Rollback(); // 全部作废，Broker 不会看到其中任何一条 Console.WriteLine($"batch rolled back: {ex.Message}"); }

事务的边界要按业务来划，不要按条数来划。一批里只要有一条失败就得全部回滚，那就说明这批消息在业务上本来就是一个整体。为了凑数量硬塞在一起，回滚时会把不该回滚的也带下去。

注意事务会话和 ACK 模式是互斥的：一旦选了Transactional，就不要再调message.Acknowledge()，回执由Commit负责。

4.4 RedeliveryPolicy 与死信：重投几次、间隔多久、进哪个队列

消费失败后 Broker 不会无限重投，重投到上限就扔进死信队列（DLQ）。这个次数、间隔、退避策略都可以配。

var factory = new ConnectionFactory("activemq:tcp://127.0.0.1:61616") { UserName = "admin", Password = "admin", // 重投策略挂在工厂上，工厂创建的消费者都继承这套 RedeliveryPolicy = new RedeliveryPolicy { MaximumRedeliveries = 3, // 3 次还处理不了，多半是代码问题，别死磕 InitialRedeliveryDelay = 2000, // 首次等 2 秒，给下游恢复留时间 UseExponentialBackOff = true, // 2s -> 10s -> 50s，避免瞬间打爆下游 BackOffMultiplier = 5, }, };

MaximumRedeliveries调小还是调大，取决于失败原因。下游数据库短暂抖动，重投几次就恢复了，默认值够用；如果是消费端代码 bug，重投 6 次只是把失败重复 6 遍，还会拖慢整个队列的消费。我一般会把它压到 2 到 3 次，配合监控盯 DLQ 的数量，DLQ 一涨就说明有问题，比在生产队列里慢慢重投更容易发现。

死信队列的默认名字是ActiveMQ.DLQ，所有队列的失败消息都混在一起。在conf/activemq.xml里可以配置individualDeadLetterStrategy，让每个队列有自己的 DLQ，排查时能直接看出是哪条业务链路出的问题。

5. 排查手册：连不上、消费不动、反序列化失败与句柄泄漏

这几条都是我在实际项目里遇到过的，按现象、原因、解决三段写，遇到时可以直接对号入座。

5.1 现象：本机跑得好好的，换台机器就连接超时

原因：Broker 的transportConnector默认绑定地址可能是0.0.0.0也可能是某个具体网卡地址，取决于配置；另一种情况是客户端 URI 写的是localhost，在容器里localhost指向容器自己，不是宿主机。

解决：先在目标机器上用telnet 172.x.x.x 61616或Test-NetConnection确认端口通不通，不通就去改conf/activemq.xml的transportConnector绑定地址并重启；URI 里不要用localhost，写实际 IP。生产环境直接上failover:(tcp://a:61616,tcp://b:61616)?randomize=false&initialReconnectDelay=1000，节点切换由客户端负责，别在代码里自己写重连循环。

5.2 现象：消费者挂上去，日志一条不打印，CPU 也不高

原因：同步Receive和异步Listener混用，或者开了两个消费者但只 Start 了其中一个连接。还有一种更隐蔽的：队列名在生产端和消费端大小写不一致，order.status和Order.Status是两个不同的队列，消费端挂在一个空队列上，永远不会被唤醒。

解决：去 8161 控制台看这个队列的 Consumers 数量和 Messages Enqueued。Consumers 是 0 说明连接压根没建起来，去查连接和Start()；Consumers 有值但 Enqueued 一直是 0，说明生产端发到别的队列去了，把两边的名字贴出来逐字符比。同步和异步不要混用，一个消费者只用一种方式。

5.3 现象：升级了一版业务类，消费端开始频繁报反序列化异常

原因：用了IObjectMessage，生产端程序集版本变了，消费端的类结构对不上，二进制反序列化直接失败。这类消息已经被 Broker 投递出去，失败后进重投，重投到上限进 DLQ，队列里看起来消息消失了。

解决：短期先去 DLQ 把消息捞出来人工处理；长期换掉IObjectMessage，改成IBytesMessage加 JSON 或 Protobuf，并在消息属性里带schemaVersion，消费端按版本分支兼容。Java 端的ObjectMessage也是同样的问题，跨语言场景下基本不能用。

5.4 现象：服务跑几个小时后连接数暴涨，Broker 侧看到大量空闲连接

原因：IConnection、ISession、IMessageConsumer没有 Dispose。NMS 的对象实现了IDisposable，但很多人只在方法里new出来用完就丢，GC 回收之前连接一直挂着。连接本身是相对重的资源，泄漏几十个之后 Broker 就开始拒绝新连接。

解决：所有 NMS 对象一律用using，或者在一个长生命周期的作用域里创建、在应用关闭时统一释放。连接和会话应该复用，不要每条消息建一个连接——那是压测时最容易翻车的地方，本机测几十条没问题，一上量直接打满。另外注意Listener回调里不要抛未捕获异常，NMS 内部线程被异常打断后，消费者会静默失联，队列积压但没人消费。

6. 上线前的一次体检：探针消息加三个 Broker 指标

Demo 跑通不等于链路可用。上线前我会做一次端到端体检，做法是往真实队列发一条带标记的探针消息，确认它被消费、被回执、没有进 DLQ。

// 探针消息：带唯一标记，消费端识别后打日志并回执，不会污染业务数据 string probeId = $"probe-{DateTime.UtcNow:yyyyMMddHHmmss}-{Guid.NewGuid():N}"; ITextMessage probe = session.CreateTextMessage("{\"type\":\"healthcheck\"}"); probe.Properties["probeId"] = probeId; probe.Properties["source"] = "preflight"; producer.Send(probe, MsgDeliveryMode.Persistent, MsgPriority.Normal, TimeSpan.FromMinutes(1)); // 同一进程里同步等一会儿，确认消息没有堆积 using IMessageConsumer checker = session.CreateConsumer(session.GetQueue("order.status"), $"probeId = '{probeId}'"); IMessage? echo = checker.Receive(TimeSpan.FromSeconds(10)); Console.WriteLine(echo is null ? "探针未回收，链路有问题" : "探针回收成功，端到端通");

probeId用选择器过滤，避免捞到业务消息；过期时间设 1 分钟，探针本身不该长期占用存储。这段代码放在部署脚本的最后一步，失败就直接拦下发布。

光有探针还不够，Broker 侧有三个指标要盯：队列的 Messages Enqueued 与 Dequeued 差值（积压量）、Consumers 数量（消费能力）、DLQ 的消息数（失败量）。这三个数在 8161 控制台都能看到，也可以走 JMX 拉出来接监控。积压持续上涨说明消费能力不够，Consumers 突然归零说明消费端挂了，DLQ 上涨说明业务处理在报错——三条曲线一起看，基本能定位九成的问题。

那次工单丢失之后，我给自己定了个规矩：任何消息链路交付前，探针消息、积压曲线、DLQ 数量这三样必须当着运维的面走一遍，光说"我本地跑通了"不算数。这套动作花不了十分钟，但能挡掉大部分上线后才发现的问题。希望帮到你。

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