简介本资源是一套面向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生态里怎么真正落地你手头有个C#写的订单服务想加个异步通知模块——比如用户下单后不卡主流程而是发个消息给短信服务、库存服务、日志服务各自去处理。这时候ActiveMQ常被推荐但搜到的所谓“C# Demo”十有八九是Java客户端改个命名空间、或者用NMS简单连一下就戛然而止。真实项目里你会立刻撞上为什么消费者收不到消息为什么重启后消息丢了为什么用Queue能通换Topic就静默为什么.NET Framework能连.NET 6反而报System.NotSupportedException这个ActiveMQ DemoC#资源就是为解决这些非教学场景下的实操断点而生的它包含完整可运行的生产级结构——带连接池管理、异常重连策略、消息体序列化统一入口、死信队列DLQ自动路由配置以及最关键的一份针对Windows Server IIS .NET Core 3.1 / .NET 6双环境验证过的activemq.xml精简配置模板。适合正在把老系统迁移到微服务架构、或需要在.NET生态里嵌入可靠消息解耦能力的后端开发者。它不讲JMS规范理论只告诉你哪几行代码决定消息是否持久、哪个XML节点控制重试次数、为什么AcknowledgeMode.Transactional在C#里必须配transactedtrue才生效。2. 搭建与验证从ActiveMQ服务端启动到C#客户端首次收发消息2.1 服务端部署用最简路径绕过Java环境陷阱ActiveMQ官方包依赖JRE但很多.NET团队服务器上没装Java。别急着配JAVA_HOME——直接下载ActiveMQ 5.18.3 Windows ZIP版注意不是tar.gz也不是最新5.19.x那个对.NET NMS兼容性有已知问题。解压后进入conf/activemq.xml重点修改三处!-- 1. 开启STOMP协议支持C#常用比OpenWire更稳定 -- transportConnectors transportConnector namestomp uristomp://0.0.0.0:61613?transport.closeAsyncfalse/ /transportConnectors !-- 2. 关闭AMQP避免.NET客户端因协议协商失败静默退出 -- !-- 注释掉整个 amqp ... transportConnector 块 -- !-- 3. 配置持久化为KahaDB非默认LevelDB后者在Windows长路径下易出错 -- persistenceAdapter kahaDB directory${activemq.data}/kahadb journalMaxFileLength32mb enableIndexWriteAsynctrue/ /persistenceAdapter提示transport.closeAsyncfalse是关键。NMS客户端在连接关闭时若遇到异步关闭会触发ObjectDisposedException并中断重连逻辑此参数强制同步关闭让C#重试机制真正生效。启动服务管理员权限运行bin\activemq.bat start。检查data/activemq.log末尾是否有INFO | Apache ActiveMQ 5.18.3 (localhost, ID:xxx) started。再用浏览器访问http://localhost:8161/admin/默认账号admin/admin确认Queues列表为空——说明服务干净启动。2.2 C#客户端工程结构按生产环境分层组织新建.NET 6 Console App不要用.NET FrameworkNMS对新版本适配更好。NuGet安装Apache.NMSv1.7.2核心抽象层Apache.NMS.Stompv1.7.2STOMP协议实现Microsoft.Extensions.DependencyInjection用于解耦连接管理目录结构按职责拆分/src /Core ← NMS连接工厂、连接池、重连策略 /Models ← 消息体DTO带[Serializable]和JSON序列化契约 /Services ← 生产者Service、消费者Service含事件回调 /Config ← activemq.json连接字符串、重试次数、超时等核心连接工厂代码Core/NmsConnectionFactory.cspublic class NmsConnectionFactory : INmsConnectionFactory { private readonly ILoggerNmsConnectionFactory _logger; private readonly IOptionsActiveMqOptions _options; public NmsConnectionFactory(ILoggerNmsConnectionFactory logger, IOptionsActiveMqOptions options) { _logger logger; _options options; } public IConnection CreateConnection() { try { var factory new ConnectionFactory(_options.Value.BrokerUri); // stomp://localhost:61613 var connection factory.CreateConnection( _options.Value.Username, _options.Value.Password); // 关键启用事务性会话否则持久化消息不生效 connection.ClientId $client-{Guid.NewGuid():N}; connection.Start(); return connection; } catch (Exception ex) { _logger.LogError(ex, Failed to create ActiveMQ connection); throw; } } }参数说明BrokerUri必须用stomp://前缀不能写tcp://那是OpenWire协议.NET客户端支持不稳定ClientId非空且唯一这是Topic模式下消费者身份识别的硬性要求缺了就收不到广播消息connection.Start()必须显式调用NMS设计如此不调用则会话处于STOPPED状态发消息无响应2.3 发送第一条消息用ProducerService验证端到端链路在Services/ProducerService.cs中注入INmsConnectionFactory实现发送逻辑public class ProducerService : IHostedService { private readonly ILoggerProducerService _logger; private readonly INmsConnectionFactory _factory; private readonly string _queueName ORDER.CREATED; public ProducerService(ILoggerProducerService logger, INmsConnectionFactory factory) { _logger logger; _factory factory; } public async Task StartAsync(CancellationToken cancellationToken) { using var connection _factory.CreateConnection(); using var session connection.CreateSession(AcknowledgementMode.AutoAcknowledge); using var destination session.GetQueue(_queueName); using var producer session.CreateProducer(destination); // 设置消息持久化否则ActiveMQ重启后消息丢失 producer.DeliveryMode MsgDeliveryMode.Persistent; var message session.CreateTextMessage(OrderIDORD-2024-001;Amount299.99); message.Properties.SetString(MessageType, OrderCreatedEvent); message.Properties.SetLong(Timestamp, DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()); producer.Send(message); _logger.LogInformation(Sent message to queue: {Queue}, _queueName); } public Task StopAsync(CancellationToken cancellationToken) Task.CompletedTask; }逻辑说明MsgDeliveryMode.Persistent是持久化的开关必须显式设置。NMS默认是NonPersistent即使ActiveMQ配置了KahaDB消息也会进内存队列服务重启即丢。Properties写入自定义头信息这是跨服务传递元数据的标准方式如消息来源、业务类型比塞进消息体更轻量、更易过滤。CreateTextMessage用字符串而非对象避免.NET序列化器与Java端不兼容。生产环境建议统一用JSON字符串由消费者自行反序列化。验证方法启动程序后立即打开ActiveMQ Web Console → Queues → 点击ORDER.CREATED→ 查看Number Of Pending Messages是否为1。若为0说明消息未送达重点检查BrokerUri协议前缀和DeliveryMode设置。3. 消费端实现带重连、死信、手动签收的健壮消费者3.1 消费者Service用BackgroundService托管长连接Services/ConsumerService.cs继承BackgroundService确保应用生命周期内持续监听public class ConsumerService : BackgroundService { private readonly ILoggerConsumerService _logger; private readonly INmsConnectionFactory _factory; private readonly string _queueName ORDER.CREATED; private IConnection _connection; private ISession _session; private IMessageConsumer _consumer; public ConsumerService(ILoggerConsumerService logger, INmsConnectionFactory factory) { _logger logger; _factory factory; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { await ConnectAndConsume(stoppingToken); } private async Task ConnectAndConsume(CancellationToken cancellationToken) { while (!cancellationToken.IsCancellationRequested) { try { _connection _factory.CreateConnection(); _session _connection.CreateSession(AcknowledgementMode.ClientAcknowledge); var destination _session.GetQueue(_queueName); _consumer _session.CreateConsumer(destination); // 关键注册消息回调非轮询 _consumer.Listener async msg { try { await ProcessMessage(msg); await msg.Acknowledge(); // 手动签收 } catch (Exception ex) { _logger.LogError(ex, Error processing message); // 不Ack → 消息重回队列最多重试6次见下一节 } }; _logger.LogInformation(Started consuming from queue: {Queue}, _queueName); await Task.Delay(Timeout.Infinite, cancellationToken); } catch (Exception ex) when (ex is IOException or NMSException) { _logger.LogWarning(ex, Connection lost, retrying in 5s...); CleanupResources(); await Task.Delay(TimeSpan.FromSeconds(5), cancellationToken); } } } private async Task ProcessMessage(IMessage message) { if (message is ITextMessage textMsg) { _logger.LogInformation(Received: {Content}, textMsg.Text); // 这里放你的业务逻辑解析JSON、调用短信API、更新数据库... await Task.Delay(100); // 模拟耗时操作 } } private void CleanupResources() { _consumer?.Dispose(); _session?.Dispose(); _connection?.Dispose(); _consumer null; _session null; _connection null; } public override async Task StopAsync(CancellationToken cancellationToken) { CleanupResources(); await base.StopAsync(cancellationToken); } }参数说明AcknowledgementMode.ClientAcknowledge手动签收模式。这是生产环境必须选项——只有调用msg.Acknowledge()ActiveMQ才认为消息处理成功否则网络抖动、进程崩溃时消息会自动重回队列等待重试。BackgroundService的ExecuteAsync用while循环Task.Delay(Timeout.Infinite)保持线程存活比传统Thread.Sleep更符合.NET Core托管模型。CleanupResources在Stop和异常时释放资源NMS连接是重量级对象不释放会导致句柄泄漏最终Too many open files错误。3.2 死信队列DLQ配置让失败消息不卡主线程ActiveMQ默认将6次重试失败的消息移入ActiveMQ.DLQ队列。但你需要主动配置避免消息堆积在DLQ无人处理。编辑conf/activemq.xml在broker节点内添加destinationPolicy policyMap policyEntries !-- 对所有队列启用DLQ -- policyEntry queue producerFlowControltrue memoryLimit1mb deadLetterStrategy individualDeadLetterStrategy queuePrefixDLQ. useQueueForTopicMessagestrue/ /deadLetterStrategy /policyEntry /policyEntries /policyMap /destinationPolicy然后新增一个消费者专门监听DLQ// 在ConsumerService中追加 private void StartDlqConsumer() { var dlqDestination _session.GetQueue(DLQ.ORDER.CREATED); // 注意前缀 var dlqConsumer _session.CreateConsumer(dlqDestination); dlqConsumer.Listener msg { _logger.LogError(DLQ Message: {Text}, (msg as ITextMessage)?.Text); // 这里应发告警、存DB、人工介入 msg.Acknowledge(); }; }注意queuePrefixDLQ.生成的DLQ队列名是DLQ.ORDER.CREATED不是ActiveMQ.DLQ。这样便于按原队列分类处理失败消息。3.3 避坑C#消费者常见5类翻车现场与血泪修复方案现象 → 原因 → 解决现象消费者启动后控制台无日志Web Console显示Number Of Consumers0原因IConnection.Start()未被调用或IMessageConsumer.Listener赋值在Start()之前解决严格按顺序执行——connection.Start()→session.CreateConsumer()→consumer.Listener ...。用调试器确认consumer对象IsStarted属性为true现象消息发送成功但消费者收不到重启ActiveMQ后消息消失原因producer.DeliveryMode未设为Persistent或session创建时AcknowledgementMode用了AutoAcknowledge导致消息被自动删除解决发送端显式设producer.DeliveryMode MsgDeliveryMode.Persistent消费端必须用ClientAcknowledge并手动调用Acknowledge()现象.NET 6项目编译报错System.NotSupportedException: Specified method is not supported.原因NuGet安装了Apache.NMS.ActiveMQ仅支持.NET Framework而非Apache.NMS.Stomp解决卸载Apache.NMS.ActiveMQ只保留Apache.NMS和Apache.NMS.Stomp。STOMP协议是跨平台首选现象消费者处理消息时抛异常但消息未重回队列直接消失原因AcknowledgementMode设为AutoAcknowledge异常发生前消息已被自动签收解决必须用ClientAcknowledge并在try块内Acknowledge()catch块不调用Acknowledge()让ActiveMQ自动重发现象同一消息被重复消费多次幂等性破坏原因消费者处理逻辑耗时过长超过ActiveMQ的consumer.dispatchAsync超时默认30秒Broker认为消费者挂起将消息分发给其他实例解决在conf/activemq.xml中增加transportConnector ...?jms.prefetchPolicy.all1降低预取数量或在消费者中缩短单条消息处理时间拆分为快速响应异步后续处理4. 高级配置让ActiveMQ在Windows服务与IIS中稳定驻留4.1 将ActiveMQ注册为Windows服务告别cmd黑窗ActiveMQ自带WinService脚本但默认配置有坑。进入bin/win32/目录用管理员PowerShell执行# 1. 安装服务注意路径需用正斜杠且无空格 .\installService.ps1 -ServiceName ActiveMQ-Broker -ActiveMQHome C:/opt/activemq-5.18.3 # 2. 修改服务启动类型为自动非手动 Set-Service -Name ActiveMQ-Broker -StartupType Automatic # 3. 启动服务 Start-Service -Name ActiveMQ-Broker关键配置项conf/wrapper.conf# 必须指定JVM路径即使你装了JRE也要显式写 wrapper.java.commandC:\Program Files\Java\jre1.8.0_381\bin\java.exe # 增加堆内存避免Windows服务内存回收 wrapper.java.maxmemory1024 # 日志输出到独立文件方便排查 wrapper.logfile../data/wrapper.log wrapper.logfile.maxsize10m wrapper.logfile.maxfiles5提示installService.ps1脚本在PowerShell 7下可能报ExecutionPolicy错误。临时解决Set-ExecutionPolicy RemoteSigned -Scope CurrentUser。生产环境应签名脚本而非永久放开策略。4.2 IIS托管C# Web API时的ActiveMQ连接陷阱当C#消费者放在ASP.NET Core Web API中如用IHostedServiceIIS应用池回收会导致连接中断。解决方案分两层第一层应用池配置在IIS管理器 → 应用池 → 高级设置闲置超时分钟→ 设为0禁用定期回收工作进程分钟→ 设为0禁用启动模式→始终运行第二层代码级保活在ConsumerService中增加心跳检测private async Task StartHeartbeat(CancellationToken cancellationToken) { while (!cancellationToken.IsCancellationRequested) { try { // 发送空消息测试连接 using var testConn _factory.CreateConnection(); _logger.LogDebug(Heartbeat OK); } catch { _logger.LogWarning(Heartbeat failed, forcing reconnect...); // 触发重连逻辑 } await Task.Delay(TimeSpan.FromMinutes(2), cancellationToken); } }4.3 消息体序列化JSON统一契约避免.NET-Java类型错位不要用BinaryMessage或ObjectMessage——Java和.NET的二进制序列化器完全不兼容。强制约定所有消息体为UTF-8 JSON字符串并定义基类public abstract class BaseMessage { [JsonProperty(eventId)] public string EventId { get; set; } Guid.NewGuid().ToString(); [JsonProperty(timestamp)] public long Timestamp { get; set; } DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); [JsonProperty(source)] public string Source { get; set; } OrderService; } public class OrderCreatedMessage : BaseMessage { [JsonProperty(orderId)] public string OrderId { get; set; } [JsonProperty(amount)] public decimal Amount { get; set; } [JsonProperty(items)] public ListOrderItem Items { get; set; } }发送端var orderMsg new OrderCreatedMessage { OrderId ORD-2024-001, Amount 299.99 }; var json JsonSerializer.Serialize(orderMsg); producer.Send(session.CreateTextMessage(json));消费端if (message is ITextMessage textMsg) { var orderMsg JsonSerializer.DeserializeOrderCreatedMessage(textMsg.Text); // 安全反序列化无需担心类型爆炸 }注意JsonSerializer用System.Text.Json.NET Core内置不要用Newtonsoft.Json后者在高并发下有额外GC压力。5. 生产验证技巧用三类工具交叉确认消息链路健康度5.1 Web Console JMX实时监控队列水位与消费者状态ActiveMQ Web Consolehttp://localhost:8161/admin/是第一道防线但只能看静态快照。开启JMX可获取实时指标编辑conf/activemq.xml在broker节点添加managementContext managementContext createConnectortrue connectorPort1099/ /managementContext用JDK自带jconsole连接service:jmx:rmi:///jndi/rmi://localhost:1099/jmxrmi导航到org.apache.activemq→TypeBroker→Attributes重点关注-TotalEnqueueCount总入队数验证生产者是否工作-TotalDequeueCount总出队数验证消费者是否消费-MemoryPercentUsage内存使用率90%需扩容或调低memoryLimit-ConsumerCount当前活跃消费者数应≥1否则消息积压血泪经验某次线上事故ConsumerCount显示为0但代码明明在跑。最后发现是IIS应用池回收后BackgroundService未正确重启——因为IHostedService.StartAsync抛了未捕获异常导致整个宿主停止。从此我每次部署都先查JMX的ConsumerCount再查日志。5.2 命令行工具用activemq-admin做消息级诊断ActiveMQ自带bin/activemq-adminLinux或bin/activemq-admin.batWindows可脱离Web界面操作# 查看指定队列详情含未消费消息数、消费者数 activemq-admin query --query TypeQueue,DestinationORDER.CREATED --view QueueSize,ConsumerCount,EnqueueCount # 强制清除队列仅开发环境 activemq-admin purge --destination ORDER.CREATED # 查看DLQ中的消息内容Base64编码需解码 activemq-admin browse --destination DLQ.ORDER.CREATED --view JMSMessageID,Text提示browse命令输出的Text字段是Base64复制后用在线工具解码或用PowerShell[System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String(SGVsbG8gV29ybGQh))5.3 自研健康检查Endpoint让K8s/Liveness Probe真正有效在ASP.NET Core中暴露/health/activemq端点返回连接状态app.MapGet(/health/activemq, async (HttpContext ctx) { try { // 创建轻量连接测试 using var conn _factory.CreateConnection(); using var sess conn.CreateSession(); using var dest sess.GetQueue(HEALTH.CHECK); using var prod sess.CreateProducer(dest); prod.DeliveryMode MsgDeliveryMode.NonPersistent; // 测试用不持久化 prod.Send(sess.CreateTextMessage(PING)); ctx.Response.StatusCode StatusCodes.Status200OK; await ctx.Response.WriteAsync(ActiveMQ OK); } catch (Exception ex) { ctx.Response.StatusCode StatusCodes.Status503ServiceUnavailable; await ctx.Response.WriteAsync($ActiveMQ DOWN: {ex.Message}); } });然后在K8slivenessProbe中配置livenessProbe: httpGet: path: /health/activemq port: 80 initialDelaySeconds: 30 periodSeconds: 10从那以后我每次上线新服务都强制走一遍这三步验证先用activemq-admin query确认队列存在且ConsumerCount 0再用JMX看TotalEnqueueCount和TotalDequeueCount是否同步增长最后curl健康检查接口看返回200且无异常日志。漏掉任何一步我都不会点发布按钮——因为ActiveMQ的“静默失败”太常见它不会报错只会让你的消息在黑洞里漂流。希望帮到你。本文还有配套的精品资源点击获取