简介本资源是一套面向C#开发者与消息中间件初学者的ActiveMQ入门实践Demo聚焦WinForm桌面应用中消息队列的集成开发帮助读者快速掌握ActiveMQ在.NET环境下的基础发送与接收逻辑。压缩包共36个文件含19个核心C#源码如MQ.cs、MQDemoProducer.csproj、Form1.cs等、4个资源文件.resx、4个依赖DLL、3个可执行EXE及1个完整Visual Studio解决方案.sln整体仅326KB轻量易导入结构清晰便于逐模块理解生产者/消费者模型与UI交互设计。已有1292人学习下载适合刚接触消息中间件、需从零构建可运行示例的中级C#开发者。读者可直接编译运行两个独立程序——图形化消息发送端与接收端结合代码注释与模块化分层UI层、业务逻辑层、MQ通信层深入理解连接管理、队列操作、异常处理等关键实践细节。1. ActiveMQ DemoC#不是“跑个Hello World”就完事而是摸清消息中间件在Windows服务场景下的真实心跳节律你手头有个.NET Framework 4.7.2的旧系统要对接一个外部物流平台——对方只提供ActiveMQ的Topic推送每秒30条运单状态变更。你搜到一堆“C#连接ActiveMQ”的Demo照着改了IP和端口本地控制台能收发一丢进Windows服务里就隔三差五断连、消息堆积、重启后丢数据。这不是Demo写得不对是绝大多数C#版ActiveMQ示例根本没碰过生产级长连接管理、异常重连策略、会话生命周期绑定这些黑匣子。这个ActiveMQ DemoC#资源包就是为这种场景打磨出来的它不教你怎么装Apache ActiveMQ而是直接给你一套可嵌入Windows服务、支持自动重连断线补偿日志追踪的轻量级封装核心代码不到800行但覆盖了从ConnectionFactory初始化、Session事务控制、MessageConsumer异步回调到IDisposable资源释放的全链路。适合正在维护遗留.NET系统、需要快速接入JMS生态又不想引入NServiceBus或MassTransit重型框架的开发者。它解决的不是“能不能通”而是“通了之后稳不稳定、断了之后找不找得到、压测时扛不扛得住”。2. 搭建环境与基础通信用Apache.NMS.ActiveMQ驱动跑通第一条消息这个Demo依赖的是Apache官方维护的.NET客户端库Apache.NMS.ActiveMQ不是第三方封装或过时的Stomp实现。它直接对接ActiveMQ的OpenWire协议性能和兼容性更可控。我们不走NuGet一键安装的老路而是手动验证依赖版本边界——因为很多翻车都出在DLL地狱上。2.1 环境准备确认ActiveMQ服务端版本与客户端匹配ActiveMQ 5.16.x及以后版本默认禁用OpenWire协议的匿名访问而Apache.NMS.ActiveMQ 1.7.2当前稳定版对新安全策略支持不完整。必须降级到ActiveMQ 5.15.13这是最后一个默认开启OpenWire且无需额外配置JAAS的版本。下载地址在Apache存档库archive.apache.org/dist/activemq/5.15.13/解压后启动# Windows下执行 bin\activemq.bat start提示启动后访问http://localhost:8161/admin/默认账号密码均为admin。进入Queues页确认TestQueue不存在——我们将用代码创建它避免预置队列带来的权限干扰。2.2 引入Apache.NMS.ActiveMQ并验证基础连接新建一个.NET Framework 4.7.2控制台项目不要通过Visual Studio NuGet GUI安装。打开Package Manager Console执行Install-Package Apache.NMS.ActiveMQ -Version 1.7.2 -Source https://api.nuget.org/v3/index.json为什么指定1.7.2因为1.8.0开始强制要求.NET Standard 2.0而你的Windows服务大概率跑在Framework下。安装后在App.config中添加连接字符串配置configuration appSettings add keyActiveMQ.BrokerUri valuetcp://localhost:61616 / add keyActiveMQ.Username valueadmin / add keyActiveMQ.Password valueadmin / /appSettings /configuration注意tcp://前缀不能省略61616是OpenWire默认端口不是Web控制台的8161。现在写最简发送端using System; using Apache.NMS; using Apache.NMS.ActiveMQ; class Program { static void Main() { var uri System.Configuration.ConfigurationManager.AppSettings[ActiveMQ.BrokerUri]; var factory new ConnectionFactory(uri); using (IConnection connection factory.CreateConnection( System.Configuration.ConfigurationManager.AppSettings[ActiveMQ.Username], System.Configuration.ConfigurationManager.AppSettings[ActiveMQ.Password])) { connection.Start(); // 必须显式调用Start() using (ISession session connection.CreateSession(AcknowledgementMode.AutoAcknowledge)) { IDestination destination session.GetQueue(TestQueue); using (IMessageProducer producer session.CreateProducer(destination)) { ITextMessage message session.CreateTextMessage(Hello from C# Demo!); producer.Send(message); Console.WriteLine(Sent: message.Text); } } } } }逻辑说明ConnectionFactory构造时传入tcp://URI底层会解析成TcpTransportFactoryconnection.Start()是关键一步不调用则后续所有操作抛IllegalStateExceptionAutoAcknowledge模式下Broker收到消息即认为消费成功适合测试生产环境应改用ClientAcknowledge并手动调用message.Acknowledge()session.CreateProducer(destination)中的destination必须是IQueue或ITopic实例不能传字符串——这是新手高频报错点。运行后去ActiveMQ Web控制台的Queues页刷新能看到TestQueue已自动创建且Enqueue Count为1。再写接收端验证using System; using Apache.NMS; using Apache.NMS.ActiveMQ; class Receiver { static void Main() { var uri System.Configuration.ConfigurationManager.AppSettings[ActiveMQ.BrokerUri]; var factory new ConnectionFactory(uri); using (IConnection connection factory.CreateConnection(admin, admin)) { connection.Start(); using (ISession session connection.CreateSession(AcknowledgementMode.AutoAcknowledge)) { IDestination destination session.GetQueue(TestQueue); using (IMessageConsumer consumer session.CreateConsumer(destination)) { ITextMessage message consumer.Receive(TimeSpan.FromSeconds(5)) as ITextMessage; if (message ! null) Console.WriteLine(Received: message.Text); else Console.WriteLine(No message received within timeout.); } } } } }参数说明consumer.Receive(TimeSpan.FromSeconds(5))设置5秒超时避免无限阻塞返回null表示超时未收到不是异常需业务层判断此处用AutoAcknowledge消息被Receive()取回即自动确认不会重回队列。3. 封装生产级消费者解决Windows服务中连接中断、消息丢失、资源泄漏三大顽疾控制台程序能跑通不等于Windows服务能稳住。把上面的接收逻辑直接塞进ServiceBase.OnStart()不出三天必出问题连接断开不重连、异常未捕获导致服务僵死、IConnection未释放引发句柄耗尽。这个Demo的核心价值在于它把IConnection生命周期、重连策略、异常隔离全部封装进一个ActiveMQConsumerService类。3.1 设计原则连接与会话分离重连与消费解耦传统写法常把IConnection、ISession、IMessageConsumer全塞在一个类里导致重连时必须销毁整个对象树。本Demo采用分层设计ConnectionManager专职管理IConnection内置指数退避重连初始1秒最大30秒每次×1.5SessionPool维护ISession对象池每个IMessageConsumer绑定独立ISession避免多消费者共享会话导致的ACK冲突MessageHandler纯业务逻辑接口只处理ITextMessage不接触NMS底层对象。这样当网络抖动导致IConnection断开时ConnectionManager后台静默重连SessionPool检测到连接失效后自动重建会话MessageHandler完全无感——这才是服务该有的韧性。3.2 关键代码ConnectionManager的重连引擎与状态机using System; using System.Threading; using Apache.NMS; using Apache.NMS.ActiveMQ; public class ConnectionManager : IDisposable { private readonly string _brokerUri; private readonly string _username; private readonly string _password; private IConnection _connection; private readonly object _lock new object(); private volatile bool _isRunning true; private volatile ConnectionState _state ConnectionState.Disconnected; public ConnectionManager(string brokerUri, string username, string password) { _brokerUri brokerUri; _username username; _password password; } public IConnection GetConnection() { if (_state ConnectionState.Connected _connection ! null) return _connection; lock (_lock) { if (_state ConnectionState.Connected _connection ! null) return _connection; // 断开旧连接 DisposeConnection(); // 启动重连循环 ThreadPool.QueueUserWorkItem(_ ReconnectLoop()); return null; // 调用方需轮询GetConnection()直到非null } } private void ReconnectLoop() { int retryCount 0; TimeSpan nextDelay TimeSpan.FromSeconds(1); while (_isRunning _state ! ConnectionState.Connected) { try { var factory new ConnectionFactory(_brokerUri); _connection factory.CreateConnection(_username, _password); _connection.ExceptionListener OnException; _connection.Start(); Interlocked.Exchange(ref _state, ConnectionState.Connected); Console.WriteLine($[ConnectionManager] Connected successfully after {retryCount} retries.); return; } catch (Exception ex) { retryCount; Console.WriteLine($[ConnectionManager] Connect attempt #{retryCount} failed: {ex.Message}); if (retryCount 10) // 防止无限重试 break; Thread.Sleep(nextDelay); nextDelay TimeSpan.FromMilliseconds(Math.Min(nextDelay.TotalMilliseconds * 1.5, 30000)); } } } private void OnException(Exception exception) { Console.WriteLine($[ConnectionManager] Broker exception: {exception.Message}); Interlocked.Exchange(ref _state, ConnectionState.Disconnected); } private void DisposeConnection() { _connection?.Close(); _connection?.Dispose(); _connection null; } public void Dispose() { _isRunning false; DisposeConnection(); } } public enum ConnectionState { Disconnected, Connecting, Connected }逻辑说明GetConnection()返回null是故意设计迫使调用方用while (conn null) { Thread.Sleep(100); conn mgr.GetConnection(); }轮询避免阻塞主线程ReconnectLoop()中nextDelay使用Math.Min(..., 30000)硬限30秒防止网络长期不可达时重试间隔爆炸OnException监听Broker主动断连如服务重启触发状态重置让重连循环再次启动Dispose()中_isRunning false是关键确保服务停止时重连线程能退出。3.3 SessionPool为每个消费者分配独立会话规避ACK污染using System; using System.Collections.Concurrent; using System.Threading; using Apache.NMS; public class SessionPool { private readonly ConnectionManager _connectionManager; private readonly ConcurrentDictionarystring, WeakReferenceISession _pool new ConcurrentDictionarystring, WeakReferenceISession(); private readonly object _lock new object(); public SessionPool(ConnectionManager connectionManager) { _connectionManager connectionManager; } public ISession GetSession(string sessionId) { if (_pool.TryGetValue(sessionId, out var weakRef) weakRef.TryGetTarget(out var session)) { if (session.IsStarted) return session; } // 创建新会话 var connection WaitForConnection(); var newSession connection.CreateSession(AcknowledgementMode.ClientAcknowledge); _pool.AddOrUpdate(sessionId, _ new WeakReferenceISession(newSession), (_, _) new WeakReferenceISession(newSession)); return newSession; } private IConnection WaitForConnection() { IConnection conn null; int waitCount 0; while (conn null waitCount 60) // 最多等待60秒 { conn _connectionManager.GetConnection(); if (conn null) Thread.Sleep(1000); waitCount; } if (conn null) throw new TimeoutException(Failed to get connection from ConnectionManager); return conn; } public void ReleaseSession(string sessionId) { _pool.TryRemove(sessionId, out _); } }参数说明WeakReferenceISession避免会话对象被长期强引用配合GC及时回收WaitForConnection()中60秒超时是底线超过则抛异常防止服务卡死ClientAcknowledge模式下必须由业务代码显式调用message.Acknowledge()否则消息会重复投递——这是保证Exactly-Once语义的基础。4. 避坑指南Windows服务部署中踩过的五个真实血泪坑把Demo代码扔进Windows服务看似一步之遥实则暗礁密布。以下是我在某高校实验室部署物流状态同步服务时连续三天排查出的五个高频问题每个都附带现象、根因和可落地的修复方案。4.1 现象服务启动后CPU飙升至100%Event Log里大量“The service did not respond to the start or control request in a timely fashion”原因OnStart()方法里直接调用ConnectionManager.GetConnection()并死等返回而首次连接因防火墙拦截超时30秒Windows服务管理器在15秒内判定启动失败反复尝试重启形成雪崩。解决OnStart()中只启动后台线程不阻塞主线程。修改ServiceBase.OnStart()如下protected override void OnStart(string[] args) { // 启动连接管理器但不等待连接建立 _connectionManager new ConnectionManager(brokerUri, username, password); _connectionManager.Start(); // 新增Start()方法内部启动ReconnectLoop线程 // 启动消息消费器同样不阻塞 _consumerService new ActiveMQConsumerService(_connectionManager, _sessionPool); _consumerService.StartConsuming(TestQueue, HandleMessage); }注意ConnectionManager需新增Start()方法将ReconnectLoop()改为Task.Run(() ReconnectLoop())避免ThreadPool.QueueUserWorkItem在服务上下文中的不确定性。4.2 现象服务运行数小时后netstat -ano | findstr :61616显示ESTABLISHED连接数持续增长最终达到系统上限原因IMessageConsumer创建后未调用Dispose()其底层TcpTransport连接未关闭。Apache.NMS.ActiveMQ的IMessageConsumer实现了IDisposable但文档极少强调这点。解决在ActiveMQConsumerService中为每个消费者注册AppDomain.CurrentDomain.ProcessExit和ServiceBase.OnStop事件确保进程退出时清理private void RegisterCleanupHandlers() { AppDomain.CurrentDomain.ProcessExit (s, e) Cleanup(); // 若为Windows服务还需在OnStop中调用Cleanup() } private void Cleanup() { _consumers.ForEach(c c?.Dispose()); // _consumers是ListIMessageConsumer _sessionPool?.Dispose(); // SessionPool也需实现IDisposable _connectionManager?.Dispose(); }4.3 现象ActiveMQ Web控制台显示TestQueue的Dequeue Count为0但业务数据库里没有新记录消息石沉大海原因HandleMessage(ITextMessage message)业务方法中抛出未捕获异常IMessageConsumer的异步回调线程崩溃后续消息不再投递且无任何日志。解决在MessageHandler调用链最外层加全局异常捕获并强制ACK失败消息以防堆积private void SafeHandleMessage(ITextMessage message) { try { _messageHandler.Handle(message); message.Acknowledge(); // 显式ACK } catch (Exception ex) { // 记录错误日志 _logger.Error($Failed to process message {message.NMSMessageId}: {ex}); // 强制ACK避免消息卡在ActiveMQ中生产环境应转存死信队列 try { message.Acknowledge(); } catch { /* ACK失败也放弃避免二次异常 */ } } }4.4 现象服务重启后部分消息被重复处理两次订单状态从“已发货”又被刷成“已发货”原因AcknowledgementMode.AutoAcknowledge下消息被Receive()取回即ACK但业务逻辑执行到一半服务崩溃消息已丢失ACK状态重启后Broker重发。解决切换为ClientAcknowledge并在业务逻辑成功提交数据库事务后再调用message.Acknowledge()public void Handle(ITextMessage message) { using (var scope new TransactionScope()) { // 1. 解析消息体 var order JsonConvert.DeserializeObjectOrder(message.Text); // 2. 更新数据库 _orderRepository.UpdateStatus(order.Id, order.Status); // 3. 仅在此处ACK确保DB写入成功 message.Acknowledge(); scope.Complete(); } }4.5 现象服务在低负载时正常压测每秒50条消息时Receive()开始超时Enqueue Count远大于Dequeue Count原因IMessageConsumer.Receive(timeout)是同步阻塞调用单线程消费无法应对高吞吐消息在Broker端堆积。解决改用AsyncCallback异步消费配合线程池并发处理public void StartConsuming(string queueName, MessageHandler handler) { var destination _session.CreateQueue(queueName); var consumer _session.CreateConsumer(destination); // 注册异步回调 consumer.Listener (msg) { ThreadPool.QueueUserWorkItem(_ SafeHandleMessage((ITextMessage)msg)); }; }提示ThreadPool默认最大线程数为CPU核数×2550条/秒通常够用若需更高并发用Task.Run()替代ThreadPool.QueueUserWorkItem并配置TaskScheduler。5. 进阶技巧用自定义MessageSelector实现动态路由绕过Topic/Queue的硬编码限制ActiveMQ原生支持JMS Message Selector允许消费者按消息属性而非内容过滤。比如物流平台推送的消息带MessageTypeShipmentUpdate或MessageTypeDeliveryConfirmation你不想为每种类型建独立Queue而是用一个AllEventsQueue靠Selector分流。这个Demo预留了MessageSelector参数入口但多数人不知道怎么安全注入。5.1 构造安全的Selector字符串防SQL注入式拼接Message Selector语法类似SQL WHERE子句但不支持参数化查询。直接拼接用户输入会导致Selector语法错误甚至Broker拒绝服务。正确做法是白名单校验public static string BuildSelector(string messageType) { var allowedTypes new HashSetstring { ShipmentUpdate, DeliveryConfirmation, OrderCancelled }; if (!allowedTypes.Contains(messageType)) throw new ArgumentException($Invalid message type: {messageType}); return $MessageType {messageType}; } // 使用示例 var selector BuildSelector(ShipmentUpdate); using (IMessageConsumer consumer session.CreateConsumer(destination, selector)) { // 只接收MessageType为ShipmentUpdate的消息 }5.2 在Windows服务中热更新Selector不用重启服务切换监听类型硬编码Selector意味着改配置就得重启服务。我们利用IMessageConsumer的Close()和重建能力实现运行时切换public class DynamicConsumer : IDisposable { private IMessageConsumer _currentConsumer; private readonly ISession _session; private readonly IDestination _destination; private readonly MessageHandler _handler; public DynamicConsumer(ISession session, IDestination destination, MessageHandler handler) { _session session; _destination destination; _handler handler; } public void SwitchToSelector(string newSelector) { _currentConsumer?.Close(); // 关闭旧消费者 // 重建带新Selector的消费者 _currentConsumer _session.CreateConsumer(_destination, newSelector); _currentConsumer.Listener msg _handler.Handle((ITextMessage)msg); } public void Dispose() { _currentConsumer?.Close(); } } // 在服务中调用 var dynamicConsumer new DynamicConsumer(session, destination, handler); dynamicConsumer.SwitchToSelector(MessageType ShipmentUpdate); // 运行中可随时切换 dynamicConsumer.SwitchToSelector(MessageType DeliveryConfirmation);5.3 消息属性与内容分离用NMSXGroupID实现同一业务流的顺序处理物流场景中“运单创建→揽件→发货→签收”必须严格顺序。ActiveMQ的NMSXGroupID可将消息分组确保同组消息被同一消费者串行处理。在发送端设置ITextMessage message session.CreateTextMessage(jsonPayload); message.NMSXGroupID $ORDER_{order.Id}; // 所有该订单的消息用相同GroupID message.NMSXGroupSeq seq; // 组内序列号Broker据此排序 producer.Send(message);在消费端无需额外配置只要IMessageConsumer不跨线程共享Broker自动保证NMSXGroupID相同的消息按NMSXGroupSeq顺序投递。这是比应用层排序更可靠的方案。从那以后我每次给Windows服务集成消息中间件都强制走一遍这五个坑的自查清单连接是否异步启动、消费者是否显式Dispose、ACK是否在DB事务后、异常是否兜底捕获、Selector是否白名单校验。少踩一个上线后就能少熬一个通宵。希望帮到你。本文还有配套的精品资源点击获取