简介：这份 ActiveMQ Demo（C#）资源面向使用 .NET 平台进行消息中间件开发的学习者与工程师，重点解决在 WinForm 环境下如何快速跑通 ActiveMQ 消息发送与接收流程的问题。压缩包共 36 个文件，约 326KB，包含 19 个 cs 源码文件、4 个 resx 资源文件、4 个 dll 依赖库、3 个 exe 可执行程序以及 csproj、sln 等工程配置，整体分为生产者与消费者两个独立项目，源码结构清晰，便于对照理解连接工厂、会话、队列与消息监听等核心环节。目前已有 1292 人学习下载，说明该示例在同类入门资料中具备一定参考价值。读者可直接打开解决方案运行调试，观察消息从生产端到消费端的完整流转过程，并在此基础上修改队列名称、消息内容与监听逻辑，快速迁移到自己的业务场景中，适合作为 ActiveMQ 与 C# 集成的起步模板。

1. 从一次消息丢失排查说起：这个 C# 版 ActiveMQ Demo 到底能干什么

去年帮一个做订单系统的朋友排查问题，现象很邪门：用户付完款，订单状态偶尔卡在“待处理”，后台日志干干净净，没有任何异常。查了两天才定位到——他们的消息生产者发完消息就断开连接，消费者还没起来，消息直接进了黑洞。这种“发出去就不管”的写法，在消息队列里就是典型的翻车现场。

这份 ActiveMQ Demo（C#）要解决的，就是让 .NET 开发者能在一个可运行的最小工程里，把消息的生产、消费、持久化、事务确认这几件事跑通。它适合两类人：一是刚接触消息队列、想找个能直接跑起来的 C# 示例的新手；二是已经在用 ActiveMQ 但想看看标准写法长什么样的熟手。整个 Demo 围绕一个核心场景展开——模拟订单创建后异步通知库存系统，把点对点队列和发布订阅两种模式都覆盖到。

2. 技术选型与工程结构：为什么是 NMS 而不是原生 JMS

2.1 ActiveMQ 在 .NET 生态里的接入方式

ActiveMQ 是 Java 写的，但提供了跨语言的客户端协议。C# 这边接入 ActiveMQ，主流做法是走 Apache.NMS（.NET Messaging Service）这套 API。NMS 是 JMS 在 .NET 上的对应实现，接口设计几乎一一对应，比如IConnectionFactory、ISession、IMessageProducer、IMessageConsumer，熟悉 JMS 的人切过来基本零成本。

这个 Demo 用的就是 Apache.NMS.ActiveMQ 这个包。它底层走 OpenWire 协议，和 Java 端的 ActiveMQ Broker 直接通信，不需要额外装什么桥接服务。常见做法是在 NuGet 里装两个包：Apache.NMS提供接口抽象，Apache.NMS.ActiveMQ提供具体实现。版本上建议锁一个稳定版，别追最新，NMS 的版本兼容性偶尔会有玄学问题。

2.2 工程目录一览

Demo 的目录结构不复杂，但该分的层都分开了：

ActiveMQDemo/ ├── ActiveMQDemo.sln ├── src/ │ ├── ActiveMQDemo.Producer/ # 生产者控制台 │ │ ├── Program.cs │ │ └── Services/ │ │ └── OrderMessageService.cs │ ├── ActiveMQDemo.Consumer/ # 消费者控制台 │ │ ├── Program.cs │ │ └── Handlers/ │ │ └── OrderMessageHandler.cs │ └── ActiveMQDemo.Shared/ # 公共模型与配置 │ ├── Models/ │ │ └── OrderMessage.cs │ └── Config/ │ └── MqConfig.cs └── README.md

Producer 和 Consumer 是两个独立控制台程序，Shared 放消息实体和连接配置。这种拆法在 Demo 里够用，实际项目里我会把 Shared 换成独立的类库项目，方便被 Web API 或 Worker Service 引用。

2.3 消息模型定义

消息体用一个简单的订单模型，字段不多但覆盖了典型场景：

// Shared/Models/OrderMessage.cs namespace ActiveMQDemo.Shared.Models { public class OrderMessage { public string OrderId { get; set; } // 订单号 public string ProductName { get; set; } // 商品名 public int Quantity { get; set; } // 数量 public DateTime CreatedAt { get; set; } // 创建时间 public string Action { get; set; } // 动作类型：Create/Cancel } }

OrderId是业务主键，消费端靠它做幂等；Action字段用来区分是创建还是取消，这样一条队列可以承载多种业务动作，不用为每个动作单独建队列。实际项目里我一般还会加一个TraceId，方便全链路追踪。

2.4 连接配置的集中管理

连接字符串不硬编码在业务代码里，抽到MqConfig：

// Shared/Config/MqConfig.cs namespace ActiveMQDemo.Shared.Config { public static class MqConfig { // ActiveMQ Broker 地址，默认端口 61616 public const string BrokerUri = "tcp://localhost:61616"; // 队列名：点对点模式 public const string OrderQueue = "order.queue"; // 主题名：发布订阅模式 public const string OrderTopic = "order.topic"; // 用户名密码，默认 admin/admin public const string UserName = "admin"; public const string Password = "admin"; } }

BrokerUri的协议头是tcp://，这是 OpenWire 的默认传输方式。如果 Broker 开了 SSL，这里要换成ssl://并配证书。队列名和主题名分开定义，避免拼写错误导致消息发到错误的目的地——这种错误在日志里往往看不出来，因为消息确实发出去了，只是没人消费。

3. 生产者实现：从连接建立到消息确认的完整链路

3.1 建立连接与会话

生产者的核心是拿到一个ISession，所有的消息发送都通过它。连接建立的过程有几步固定动作：

// Producer/Services/OrderMessageService.cs using Apache.NMS; using Apache.NMS.ActiveMQ; using ActiveMQDemo.Shared.Config; using ActiveMQDemo.Shared.Models; using System.Text.Json; namespace ActiveMQDemo.Producer.Services { public class OrderMessageService : IDisposable { private readonly IConnectionFactory _factory; private readonly IConnection _connection; private readonly ISession _session; public OrderMessageService() { // 1. 创建连接工厂 _factory = new ConnectionFactory(MqConfig.BrokerUri); // 2. 建立连接（传入用户名密码） _connection = _factory.CreateConnection( MqConfig.UserName, MqConfig.Password); // 3. 启动连接，必须调用，否则无法收发消息 _connection.Start(); // 4. 创建会话，AcknowledgeMode 决定确认模式 _session = _connection.CreateSession( AcknowledgementMode.AutoAcknowledge); } // ... 发送方法见下 } }

这里有几个参数值得说清楚。AcknowledgementMode有四种取值：AutoAcknowledge是收到就自动确认，性能最好但可能丢消息；ClientAcknowledge需要手动调Acknowledge()，适合对可靠性要求高的场景；Transacted走事务，配合Commit()和Rollback()使用；DupsOkAcknowledge是延迟确认，允许重复但不丢。Demo 默认用AutoAcknowledge，方便新手先跑通，实际生产环境我会根据业务选ClientAcknowledge或Transacted。

_connection.Start()这一步经常被漏掉。NMS 的连接创建后处于“停止”状态，不调Start()消息发不出去，而且不会报错，就是静默失败。这个坑我踩过不止一次。

3.2 发送消息到队列

发送逻辑本身不复杂，关键是把对象序列化和消息属性设置做对：

public void SendToQueue(OrderMessage order) { // 创建目标队列 IDestination destination = _session.GetQueue(MqConfig.OrderQueue); // 创建生产者 using IMessageProducer producer = _session.CreateProducer(destination); // 设置持久化模式：Persistent 表示 Broker 重启后消息还在 producer.DeliveryMode = MsgDeliveryMode.Persistent; // 对象序列化为 JSON 文本 string json = JsonSerializer.Serialize(order); ITextMessage message = _session.CreateTextMessage(json); // 设置自定义属性，消费端可据此过滤 message.Properties["Action"] = order.Action; message.Properties["CreatedAt"] = order.CreatedAt.ToString("o"); // 发送 producer.Send(message); Console.WriteLine($"[发送] 订单 {order.OrderId} 已入队"); }

DeliveryMode设为Persistent时，消息会落盘到 KahaDB，Broker 重启不丢；设为NonPersistent则只在内存里，性能高但重启即失。Demo 里订单消息显然要用持久化。message.Properties是自定义属性，消费端可以用选择器（Selector）只消费特定Action的消息，比如Action = 'Create'，这样一条队列能分流给不同消费者处理。

3.3 发布订阅模式的发送

主题的发送和队列几乎一样，区别在GetTopic和消息的“广播”语义：

public void PublishToTopic(OrderMessage order) { IDestination destination = _session.GetTopic(MqConfig.OrderTopic); using IMessageProducer producer = _session.CreateProducer(destination); // 主题消息通常不持久化，因为订阅者可能不在线 producer.DeliveryMode = MsgDeliveryMode.NonPersistent; string json = JsonSerializer.Serialize(order); ITextMessage message = _session.CreateTextMessage(json); producer.Send(message); Console.WriteLine($"[发布] 订单 {order.OrderId} 已广播"); }

主题和队列最大的区别是：队列的消息只要有一个消费者拿走就没了，主题的消息每个活跃订阅者都会收到一份。如果订阅者不在线，非持久化主题消息直接丢弃。要做持久订阅，需要给消费者设置ClientId和SubscriptionName，这个在消费者章节展开。

3.4 资源释放的正确姿势

IConnection、ISession都实现了IDisposable，释放顺序有讲究：

public void Dispose() { _session?.Close(); _session?.Dispose(); _connection?.Close(); _connection?.Dispose(); }

先关 Session 再关 Connection，反过来会抛异常。Close()和Dispose()都调一遍是保险做法，NMS 在某些版本里Dispose()不会自动Close()。用using包起来也行，但连接对象通常生命周期较长，手动管理更灵活。

4. 消费者实现：监听、确认与异常处理

4.1 同步接收与异步监听

消费者有两种收消息的方式。同步用Receive()，阻塞等待；异步用事件监听，消息到了自动触发回调。Demo 两种都演示了，先看同步：

// Consumer/Program.cs using Apache.NMS; using Apache.NMS.ActiveMQ; using ActiveMQDemo.Shared.Config; var factory = new ConnectionFactory(MqConfig.BrokerUri); using IConnection connection = factory.CreateConnection( MqConfig.UserName, MqConfig.Password); connection.Start(); using ISession session = connection.CreateSession( AcknowledgementMode.ClientAcknowledge); IDestination queue = session.GetQueue(MqConfig.OrderQueue); using IMessageConsumer consumer = session.CreateConsumer(queue); Console.WriteLine("等待消息，按 Ctrl+C 退出..."); while (true) { // 阻塞 1 秒，超时返回 null IMessage message = consumer.Receive(TimeSpan.FromSeconds(1)); if (message == null) continue; if (message is ITextMessage textMessage) { Console.WriteLine($"[收到] {textMessage.Text}"); // 手动确认，告诉 Broker 这条消息处理完了 textMessage.Acknowledge(); } }

Receive传TimeSpan是设超时，不传则无限阻塞。用ClientAcknowledge模式时，必须显式调Acknowledge()，否则消息会被 Broker 认为未消费，连接断开后重新投递。这个机制保证了“至少一次”投递，但也意味着消费端要做幂等。

4.2 异步监听器的写法

异步方式更适合常驻服务：

using IMessageConsumer consumer = session.CreateConsumer(queue); // 注册消息到达事件 consumer.Listener += (IMessage message) => { try { if (message is ITextMessage textMessage) { var order = JsonSerializer.Deserialize<OrderMessage>(textMessage.Text); Console.WriteLine($"[处理] 订单 {order.OrderId}，动作 {order.Action}"); // 业务处理... textMessage.Acknowledge(); } } catch (Exception ex) { Console.WriteLine($"[异常] {ex.Message}"); // 这里可以选择 Nack 或让消息重投 } }; Console.WriteLine("监听中..."); Console.ReadKey();

Listener是事件回调，运行在 NMS 的内部线程上。回调里不要做耗时操作，否则会阻塞后续消息的接收。重活应该丢到线程池或独立的处理队列。异常一定要在回调内部捕获，抛出去会导致监听器断开，而且不一定有日志。

4.3 持久订阅的实现

主题的持久订阅需要两个标识：

connection.ClientId = "order-consumer-01"; // 客户端唯一标识 IDestination topic = session.GetTopic(MqConfig.OrderTopic); // 创建持久订阅者，SubscriptionName 是订阅名 using IMessageConsumer consumer = session.CreateDurableConsumer( topic, "order-subscription", null, false);

ClientId在同一个 Broker 上必须唯一，重复会连接失败。SubscriptionName标识这个订阅，消费者离线期间 Broker 会把消息存下来，等它回来再投递。注意持久订阅的消息默认是持久化的，Broker 磁盘会持续增长，要定期清理不用的订阅。

4.4 消息选择器过滤

如果只想消费特定类型的消息，用选择器：

// 只接收 Action 为 Create 的消息 using IMessageConsumer consumer = session.CreateConsumer( queue, "Action = 'Create'");

选择器语法类似 SQL 的 WHERE，支持=、<>、>、<、AND、OR、IN等。属性名区分大小写，值用单引号。这个功能在多消费者分流时很有用，比如一个队列放所有订单消息，库存消费者只关心Create，取消消费者只关心Cancel。

5. 避坑与常见问题排查

5.1 消息发出去了但消费者收不到

现象：生产者日志显示发送成功，消费者端毫无反应。原因通常是连接没Start()，或者队列名不一致。NMS 的连接创建后默认是停止状态，必须调_connection.Start()。队列名大小写敏感，order.queue和Order.Queue是两个不同的队列。排查方法是在 ActiveMQ 管理控制台（默认 8161 端口）看队列的 Enqueue 和 Dequeue 计数，如果 Enqueue 有数 Dequeue 为零，说明消息进了队列但没人消费。

5.2 消费者重启后消息重复消费

现象：同一条消息被处理了两次。原因是用了AutoAcknowledge或ClientAcknowledge但确认时机不对——消息处理到一半连接断了，Broker 认为没确认，重投。解决方式是消费端做幂等，用OrderId做唯一键，处理前先查是否已处理过。或者改用Transacted模式，业务处理和Commit()放在一起，要么都成功要么都回滚。

5.3 连接数暴涨导致 Broker 拒绝服务

现象：运行一段时间后报Too many connections。原因是每次发消息都新建连接，用完没释放。正确做法是连接和会话复用，一个应用维护一个IConnection，发消息时只创建IMessageProducer并using释放。生产者对象是轻量的，连接是重量的，别搞反了。

5.4 持久化消息导致磁盘写满

现象：Broker 所在机器磁盘告警。原因是所有消息都设了Persistent，KahaDB 日志持续增长。解决方式是区分消息重要性，非关键消息用NonPersistent；同时配置 Broker 的storeUsage上限和日志清理策略。Demo 里订单消息必须持久化，但通知类消息可以放宽。

5.5 中文消息乱码

现象：消费端收到的中文变成问号。原因是消息编码不一致。NMS 的CreateTextMessage默认用 UTF-8，但如果 Broker 端配置了其他编码，或者消息在传输中被转换，就会乱码。解决方式是在发送前显式指定编码，或者干脆把对象序列化成 Base64 再传。我一般直接用JsonSerializer序列化成 UTF-8 字符串，目前没遇到过乱码。

6. 二次开发改造：从 Demo 到生产级消息组件

Demo 跑通只是第一步，真正用到项目里还得做几件事。我把改造思路按优先级列一下。

第一件事是连接管理。Demo 里每个服务自己管连接，实际项目应该抽一个IMqConnectionFactory单例，用依赖注入注册。连接要加心跳和自动重连，NMS 的ConnectionFactory有OnException事件，可以在里面做重连逻辑。我一般会包一层ResilientConnection，断线后指数退避重试。

第二件事是消息序列化。Demo 用System.Text.Json，够用但不够灵活。生产环境建议定义一个IMessageSerializer接口，支持 JSON、MessagePack 甚至 Protobuf 切换。消息体里加Version字段，方便后续做消息格式升级时的兼容处理。

第三件事是消费端的幂等和重试。幂等靠业务唯一键加去重表，重试靠死信队列。ActiveMQ 支持配置DLQ（Dead Letter Queue），消息重试超过阈值后自动进死信队列，人工介入处理。配置方式是在 Broker 的activemq.xml里设置redeliveryPolicy。

第四件事是监控。ActiveMQ 的 JMX 接口能拿到队列深度、消费者数量、入队出队速率等指标，接一个 Prometheus 的 JMX Exporter 就能做可视化。关键告警项是队列深度持续增长和消费者数量掉零。

验证改造是否成功，我一般用三个场景压：一是 Broker 重启，看生产者和消费者能否自动恢复；二是消费者宕机十分钟，看消息是否堆积、恢复后能否消费完；三是发一万条消息，看吞吐和延迟是否在预期内。

从那以后我每次接消息队列，都强制先跑一遍“发出去、断连接、重启、再消费”的完整链路，确认没有静默丢消息才敢上业务。希望这份 Demo 能帮你少走点弯路。

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