企业官网建设流程全解析

简介：这份 ActiveMQ Demo（C#）资源面向使用 .NET 平台进行消息中间件开发的学习者与工程师，重点解决在 WinForm 环境下如何快速跑通 ActiveMQ 消息发送与接收流程的问题。压缩包共 36 个文件，约 326KB，包含 19 个 cs 源码文件、4 个 resx 资源文件、4 个 dll 依赖库、3 个 exe 可执行程序以及 csproj、sln 等工程配置，整体结构完整，可直接用 Visual Studio 打开编译运行。资源内含生产者与消费者两个独立工程，分别演示消息的创建、发送、监听与接收处理，并封装了 MQ 连接、全局函数与列表视图排序等辅助类，便于理解 ActiveMQ 在 C# 客户端中的典型调用方式。目前已有 1292 人学习下载，适合刚接触消息队列、需要一份可运行示例来对照调试的开发者参考，也可作为 WinForm 项目中集成 ActiveMQ 的起步模板。

1. 从一次消息丢失事故说起：ActiveMQ 的 C# 接入到底难在哪

去年帮一个做订单系统的团队排查问题，现象很诡异：生产端日志显示消息发送成功，消费端却隔三差五丢单。查了两天才发现，他们用的是默认的自动确认模式，消费者拿到消息后业务逻辑抛了异常，消息已经被 broker 标记为消费完成，直接进了黑洞。这类问题在 C# 接入 ActiveMQ 的场景里特别常见，因为 .NET 生态里主流的客户端库 Apache.NMS.ActiveMQ 和 Java 侧的 API 差异不小，很多从 Java 转过来的开发者会想当然地套用经验。

这份 ActiveMQ Demo（C#）就是围绕这个痛点整理的：它把连接工厂创建、会话管理、生产者发送、消费者接收、事务与确认模式这几条主线用可运行的代码串起来，覆盖点对点队列和发布订阅两种模型。适合两类人：一是刚接触消息中间件、需要一份能跑通的 C# 参考实现的开发者；二是已经在用但被确认机制、连接复用、序列化这些问题折腾过的老手，可以拿它当对照检查自己的代码。下面我按实际拆包的顺序，把这份 Demo 的结构、跑法、参数和坑一条条讲清楚。

2. 拆开 Demo 看结构：连接工厂、会话与两种消息模型

拿到一份 C# 的 ActiveMQ Demo，第一件事不是急着跑，而是先看清它把哪些概念落成了代码。ActiveMQ 的核心抽象其实就四层：ConnectionFactory 负责建连接，Connection 是一条到 broker 的物理链路，Session 是轻量的会话上下文（单线程使用），Destination 则是队列或主题。Demo 通常会把这几层封装成工具类，再在上面写生产者和消费者。

2.1 连接字符串与 ConnectionFactory 的构造

C# 侧用的是 Apache.NMS 和 Apache.NMS.ActiveMQ 两个包，前者是接口规范，后者是具体实现。连接字符串的写法直接决定连的是哪台 broker、用哪种传输协议。Demo 里一般会把它抽到配置文件，常见形式是这样：

<!-- App.config 中的连接配置，brokerUri 指向 ActiveMQ 的 OpenWire 端口 --> <appSettings> <add key="ActiveMQ.BrokerUri" value="tcp://localhost:61616"/> <add key="ActiveMQ.UserName" value="admin"/> <add key="ActiveMQ.Password" value="admin"/> <add key="ActiveMQ.QueueName" value="order.queue"/> </appSettings>

对应的初始化代码：

// 从配置读取连接参数，创建连接工厂 var brokerUri = ConfigurationManager.AppSettings["ActiveMQ.BrokerUri"]; var factory = new ConnectionFactory(brokerUri); // 创建连接并启动，Start() 之后才能真正收发消息 IConnection connection = factory.CreateConnection( ConfigurationManager.AppSettings["ActiveMQ.UserName"], ConfigurationManager.AppSettings["ActiveMQ.Password"]); connection.Start(); // 创建会话，AcknowledgementMode 决定确认行为，这里先用自动确认 ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);

这里有几个参数值得说清楚。tcp://是 OpenWire 协议，也是 .NET 客户端默认走的；如果 broker 开了其他端口，URI 要跟着改。CreateConnection传入的用户名密码对应 broker 的认证配置，本地默认环境通常是 admin/admin，生产环境必须换。CreateSession的确认模式是整个 Demo 里最关键的参数，后面单独讲。连接创建后必须调用Start()，否则消息不会真正流动，这是新手最容易漏的一步。

2.2 队列与主题：点对点和发布订阅的代码差异

Demo 一般会同时演示 Queue 和 Topic 两种 Destination。Queue 是点对点，一条消息只被一个消费者拿走；Topic 是发布订阅，所有当前订阅者都能收到。代码上的差别只在创建 Destination 的那一行：

// 点对点：多个消费者竞争同一条消息 IDestination queue = session.GetQueue("order.queue"); // 发布订阅：所有订阅者各收一份 IDestination topic = session.GetTopic("order.topic"); // 生产者发送 IMessageProducer producer = session.CreateProducer(queue); producer.DeliveryMode = MsgDeliveryMode.Persistent; // 持久化，broker 重启不丢 ITextMessage message = session.CreateTextMessage("订单创建:1001"); producer.Send(message);

DeliveryMode有两个取值：Persistent会把消息落盘，broker 重启后还在；NonPersistent只放内存，性能高但重启即丢。订单类业务必须用 Persistent，日志类可以放宽。消费者侧用CreateConsumer加Listener事件回调接收：

// 消费者注册监听，消息到达时触发回调 IMessageConsumer consumer = session.CreateConsumer(queue); consumer.Listener += new MessageListener(OnMessageReceived); private static void OnMessageReceived(IMessage message) { // 判断消息类型再转换，避免强转异常 if (message is ITextMessage textMessage) { Console.WriteLine($"收到: {textMessage.Text}"); } }

Listener是异步回调，回调里抛异常不会自动重试，这点和同步Receive()的行为不同。Demo 里如果两种都写了，可以对比着看。

2.3 把 Demo 跑起来的最小步骤

假设本地已经装好 broker，跑通 Demo 的流程大致是：先确认 broker 的 OpenWire 端口（默认 61616）在监听，再还原 NuGet 包，然后先启动消费者进程，再启动生产者进程。顺序反了的话，非持久化消息会丢，持久化消息虽然不丢但队列里会堆积。还原依赖的命令：

# 还原 Apache.NMS 相关包，版本以 Demo 的 packages.config 为准 nuget restore ActiveMQDemo.sln # 编译 msbuild ActiveMQDemo.sln /p:Configuration=Release # 先跑消费者，再跑生产者 ActiveMQDemo.Consumer.exe ActiveMQDemo.Producer.exe

如果消费者一直收不到消息，先别怀疑代码，去 broker 的管理控制台看队列的 Enqueue 和 Dequeue 计数。Enqueue 涨了 Dequeue 不动，说明消息进了队列但没人消费，多半是消费者没连上或者监听的 Destination 名字对不上。

3. 确认模式与事务：消息不丢不重的关键参数

Demo 里最值得反复看的就是确认模式这块，因为它直接决定消息会不会丢、会不会重复。很多线上事故的根因都在这几行代码上。

3.1 四种确认模式的适用场景

Apache.NMS 提供四种AcknowledgementMode，行为差异很大：

Demo 默认多半用 AutoAcknowledge，因为它代码最简单，但这也是最容易翻车的模式。业务处理失败时消息已经被确认，broker 不会重发，等于静默丢失。生产环境我一般建议改成 ClientAcknowledge，处理成功后再手动确认：

// 手动确认模式，业务成功后再 Acknowledge ISession session = connection.CreateSession(AcknowledgementMode.ClientAcknowledge); IMessageConsumer consumer = session.CreateConsumer(queue); consumer.Listener += (msg) => { try { var text = (msg as ITextMessage)?.Text; ProcessOrder(text); // 业务处理 msg.Acknowledge(); // 只有成功才确认 } catch (Exception ex) { // 不确认，broker 会在会话关闭或重连后重投 Console.WriteLine($"处理失败，不确认: {ex.Message}"); } };

注意Acknowledge()的粒度：在 ClientAcknowledge 下，确认一条会连带确认该会话中之前所有已投递的消息，所以处理顺序和确认时机要配合好。

3.2 事务会话的提交与回滚

如果业务需要「本地数据库写入 + 消息确认」保持原子性，就得用事务会话。Demo 里通常有一段对比代码：

// 事务会话，发送和确认都在事务内，Commit 才生效 ISession session = connection.CreateSession(AcknowledgementMode.Transactional); IMessageProducer producer = session.CreateProducer(queue); try { producer.Send(session.CreateTextMessage("订单创建:1002")); SaveToDatabase(); // 本地操作 session.Commit(); // 一起提交 } catch { session.Rollback(); // 一起回滚，消息不会发出 }

Commit()之前消息对消费者不可见，Rollback()会丢弃本次事务内所有发送和确认。这里有个边界：事务会话不能和异步Listener混用，因为事务是绑定在会话线程上的，异步回调线程模型对不上，容易出诡异问题。要用事务就老老实实用同步Receive()。

3.3 持久化与预取：影响吞吐的两个隐藏参数

除了确认模式，还有两个参数在 Demo 的配置里经常被忽略。一个是生产者的DeliveryMode，前面提过。另一个是消费者的预取大小Consumer.PrefetchSize，它决定 broker 一次往消费者推多少条消息到本地缓冲。默认值偏大，消费者一多就容易出现「消息都堆在某一个消费者本地、其他消费者闲着」的负载不均。可以在连接 URI 上调整：

// 通过 URI 参数控制预取，prefetchPolicy 越小负载越均衡，但吞吐略降 var uri = "tcp://localhost:61616?nms.prefetchPolicy.queuePrefetch=1"; var factory = new ConnectionFactory(uri);

queuePrefetch=1适合消费耗时长的场景，能让消息均匀分给多个消费者；如果单条处理很快、追求吞吐，可以适当调大。这个值没有万能解，得压测。

4. 避坑与排查：C# 接入 ActiveMQ 的五个高频翻车点

Demo 能跑通不代表生产能用，下面这几条是我和身边人踩过的真实坑，按现象、原因、解决来写。

4.1 消费者收不到消息，队列却在涨

现象：broker 控制台显示队列 Enqueue 持续增加，Dequeue 为 0，消费者进程没报错。原因通常是 Destination 名字不一致，比如生产者发到order.queue，消费者监听Order.Queue，ActiveMQ 对大小写敏感，这是两个不同的队列。解决：把队列名抽成常量或配置项，生产消费共用同一份，别在两处手写字符串。

4.2 连接空闲一段时间后自动断开

现象：服务跑几小时后第一次发消息抛Connection is closed或超时。原因是 broker 默认的wireFormat.maxInactivityDuration是 30 秒，空闲超时会断开连接，而客户端没配心跳或重连。解决：在连接 URI 上加心跳参数，并开启客户端的自动重连：

// 心跳 30 秒，客户端自动重连，避免空闲断连 var uri = "tcp://localhost:61616?wireFormat.maxInactivityDuration=30000" + "&wireFormat.maxInactivityDurationInitalDelay=10000"; var factory = new ConnectionFactory(uri); factory.ClientId = "order-consumer-01"; // 连接异常时触发重连，重连后需要重建 Session 和 Consumer

注意自动重连只恢复连接，不恢复会话和消费者，重连回调里要自己重建，否则连接回来了消费者还是死的。

4.3 消息体中文乱码

现象：发送中文，消费端收到问号或乱码。原因是ITextMessage的编码和 broker 或控制台的编码不一致，或者发送时用了CreateBytesMessage却没指定字符集。解决：统一用CreateTextMessage发文本，确保两端都用 UTF-8；如果必须用字节消息，发送前显式Encoding.UTF8.GetBytes，接收端对应解码。

4.4 大量消息时内存暴涨

现象：消费者进程内存持续增长直到 OOM。原因是预取大小太大，broker 一次推了几千条到消费者本地缓冲，加上Listener回调处理慢，消息全堆在内存里。解决：调小prefetchPolicy，或者改用同步Receive()逐条拉取，用拉模式控制节奏。另外确认模式选DupsOkAcknowledge时批量确认也会积压，要权衡。

4.5 事务会话里用 Listener 导致确认失效

现象：事务会话配异步监听，Commit()后消息状态不对，或者回滚没生效。原因是事务绑定在会话线程，异步回调在另一个线程上执行，事务上下文丢失。解决：事务场景一律用同步Receive()，在同一个线程里完成接收、业务处理、Commit()或Rollback()，别图省事混用。

5. 进阶：把 Demo 改造成可复用的消息组件

Demo 跑通之后，真正要落地还得做一层封装。我一般会把连接、会话、生产消费抽成一个MessageBus类，用连接池管理会话，加上重连和日志。这里给一个最小可用的封装思路，重点在重连和资源释放。

public class MessageBus : IDisposable { private readonly ConnectionFactory _factory; private IConnection _connection; private readonly object _lock = new object(); public MessageBus(string uri, string user, string pwd) { _factory = new ConnectionFactory(uri); _factory.UserName = user; _factory.Password = pwd; // 连接中断时触发，用于重建会话 _factory.ExceptionListener = OnException; Connect(); } private void Connect() { lock (_lock) { _connection = _factory.CreateConnection(); _connection.Start(); } } private void OnException(Exception ex) { // 断线后延迟重连，避免风暴 Thread.Sleep(3000); Connect(); } public void Send(string queueName, string body) { using (var session = _connection.CreateSession(AcknowledgementMode.ClientAcknowledge)) using (var producer = session.CreateProducer(session.GetQueue(queueName))) { producer.DeliveryMode = MsgDeliveryMode.Persistent; producer.Send(session.CreateTextMessage(body)); } } public void Dispose() { _connection?.Close(); _connection?.Dispose(); } }

几个要点：ExceptionListener是重连的入口，但重连后原有的 Session 和 Consumer 都失效了，如果消费者是长驻的，得在重连逻辑里重新注册监听。Send里用using保证 Session 和 Producer 及时释放，ActiveMQ 的会话是有限资源，泄漏了会拖垮 broker。DeliveryMode在封装层固定成 Persistent，业务侧不用关心。

验证封装是否可靠，我有个习惯：写一个压测小脚本，连续发一万条，中途手动重启 broker，看消费端最终收到的条数和发送条数是否一致（允许重复，但不允许丢）。这个测试能同时暴露重连、持久化、确认模式三方面的问题。从那以后我每次封装消息组件，都强制走一遍这个断连重连的验证，比看多少文档都管用。希望这份拆解能帮到你，把 Demo 里的每一行都跑成自己项目里能用的东西。

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