简介ActiveMQ是一款成熟可靠的开源消息中间件在企业级异步通信、系统解耦与流量削峰等场景中被广泛使用。这套基于C#语言、采用WinForm开发的演示程序提供了消息发送端与接收端的完整实现涵盖生产者、消费者、界面交互等核心模块并具备可视化操作界面可运行、可调试非常适合希望快速理解ActiveMQ与.NET整合方式的初中级开发者学习参考。压缩包共有36个文件其中C#源码、界面资源、动态链接库、可执行程序以及项目解决方案等类型一应俱全既可以直接运行观察效果也可以通过阅读代码掌握底层实现整体大小仅326KB轻量便捷。目前已有1292人学习下载作者还撰写了配套的ActiveMQ介绍文章将实践与理论结合有助于深入理解消息传递机制。仔细研读发送与接收两端代码可掌握连接建立、消息发送与异步监听、异常处理以及WinForm界面刷新的关键写法同时项目内模块划分清晰对于实际项目集成ActiveMQ具有不错的参考价值初学者可借此快速建立整体认知。1. ActiveMQ DemoC#解决的是 C# 侧接入消息队列的最后一公里做过 C# 上位机或者 WinForm/WPF 服务端的同学大概率遇到过这样的业务设备采集的数据要发给产线上另外两套系统对方不愿意共享数据库也不想跟你点对点维护 Socket 协议还要求服务端重启之后数据不能丢。这时候ActiveMQ DemoC#就成了一个很实用的着落点——用一套 C# 客户端连上 ActiveMQ完成消息的发送、接收、订阅把系统之间的耦合拆掉。标题里挂 Demo说明目标不是研究消息中间件本身而是快速拿到一段能跑通的最小代码从 NuGet 引入客户端库写好生产者和消费者确认消息能进队列、能出队列。这个方案适合做上位机数据中转、多系统通知同步、离线缓存补偿也是 C# 开发者接触 JMS 生态最现实的入口。下文会把选型、代码、参数和五个高危坑一次讲清楚。2. 选型与最小实现用 NMSActiveMQ 跑通 Queue 收发2.1 为什么 C# 侧首选 NMSActiveMQ协议与客户端库的取舍ActiveMQ 是 Java 写的 Broker原生支持的语言是 JavaC# 这边要接入第一件事是选协议和客户端库。常见做法有三条路走 OpenWire 协议用 Apache.NMS.ActiveMQ走 AMQP 1.0 用 AmqpNetLite走 OpenWire 兼容的第三方封装。对于标题这种 Demo 需求我一般直接选 NMSActiveMQ理由有三个它接收到的消息模型和 JMS 概念一一对应Queue、Topic、Consumer、Producer 都是直接可用的对象Broker 对 OpenWire 的支持是默认开启的不需要在 activemq.xml 里额外加 transportConnector网上能找到的 C# 示例绝大多数也是这套 API遇到问题好搜。如果你很熟悉 Spring Boot 整合 ActiveMQ 的那套写法会发现 NMS 的 API 是它的镜像ConnectionFactory 创建 ConnectionConnection 创建 SessionSession 再创建 MessageProducer 和 MessageConsumer。命名上 NMS 把 JMS 的 createQueue 改成了 GetQueue其余几乎翻译过来就能用。AMQP 1.0 方式更适合你不确定 Broker 是 ActiveMQ 还是别的中间件的场景但 Demo 阶段没必要引入多余抽象。与 Java 侧相比C# 侧的坑主要在依赖和运行环境上。NMSActiveMQ 是一个 NuGet 包老项目用 .NET Framework 4.5 可以直接引用新项目用 .NET 6/8 也兼容。要注意的是它依赖 Apache.NMS 核心包手动拷贝 DLL 时容易漏掉后者建议一律走 NuGet 还原不要从 bin 目录复制。Broker 版本方面5.15 之后都保留 OpenWire 端口风格下载新版解压后启动即可不用关心 Java 端代码长什么样。2.2 发送端与接收端的最小 C# Demo先给一套能直接编译跑起来的最小实现。发送端往队列 demo.queue 里推一条文本消息using System; using Apache.NMS; using Apache.NMS.ActiveMQ; class ProducerDemo { static void Main(string[] args) { // 连接字符串指向 ActiveMQ 默认 OpenWire 端口 var factory new ConnectionFactory(tcp://127.0.0.1:61616); using (IConnection connection factory.CreateConnection()) using (ISession session connection.CreateSession(AcknowledgementMode.AutoAcknowledge)) { // 声明队列Broker 上没有时会自动创建 IDestination destination session.GetQueue(demo.queue); using (IMessageProducer producer session.CreateProducer(destination)) { producer.DeliveryMode MsgDeliveryMode.Persistent; ITextMessage message session.CreateTextMessage(hello from c#); producer.Send(message); Console.WriteLine(sent); } } } }这段代码的逻辑是创建连接工厂建立 Connection再创建 SessionSession 负责生成消息和生产者。GetQueue 是幂等的Broker 会根据名字自动建队列所以代码里不需要预先管理队列。DeliveryMode 设置成 Persistent 表示消息要落盘Broker 重启后还能读出来如果只是临时通知可以改成 NonPersistent 换取吞吐。再看接收端消费端和发送端最大的区别是必须调用connection.Start()using System; using Apache.NMS; using Apache.NMS.ActiveMQ; class ConsumerDemo { static void Main(string[] args) { var factory new ConnectionFactory(tcp://127.0.0.1:61616); using (IConnection connection factory.CreateConnection()) { // 启动连接NMS 默认不自动开始消费线程 connection.Start(); using (ISession session connection.CreateSession(AcknowledgementMode.AutoAcknowledge)) { IDestination destination session.GetQueue(demo.queue); using (IMessageConsumer consumer session.CreateConsumer(destination)) { consumer.Listener message { if (message is ITextMessage text) Console.WriteLine(text.Text); }; Console.WriteLine(waiting...); Console.ReadLine(); } } } } }这里最容易踩的坑是没有调用 Start导致 Listener 永远不触发。Start 的作用是把连接置为可接收状态Session 创建好并不代表消费线程已经跑起来。Listener 模型下每条消息都会在回调里被处理如果处理逻辑抛异常消息会被视为消费失败触发重投逻辑所以回调内部最好包一层 try-catch。2.3 必调的 5 个参数与默认值说明Demo 跑通之后接下来要关心的是几个直接影响可靠性、吞吐和排错难度的参数。我在实际项目里几乎每次都要调一遍整理成一张表参数位置默认行为建议设置DeliveryModeproducer非持久核心对账消息设为 PersistentAcknowledgementModesessionAutoAcknowledge手动管理消息时改为 ClientAcknowledgeTimeToLiveproducer0 代表不过期按消息有效期限设置超时的自动进死信queuePrefetchbrokerUri 或 consumer默认 1000 条小消息大批量场景调低到 100~500startupMaxReconnectAttemptsfailover 连接串按连接 URL 决定要求快速失败时显式设成 5关于TimeToLive你可以在 Send 时传参数也可以直接在 producer 上设置。它的默认值是 0表示永不过期这在 Demo 里没问题但生产环境里积压的脏数据会一直占队列。queuePrefetch是 Consumer 一次性从 Broker 拉取的消息条数默认 1000 是为了省网络往返如果你的消费逻辑慢消息积压在客户端本地Broker 端 Dequeue 计数却不涨排查起来很诡异这时候把 prefetch 调低反而容易观察。还有一点容易被忽略NMS 的 AutoAcknowledge 是在消息交给 Listener 之前就确认了也就是说回调里崩溃也会丢消息。如果要做到「处理完才确认」Session 要用 ClientAcknowledge并在回调末尾手动调用message.Acknowledge()。这是从 Demo 走向可靠消费必须迈过的一道坎。3. 从点到面Topic 发布订阅与持久化订阅的落地写法3.1 Queue 与 Topic 在 C# Demo 里的实际差异Queue 是一对一的点对点模型一条消息只会被一个消费者消费消费完就从队列中移除。Topic 是发布订阅模型消息会被推给所有在线的订阅者。很多 C# Demo 项目在最初只用 Queue等接到「一个数据源要广播给多个展示端」的需求时就会碰到 Topic。差异用一张表说清楚对比项QueueTopic消息去向一个消费者竞争消费所有订阅者都会收到离线消费者上线后可以补消费非持久订阅者收不到离线期间消息持久化持久模式与 Broker 重启相关需要配合持久订阅C# APIsession.GetQueue(name)session.GetTopic(name)适用场景任务分配、点对点通知广播、多端同步在 NMS 的 API 上Queue 和 Topic 都是IDestination只在创建时用 GetQueue 还是 GetTopic 区分。发送端的代码几乎完全一样所以很多人会在切换时搞混把一个队列消息发给 Topic或者反过来导致消息「消失」。第一条排查原则就是先确认生产和消费两端用的是同一个 Destination 类型和同一个名字。Topic 还有一个特点需要记住普通订阅者在线才收消息离线期间的消息不会补发如果要让断线重连后的节点也能收到离线消息必须用持久化订阅。3.2 Topic 发布与订阅的 C# 代码发布端把 GetQueue 换成 GetTopic 即可其他逻辑不变using (ISession session connection.CreateSession(AcknowledgementMode.AutoAcknowledge)) { IDestination topic session.GetTopic(demo.topic); using (IMessageProducer producer session.CreateProducer(topic)) { producer.DeliveryMode MsgDeliveryMode.NonPersistent; producer.Send(session.CreateTextMessage(broadcast)); } }订阅端要注意普通订阅用的是CreateConsumer但语义上已经不是队列消费using (IConnection connection factory.CreateConnection()) { connection.Start(); using (ISession session connection.CreateSession(AcknowledgementMode.AutoAcknowledge)) { IDestination topic session.GetTopic(demo.topic); using (IMessageConsumer consumer session.CreateConsumer(topic)) { consumer.Listener message { if (message is ITextMessage text) Console.WriteLine($[subscriber] {text.Text}); }; Console.ReadLine(); } } }这个订阅者只要不退出就能收到所有推送到 demo.topic 的消息一旦进程退出离线期间的消息就永久丢失。如果你用两个订阅者进程同时跑会观察到每条广播消息两台都能收到这就是 Topic 和 Queue 最大的行为差异。调试时可以把发送端改成循环发 10 条然后观察两个进程的打印次数以此确认模型确实按预期工作。3.3 持久化订阅重启后消息不丢的关键配置要让订阅者在离线期间也能收到消息需要把普通消费者改成持久化订阅。NMS 里对应的 API 是CreateDurableConsumer有三个关键点Connection 上必须设置 ClientIdTopic 的名字要固定订阅名要唯一。代码长这样using (IConnection connection factory.CreateConnection()) { // ClientId 是持久订阅的标志Broker 靠它识别离线订阅者 connection.ClientId subscriber-a; connection.Start(); using (ISession session connection.CreateSession(AcknowledgementMode.AutoAcknowledge)) { IDestination topic session.GetTopic(demo.topic); // 最后一个参数 noLocalfalse允许接收自己发布的消息 using (IMessageConsumer consumer session.CreateDurableConsumer(topic, subscription-a, null, false)) { consumer.Listener message { if (message is ITextMessage text) Console.WriteLine(text.Text); }; Console.ReadLine(); } } }第二次运行这个程序时Broker 会识别到 ClientId 和订阅名相同的消费者恢复之前的离线消息。这里常见的翻车点是换了一台机器跑ClientId 却一样导致 Broker 认为旧节点还占着订阅新节点抢不到消费反过来同一台机器重复启动订阅名不一致又会创建多个离线订阅消息被重复投递。持久化订阅在 Broker 端是有存储成本的离线时间越长积压越多。生产项目里建议在管理控制台定期检查 Topic 的订阅者数量和待消费消息数超过阈值的离线订阅要考虑主动清理否则 Broker 内存会被积压消息拖爆。C# 侧的代码本身只管发和收这部分运维观察要补上。3.4 重连与失效转移failover 协议参数怎么设单机 Demo 用tcp://127.0.0.1:61616没问题但一旦 Broker 重启C# 客户端如果还握着旧连接后面 Send 会抛Connection refused或Broker not available。生产项目里我优先把连接字符串改成 failover 协议var factory new ConnectionFactory( failover:(tcp://127.0.0.1:61616)?startupMaxReconnectAttempts5maxReconnectAttempts-1);failover 的意思是当底层连接断开时客户端自动按 URL 列表重连期间调用的 Send 会阻塞等待恢复。括号里可以写多个地址用逗号分隔Broker 挂了自动换下一个。两个参数要理解清楚startupMaxReconnectAttempts是启动阶段的最大重连次数超过就抛异常适合需要快速失败的场景maxReconnectAttempts-1是无限重连适合常驻服务。连接串还可以追加 transport 参数比如socketTimeout、connectionTimeout和keepAlive。对 C# 上位机来说我一般会设connectionTimeout3000让连接失败在 3 秒内返回而不是默认的无限等待。NMS 客户端默认有心跳检测长时间没有消息往来时会发送 keep-alive所以不需要在业务代码里做额外的心跳。4. 避坑ActiveMQ Demo 在 C# 侧最常见的 5 个问题4.1 8161 控制台访问不了现象是浏览器里打开 http://localhost:8161/admin 能进换成局域网 IP 就超时或者干脆提示无权访问。原因通常有两个ActiveMQ 的 Web Console 默认绑定在 127.0.0.1只允许本机访问新版比如 5.18 分支对默认账号口令有调整下载解压后要先看 conf 目录下的 users.properties 和 jetty-realm.properties。解决方法是改conf/jetty.xml把当前监听地址 127.0.0.1 改成 0.0.0.0保存后重启服务如果只是想本机看队列数据直接访问 http://localhost:8161/admin 并用配置文件里的账号登录即可。注意8161 是管理端口不是消息端口改监听范围时不要顺手把 61616 也改没了。4.2 连不上 61616 端口但 broker 明明在跑现象是 C# 侧抛Could not connect to broker URL: tcp://...但你在服务器上能看到 activemq 进程。原因很常见Broker 只监听了 localhost或 Windows 防火墙拦截了 61616或 activemq.xml 里 transportConnector 的名字和端口被改过。排查顺序我一般固定为先用Test-NetConnection 目标IP -Port 61616确认 TCP 层通不通再登录 Broker 所在机器用netstat -ano | findstr 61616看监听 IP 是不是 0.0.0.0最后打开 conf/activemq.xml 确认transportConnector的 uri 里写的是不是 61616。C# 客户端连接串里的主机名必须能被解析到 Broker 网卡不要写 localhost 去连远程服务器。4.3 消息在重启后全部丢失现象是发送端显示 Send 成功Broker 一重启队列空了。原因多数是发送时没设置持久化DeliveryMode 默认不是 Persistent或者 Broker 的 data/kahadb 目录被手工清掉。解决要分两层发送端设置producer.DeliveryMode MsgDeliveryMode.PersistentBroker 端确认 activeMQ.xml 的持久化适配器是默认的 kahadb不要为了图省事改成内存版。注意持久化不等于不丢消息——持久消息在 Broker 重启后还在但非持久订阅的 Topic 消费者离线期间收不到是正常的。想验证你的消息到底有没有落盘可以去 data/kahadb 目录看日志文件大小是否变化同时结合 Web 控制台的 Enqueue 计数判断。4.4 程序退出时连接不释放现象是 C# Console 程序执行完 Main 方法进程却不结束或者关掉窗体后 task manager 里还有残留进程。原因是 IConnection 对象没有显式关闭NMS 内部的背景线程和连接池在等待重连或心跳。解决方法是把连接、会话、生产者都放进 using 语句块或者显式调用connection.Close()。有 Listener 订阅的进程要特别注意Listener 回调跑在独立线程上即使主线程结束了回调线程可能还活着。我一般会加一个取消令牌在退出逻辑里先置位令牌再 Close 连接最后等回调线程 join 超时退出。看起来是细枝末节但上位机程序要长期开机连接泄漏最终会把系统资源吃干。4.5 接收不到消息但发送成功现象是生产端代码不报错Web 控制台也能看到入队消息但消费端一个消息都收不到。原因先按三种排查生产者发的是 Topic消费者订阅的是 Queue名字再相同也白搭消费端没有调用connection.Start()AutoAcknowledge 模式下 Broker 把消息推给消费者时已经标记确认如果回调里抛异常且没有 catch消息就神秘消失。解决方式是先用 Web 控制台确认 Destination 类型和名字再看消费者进程日志里有没有抛异常最后把 AcknowledgeMode 改为 ClientAcknowledge在回调最后手动确认。这个坑越早踩越划算因为换到项目后期消息链路一长这种基础错位反而最难定位。5. 验证技巧用控制台、日志和计数把 Demo 装进交付包Demo 写完先别急着对接业务按下面三步验证一下能省掉后面大量的联调时间。第一步是「对账式验证」开两个进程分别跑生产者和消费者用控制台发 10 条带序号的文本消息消费者打印时也打印序号判断有没有乱序和丢失再把生产者的 DeliveryMode 改成 Persistent重启 Broker确认重启后消息能重新被消费。第二步是「队列计数验证」打开 8161 管理页的 Queues 标签观察 enqueue 和 dequeue 两个计数发送 10 条后 enqueue 应增加 10消费后 dequeue 同步增长如果 dequeue 不动说明消费者没起来或者 Ack 有问题。Topic 场景则切到 Topics 页看订阅者列表持久订阅会显示离线节点。第三步是「单位时间吞吐估算」用Stopwatch给 Listener 回调计时往队列里灌 10000 条小消息看客户端多久消费完这里要注意 prefetch 默认值可能让客户端一次拉 1000 条本地内存里堆着没处理统计出来的不是实时消费能力。把消费逻辑里最耗时的操作数据库写入、文件写入、网络调用单独空跑一次得出单条处理下限再回推这个客户端适合抗多少消息量。这样得到的结论放进交接文档里比任何口头描述都有说服力。调试 C# 侧 NMS 问题时还有一个值得养成的习惯在 App.config 里临时开启 Apache.NMS 的日志输出可以看到客户端和 Broker 之间的连接状态、消息确认时序。线上项目里日志级别保持 Info排查问题时再切成 Debug避免日志刷爆磁盘。我自己的惯例是每接入一个 ActiveMQ Broker就把「Broker 配置快照」「C# 客户端版本」「三条验证命令」写进项目的 README下次换机器重装环境时照着执行一遍就能确认环境健康。希望这些经验能帮你把 Demo 顺利推上线少走几趟查连接的弯路。本文还有配套的精品资源点击获取