简介本资源是一套基于C#开发的ActiveMQ消息中间件实战演示程序面向.NET开发者及消息队列初学者用于快速理解点对点与发布订阅模式下的消息收发原理与WinForm集成实践。压缩包共36个文件含19个核心C#源码如MQDemoProducer.cs、Form1.cs、MQ.cs等、4个资源文件.resx、3个可执行程序.exe及2个Visual Studio解决方案文件.sln、.csproj辅以DLL依赖与配置文件完整呈现从连接建立、消息发送到界面化接收的全流程实现包体仅326KB轻量易上手。已有1292人学习下载适合希望在Windows桌面应用中嵌入消息通信能力的开发者可直接运行调试、分析目录模块划分逻辑含生产者/消费者分离结构、ListView消息展示控件封装、全局消息处理函数等并结合作者技术博客深入理解ActiveMQ在.NET生态中的典型用法。1. 一个跑在 Windows 上的 ActiveMQ 消息收发 DemoC# 开发者拿来就能测、改三行就接入自己项目的轻量级通信样板你刚接手一个老系统改造任务后端是 .NET Framework 4.6前端是 WinForm现在要加个「订单状态实时推送」功能——不许动现有数据库不能引入新中间件但得让多个客户端秒级收到变更。这时候翻文档查 RabbitMQ 客户端兼容性、配 TLS、调证书三天都搭不起来。而这个 ActiveMQ DemoC#就是为这种场景生的它用 Apache.NMS 1.7.2 ActiveMQ 5.18 原生协议直连不依赖 Java 环境Win10/Win11 双击start-broker.bat就拉起嵌入式 BrokerC# 控制台项目双击Producer.exe和Consumer.exe就能看见消息一发一收。它不是教学玩具而是某高校实验室做设备监控系统时压测过 5000 TPS 的通信底座原型没有 Spring.NET 那套复杂配置也没有 JSON Schema 校验、死信队列、事务回滚等“企业级幻觉”就干一件事把字符串或简单对象塞进队列另一端原样吐出来。适合 C# 中级开发者快速验证消息模型、补全 WinForm/WPF 应用的异步解耦能力也适合 .NET Core 迁移前的过渡期做协议探路。2. 环境准备与运行实录从零启动 Broker 到看到第一条 “Hello World” 消息2.1 下载与目录结构说明看清这 4 个核心文件的作用项目解压后共 12 个文件真正驱动运行的是以下 4 个activemq.xmlBroker 的配置文件定义了监听端口默认 61616、持久化路径./data、是否启用 Web 控制台默认关闭start-broker.batWindows 批处理脚本执行java -jar activemq-all-5.18.0.jar启动嵌入式 BrokerProducer.exeC# 控制台程序使用 Apache.NMS 创建 Connection、Session、Producer向queue://test.queue发送文本Consumer.exeC# 控制台程序同样基于 NMS订阅同一队列并阻塞接收提示Producer.exe和Consumer.exe已编译为 x64 平台目标若在旧版 Win7 或纯 x86 环境运行失败请用 Visual Studio 打开src/Producer/Producer.csproj将PlatformTargetx64/PlatformTarget改为AnyCPU后重新生成。其余文件中lib/目录下是 Apache.NMS 1.7.2 的 DLLApache.NMS.dll,Apache.NMS.ActiveMQ.dllsrc/是完整 VS2019 解决方案含两个项目和一份README.md含连接字符串格式说明。整个包不含任何安装程序或注册表操作所有数据落盘在./data子目录删掉即重置。2.2 启动 ActiveMQ Broker不装服务、不配环境变量的极简方式ActiveMQ 5.18 自带activemq-all-5.18.0.jar已打包进项目根目录。启动只需一步# start-broker.bat 内容可直接复制到命令行执行 echo off java -Xms512M -Xmx1G -Dorg.apache.activemq.UseDedicatedTaskRunnertrue -Djava.util.logging.config.filelogging.properties -jar activemq-all-5.18.0.jar pause执行后会看到类似输出INFO | Apache ActiveMQ 5.18.0 (localhost, ID:DESKTOP-ABC123-50123-1712345678901-0:1) is starting INFO | Listening for connections at: tcp://0.0.0.0:61616?maximumConnections1000wireFormat.maxFrameSize1073741824 INFO | Connector openwire started关键点有三tcp://0.0.0.0:61616是默认监听地址局域网内其他机器可通过tcp://192.168.x.x:61616连接openwire是 ActiveMQ 原生二进制协议NMS 客户端默认走此协议性能优于 STOMP日志输出到控制台无需额外配置 log4net —— 这正是该 Demo 轻量化的体现。注意若提示java is not recognized说明未安装 JDK 8 或以上版本。此时不要去官网下 JDK 21 —— ActiveMQ 5.18 官方仅支持 JDK 8–11。推荐下载 Adoptium Temurin JDK 11 安装后将JAVA_HOME指向其根目录并把%JAVA_HOME%\bin加入PATH。2.3 运行 Producer 与 Consumer观察消息生命周期的三阶段先保持start-broker.bat窗口开着不要关新开两个命令行窗口# 窗口1启动消费者会阻塞等待消息 Consumer.exe # 窗口2启动生产者发送后立即退出 Producer.exe你会看到 Consumer 窗口打印[Consumer] Connected to tcp://localhost:61616 [Consumer] Subscribed to queue://test.queue [Consumer] Received: Hello World! Sent at 2024-04-05T10:20:30Producer 窗口打印[Producer] Connected to tcp://localhost:61616 [Producer] Sent: Hello World! Sent at 2024-04-05T10:20:30 [Producer] Disconnected.这就是一条消息的完整链路Producer 创建 Connection → Session → MessageProducer → Send() → Broker 入队 → Consumer 创建 Connection → Session → MessageConsumer → Receive() → 返回 IMessage 对象。整个过程无序列化框架介入IMessage的Text属性直接映射为 UTF-8 字符串这也是它能跑得快、改得快的根本原因。2.4 验证连接可用性用 telnet 和 JConsole 辅助定位网络层问题当 Consumer 一直卡在Subscribed...不出消息或 Producer 报Connection refused别急着改 C# 代码 —— 先确认 TCP 层通不通# 在 Producer/Consumer 所在机器执行非 Broker 机器 telnet localhost 61616 # 若显示黑屏或光标闪烁说明端口可达若报“无法连接”说明 Broker 未启动或防火墙拦截 # 查看 Broker 进程是否存活Windows tasklist /fi imagename eq java.exe | findstr activemq # 应返回类似java.exe 12344 Console 1 12,345 K更进一步可用 JDK 自带的jconsole连接 Broker 的 JMX 接口默认开启端口 1099jconsole localhost:1099 # 连上后选择「MBeans」→「org.apache.activemq」→「BrokerNamelocalhost」→「Queues」 # 展开后能看到 test.queue 的 EnqueueCount入队数、DequeueCount出队数、QueueSize当前积压这是比日志更准的“消息健康度仪表盘”。如果EnqueueCount1但DequeueCount0说明 Consumer 没真正消费成功可能异常退出或未调用Start()如果两者相等但QueueSize0说明消息已流转完毕 —— 这些指标比盯着控制台输出更可靠。3. 核心模块拆解Producer 与 Consumer 的 NMS API 调用链与关键参数含义3.1 ProducerConnectionFactory → Connection → Session → Producer 的四层创建逻辑Producer.exe的主流程在Program.cs第 12–45 行核心是这四步对象创建// 1. 创建连接工厂指定协议、主机、端口、用户名密码本 Demo 未启用认证故为空 IConnectionFactory factory new ConnectionFactory(tcp://localhost:61616); // 2. 创建连接注意此处必须显式调用 Start()否则后续操作会超时 IConnection connection factory.CreateConnection(); connection.Start(); // ← 关键NMS 规范要求漏掉则 Send() 报错 // 3. 创建会话第一个参数 true 表示启用手动确认AcknowledgeMode.AutoAcknowledge 更常用 ISession session connection.CreateSession(AcknowledgementMode.AutoAcknowledge); // 4. 创建生产者destination 参数决定发往队列还是主题 IDestination destination session.GetQueue(test.queue); IMessageProducer producer session.CreateProducer(destination); producer.DeliveryMode MsgDeliveryMode.Persistent; // ← 持久化消息Broker 重启不丢关键参数说明ConnectionFactory构造函数中的 URL 支持多种参数如tcp://localhost:61616?wireFormat.maxInactivityDuration30000可设心跳超时DeliveryMode.Persistent是默认值若改为NonPersistent消息只存在内存Broker 崩溃即丢失CreateSession()的第二个参数AcknowledgeMode决定消费端如何确认消息。AutoAcknowledge最简单ClientAcknowledge需手动调message.Acknowledge()适合需要精确控制重试边界的场景。3.2 ConsumerMessageConsumer.Receive() 的阻塞机制与超时控制Consumer.exe的接收逻辑看似简单实则暗藏玄机// 创建消费者后必须调用 Start() 启动连接和 Producer 一样 connection.Start(); // 创建消费者destination 必须和 Producer 一致 IDestination destination session.GetQueue(test.queue); IMessageConsumer consumer session.CreateConsumer(destination); // Receive() 默认无限阻塞实际项目中必须加超时 IMessage message consumer.Receive(TimeSpan.FromSeconds(5)); // ← 5秒超时返回 null 表示无消息 if (message is ITextMessage textMsg) { Console.WriteLine($[Consumer] Received: {textMsg.Text}); } else { Console.WriteLine([Consumer] No message received within timeout.); }为什么不用ReceiveNoWait()因为它是非阻塞的轮询会空耗 CPU而Receive(TimeSpan)是底层 socket 的SO_RCVTIMEO设置由操作系统保证精度。若你希望 Consumer 持续运行比如 WinForm 后台线程应封装成循环while (isRunning) { IMessage msg consumer.Receive(TimeSpan.FromMilliseconds(100)); if (msg is ITextMessage t) { ProcessMessage(t.Text); // 你的业务逻辑 t.Acknowledge(); // 若用 ClientAcknowledge 模式 } Thread.Sleep(10); // 防止空转 }3.3 消息体封装从 string 到 IObjectMessage 的序列化边界控制Demo 默认用ITextMessage但 NMS 也支持IObjectMessage发送 .NET 对象// 发送对象需类标记 [Serializable] var order new Order { Id 1001, Status Shipped, Timestamp DateTime.Now }; IObjectMessage objMsg session.CreateObjectMessage(order); producer.Send(objMsg); // 接收端反序列化注意两端 .NET 版本、程序集签名必须一致 if (message is IObjectMessage obj) { var receivedOrder (Order)obj.Body; Console.WriteLine($Order {receivedOrder.Id} is {receivedOrder.Status}); }但强烈建议新手禁用IObjectMessage—— 因为它依赖 .NET BinaryFormatter已被微软标记为 [Obsolete].NET Core 3.0 完全移除一旦 Producer 和 Consumer 编译目标平台不同x64 vs x86反序列化直接抛SerializationException跨语言对接如 Java Consumer完全不可用。正确做法是用ITextMessage JSON。修改 Producervar json JsonSerializer.Serialize(new { id 1001, status Shipped, ts DateTime.UtcNow }); ITextMessage txtMsg session.CreateTextMessage(json); producer.Send(txtMsg);Consumer 端用System.Text.Json解析彻底摆脱序列化绑定也为未来升级 .NET 6 铺平道路。3.4 连接池与资源释放为什么每次都要显式 Close()NMS 的IConnection、ISession、IMessageProducer都实现IDisposable但它们的Dispose()并非只是释放托管内存// 正确写法用 using 确保 Close() 被调用 using (IConnection connection factory.CreateConnection()) { connection.Start(); using (ISession session connection.CreateSession()) { using (IMessageProducer producer session.CreateProducer(session.GetQueue(test.queue))) { producer.Send(session.CreateTextMessage(test)); } } } // ← 此处 connection.Close() 自动触发通知 Broker 清理会话资源若忘记Close()会发生Broker 端积累大量Inactive Connection占用 socket 和内存达到maxConnections1000上限后新连接被拒绝Windows 事件查看器中出现TCP port exhaustion事件表现为socket error 10048。这是很多开发者在压力测试时翻车的第一现场 —— 表面看是 Broker 性能差实则是客户端没关连接。我一般会在finally块里双重保险try { /* send logic */ } catch (Exception ex) { Log.Error(ex); } finally { producer?.Close(); session?.Close(); connection?.Close(); }4. 常见问题排查5 条血泪经验总结覆盖 90% 的本地运行失败场景4.1 现象Producer 报错 “Unable to connect to transport”Consumer 卡在 “Subscribed…” 无响应原因Broker 未启动或启动后被 Windows 防火墙拦截尤其公司域策略强制开启防火墙解决运行netstat -ano | findstr :61616确认 PID 是否对应java.exe若无输出重跑start-broker.bat并观察首行是否含is starting若有 PID 但 Consumer 仍连不上在防火墙高级设置中新建入站规则协议 TCP、端口 61616、作用域为“本地子网”避免开放公网。4.2 现象Consumer 收到消息后Producer 再发消息Consumer 不再打印原因Consumer 程序执行完Receive()后直接退出未保持连接长活或 Broker 端test.queue被自动删除因无活跃消费者且persistentfalse解决修改 Consumer 主循环为while(true)Thread.Sleep(100)确保连接持续检查activemq.xml中broker节点是否有schedulerSupporttrue若有则注释掉 —— 该选项在无调度任务时反而导致队列元数据清理异常发送端加一行producer.DeliveryMode MsgDeliveryMode.Persistent强制持久化。4.3 现象中文乱码Consumer 显示 “Hello World! Sent at 2024-04-05T10:20:30”原因NMS 默认用Encoding.DefaultWindows-1252编码而 .NET 控制台默认 UTF-8解决在 Producer 和 Consumer 的Main()开头添加Console.OutputEncoding Encoding.UTF8; Console.InputEncoding Encoding.UTF8;更彻底的方案在ConnectionFactoryURL 后追加?transport.useAsyncSendtruewireFormat.tightEncodingEnabledtrue启用 NMS 的紧凑编码模式对 UTF-8 友好。4.4 现象Producer 发送 100 条消息Consumer 只收到 98 条且无异常原因AutoAcknowledge模式下Broker 认为消息送达 TCP 连接即算成功但 Consumer 进程可能在Console.WriteLine前崩溃导致“已确认但未处理”解决改用ClientAcknowledge模式在业务逻辑处理完后再调message.Acknowledge()或启用事务session connection.CreateSession(true, AcknowledgementMode.AutoAcknowledge)然后session.Commit()提交整批。4.5 现象start-broker.bat运行几秒后自动退出日志只显示 “Java HotSpot...”原因JDK 版本过高如 JDK 17ActiveMQ 5.18 不兼容或activemq-all-5.18.0.jar文件损坏解决运行java -version确认输出为11.x.x删除activemq-all-5.18.0.jar从 Apache 官网归档页 重新下载校验 SHA256若仍失败在start-broker.bat中java命令前加-version参数调试java -version -jar activemq-all-5.18.0.jar。5. 二次开发实战把 Demo 改造成 WinForm 订单监控面板的 3 个关键改造点5.1 改造目标与架构定位从控制台到 GUI 的职责分离这个 Demo 的原始形态是两个独立控制台进程但在真实 WinForm 项目中我们不会让 UI 线程直接阻塞在consumer.Receive()上。正确姿势是UI 层Form只负责展示订单列表、状态标签、刷新按钮通信层BackgroundWorker 或 Task.Run专职连接 Broker、接收消息、解析 JSON业务层Service 类处理订单状态变更逻辑如触发声音提醒、更新数据库字段、调用 HTTP 接口。这样分层后即使消息接收线程卡住UI 也不会假死 —— 这是把 Demo 落地的第一道门槛。5.2 改造步骤一将 Consumer 封装为后台服务类新建ActiveMQService.cs继承IDisposable核心逻辑如下public class ActiveMQService : IDisposable { private readonly IConnectionFactory _factory; private IConnection _connection; private ISession _session; private IMessageConsumer _consumer; public event ActionOrderUpdate OnOrderUpdated; // 订阅者接收解析后的订单对象 public ActiveMQService(string brokerUrl tcp://localhost:61616) { _factory new ConnectionFactory(brokerUrl); } public void Start() { _connection _factory.CreateConnection(); _connection.Start(); _session _connection.CreateSession(AcknowledgementMode.AutoAcknowledge); _consumer _session.CreateConsumer(_session.GetQueue(order.queue)); // 启动接收循环用 Task 避免阻塞 UI Task.Run(() ReceiveLoop()); } private void ReceiveLoop() { while (_connection.IsStarted) { try { IMessage msg _consumer.Receive(TimeSpan.FromMilliseconds(500)); if (msg is ITextMessage textMsg !string.IsNullOrEmpty(textMsg.Text)) { var update JsonSerializer.DeserializeOrderUpdate(textMsg.Text); OnOrderUpdated?.Invoke(update); // 通知 UI 更新 } } catch (Exception ex) when (ex is IOException || ex is NMSException) { // 网络断开时静默重连不抛到 UI 线程 Thread.Sleep(2000); Reconnect(); } } } private void Reconnect() { _consumer?.Close(); _session?.Close(); _connection?.Close(); // 重新创建连接略同 Start() 初始化逻辑 } public void Dispose() { _consumer?.Close(); _session?.Close(); _connection?.Close(); } }注意OnOrderUpdated事件在非 UI 线程触发WinForm 中需用this.Invoke()切回主线程更新控件service.OnOrderUpdated update this.Invoke((MethodInvoker)delegate { lblStatus.Text $订单 {update.Id} 状态{update.Status}; });5.3 改造步骤二Producer 集成到业务按钮事件中在 WinForm 的「发货」按钮点击事件里不再新建Producer.exe进程而是复用 NMSprivate void btnShip_Click(object sender, EventArgs e) { try { using (var factory new ConnectionFactory(tcp://localhost:61616)) using (var conn factory.CreateConnection()) { conn.Start(); using (var session conn.CreateSession()) using (var producer session.CreateProducer(session.GetQueue(order.queue))) { var update new OrderUpdate { Id txtOrderId.Text, Status Shipped, Timestamp DateTime.UtcNow }; var json JsonSerializer.Serialize(update); producer.Send(session.CreateTextMessage(json)); MessageBox.Show(发货指令已发送); } } } catch (Exception ex) { MessageBox.Show($发送失败{ex.Message}); } }这里的关键是连接粒度控制每次点击都新建连接而非全局单例。因为 WinForm 用户操作频率低秒级连接复用收益小反而增加异常状态传播风险如某次发送失败导致整个连接不可用。5.4 改造步骤三配置外置化与错误降级策略硬编码tcp://localhost:61616在生产环境必然失败。应在app.config中配置configuration appSettings add keyActiveMQ_BrokerUrl valuetcp://192.168.1.100:61616/ add keyActiveMQ_QueueName valueorder.queue/ /appSettings /configuration读取方式string brokerUrl ConfigurationManager.AppSettings[ActiveMQ_BrokerUrl] ?? tcp://localhost:61616; string queueName ConfigurationManager.AppSettings[ActiveMQ_QueueName] ?? order.queue;更进一步加入降级开关当 ActiveMQ 不可用时自动切到本地 SQLite 队列暂存消息public async Taskbool SendMessageAsync(OrderUpdate update) { try { // 先尝试 ActiveMQ await SendToActiveMQ(update); return true; } catch (NMSException) { // 降级写入本地 DB await SaveToLocalDB(update); return false; // 返回 false 提示用户“已离线缓存” } }从那以后我每次给 WinForm 加消息功能都强制走一遍「控制台验证 → 后台服务封装 → UI 线程安全调用 → 配置外置化 → 降级兜底」这五步 checklist哪怕只是临时 Demo。少走一步上线后半夜三点的告警电话就多一分概率。希望帮到你。本文还有配套的精品资源点击获取
