简介这份 ActiveMQ DemoC#资源面向使用 .NET 平台进行消息中间件开发的学习者与工程师重点解决在 WinForm 环境下如何快速跑通 ActiveMQ 消息发送与接收流程的问题。压缩包共 36 个文件约 326KB包含 19 个 cs 源码文件、4 个 resx 资源文件、4 个 dll 依赖库、3 个 exe 可执行程序以及 csproj、sln 等工程配置整体分为生产者与消费者两个独立项目源码结构清晰便于对照理解连接工厂、会话、队列与消息监听等核心环节。目前已有 1292 人学习下载说明该示例在同类入门资料中具备一定参考价值。读者可直接打开解决方案运行调试观察消息从生产端到消费端的完整流转过程并在此基础上修改队列名称、消息内容与监听逻辑快速迁移到自己的业务场景中适合作为 ActiveMQ 与 C# 集成的起步模板。1. 从一次消息丢失排查说起这个 C# 版 ActiveMQ Demo 到底能干什么去年帮一个做订单系统的朋友排查问题现象很邪门用户付完款订单状态偶尔卡在“待处理”后台日志干干净净没有任何异常。查了两天才定位到——他们的消息生产者发完消息就断开连接消费者还没起来消息直接进了黑洞。这种“发出去就不管”的写法在消息队列里就是典型的翻车现场。这份 ActiveMQ DemoC#要解决的就是让 .NET 开发者能在一个可运行的最小工程里把消息的生产、消费、持久化、事务确认这几件事跑通。它适合两类人一是刚接触消息队列、想找个能直接跑起来的 C# 示例的新手二是已经在用 ActiveMQ 但想看看标准写法长什么样的熟手。整个 Demo 围绕一个核心场景展开——模拟订单创建后异步通知库存系统把点对点队列和发布订阅两种模式都覆盖到。2. 技术选型与工程结构为什么是 NMS 而不是原生 JMS2.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.mdProducer 和 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时消息会落盘到 KahaDBBroker 重启不丢设为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(等待消息按 CtrlC 退出...); 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.DeserializeOrderMessage(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 所在机器磁盘告警。原因是所有消息都设了PersistentKahaDB 日志持续增长。解决方式是区分消息重要性非关键消息用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 支持配置DLQDead Letter Queue消息重试超过阈值后自动进死信队列人工介入处理。配置方式是在 Broker 的activemq.xml里设置redeliveryPolicy。第四件事是监控。ActiveMQ 的 JMX 接口能拿到队列深度、消费者数量、入队出队速率等指标接一个 Prometheus 的 JMX Exporter 就能做可视化。关键告警项是队列深度持续增长和消费者数量掉零。验证改造是否成功我一般用三个场景压一是 Broker 重启看生产者和消费者能否自动恢复二是消费者宕机十分钟看消息是否堆积、恢复后能否消费完三是发一万条消息看吞吐和延迟是否在预期内。从那以后我每次接消息队列都强制先跑一遍“发出去、断连接、重启、再消费”的完整链路确认没有静默丢消息才敢上业务。希望这份 Demo 能帮你少走点弯路。本文还有配套的精品资源点击获取