简介面向.NET开发者的ActiveMQ C# Demo展示如何借助NMS.NET Messaging System在C#中连接ActiveMQ服务器实现点对点队列与发布/订阅主题两种消息模型的消息收发并演示连接工厂、会话、生产者/消费者等核心API的配置方式。压缩包共194个文件以45个cs源码、29个dll运行库、29个pdb调试符号、23个xml注释、9个exe可执行程序为主体另有sln/csproj工程文件及config、resx等配置资源包体大小12.71MB可直接打开工程开展调试。已有315人学习下载适合初探消息队列或需要在.NET环境中集成ActiveMQ的开发者。内容覆盖消息创建、发送、存储、接收与确认的完整生命周期并涉及持久化、高可用、负载均衡及多协议支持等特性配合图形化的Form配置界面可填写服务器地址、端口、账号等信息快速验证消息系统是否正常工作对理解异步通信和消息中间件原理有实际帮助。1. 接手ActiveMQ对接需求时最该先跑通的是一份C# Demo接手一个ActiveMQ对接需求时开发者最想先看到的不是架构图而是一份能跑的ActiveMQ C# Demo。它能让你在十分钟内确认三件事broker地址和端口通不通、队列和主题的消息流是什么形态、C#客户端该用哪些对象。这份Demo基于Apache.NMS.ActiveMQ实现覆盖点对点队列和发布订阅主题两种模式适合做上位机、后端服务集成的C#工程师。它解决的是“先跑通”的验证问题而不是把Spring整合方案搬过来。如果你正被“控制台能打开但代码连不上”“消息发了收不到”这类问题卡着后面几章会把选型、参数、代码和踩坑记录都摊开讲。2. ActiveMQ与C#的选型为什么是NMS消息模型怎么映射到API2.1 ActiveMQ在消息队列里的定位以及C#侧为什么绕不开NMSActiveMQ是Apache基金会下的开源消息代理兼容JMS规范广泛用于异构系统之间的异步解耦。Java端可以直接用JMS API但C#没有JMS标准官方提供的.NET客户端是Apache.NMS体系具体对应ActiveMQ的包叫Apache.NMS.ActiveMQ。这套客户端在API设计上刻意沿用了JMS的命名ConnectionFactory、Connection、Session、Destination、MessageProducer、MessageConsumer把JMS里的概念几乎原样搬进了.NET命名空间。那为什么不直接用AMQP或者STOMP协议ActiveMQ本身支持OpenWire、AMQP、STOMP等协议.NET生态里也有RabbitMQ.Client这种优秀的AMQP客户端。但从工程维护角度看如果你的上游就是一个ActiveMQ集群且团队里没有单独维护协议网关的人直接用官方NMS客户端是最省心的。它使用OpenWire协议与broker通信性能上接近Java原生客户端的表现。这里有一个常见的误解很多人以为ActiveMQ和RabbitMQ一样只有exchange和queue其实ActiveMQ用的是JMS术语队列Queue和主题Topic是两种不同的Destination。队列消息只被一个消费者取走属于点对点主题消息会被所有订阅者各收到一份属于发布订阅。这个区别直接决定了后续代码里用GetQueue还是GetTopic。还有一个值得注意的点Apache.NMS.ActiveMQ的更新节奏不快但它对ActiveMQ 5.x系列一直保持兼容。如果你用的是ActiveMQ 5.18.x直接引用最新NuGet包即可。真正容易带来问题的是旧项目里手动引用了某个DLL版本再配合新版broker时出现协议握手失败。所以我的习惯是所有C#端都从NuGet拉包不手动拷DLL。这也避免了后面第5章会讲的“程序集加载失败”问题。2.2 从Connection到MessageConsumer五个核心对象与连接参数对应关系在C#里写ActiveMQ代码核心对象其实就五个。我把它们和JMS的对应关系整理成一张表方便你后续查参数C#对象对应JMS作用关键参数ConnectionFactoryConnectionFactory创建连接保存broker地址、超时和认证信息BrokerUri、RequestTimeout、UserName、PasswordIConnectionConnection物理连接需要Start()后才会消费消息ClientIdISessionSession单线程上下文创建消息、生产者和消费者AckMode、事务开关IDestinationDestination队列或主题的抽象Queue/Topic名称IMessageProducer / IMessageConsumerMessageProducer / MessageConsumer发送和接收消息DeliveryMode、TimeToLive、Listener先创建ConnectionFactory传入ActiveMQ的OpenWire地址再创建Connection然后用Connection创建Session指定Destination后在Session上创建Producer或Consumer。一个Connection可以创建多个Session但一个Session不要跨线程使用这是很多并发问题最初的来源。ConnectionFactory的BrokerUri要写成tcp://localhost:61616默认OpenWire端口是61616Web控制台是8161两者不要混。如果broker启用了认证需要在工厂上设置UserName和Password否则连接阶段就会抛SecurityException。Connection上的ClientId参数平时可以省略但做持久订阅时必须显式设置。这个值会被broker用来识别订阅者身份一旦改变ActiveMQ就会认为这是一个新订阅者旧订阅的离线消息就找不回来了。Session的AckMode是最容易被忽视的参数AutoAcknowledge表示消费时自动确认ClientAcknowledge需要手动调用message.Acknowledge()Transactional模式则要显式提交事务。这些参数在写Demo时可能体会不到差异但一旦消息开始多起来每一个都会变成线上故障点。3. 搭建环境与Demo工程下载ActiveMQ、启动Broker、建立第一个连接3.1 安装ActiveMQ并确认8161、61616端口状态ActiveMQ官网提供Windows和Linux两种安装包。Windows下直接解压zipLinux下解压tar.gz。我一般在Linux机器上这样操作下载地址以官网为准下面只是示例命令cd /opt wget https://archive.apache.org/dist/activemq/5.18.4/apache-activemq-5.18.4-bin.tar.gz tar zxvf apache-activemq-5.18.4-bin.tar.gz cd apache-activemq-5.18.4/bin ./activemq start启动后不能只看控制台输出因为activemq start会让进程后台运行。要确认broker真正起来了去看../data/activemq.log看到类似Listening for connections at: tcp://和ActiveMQ JMS Message Broker started才算成功。启动完成后立刻检查两个端口netstat -tlnp | grep -E 61616|8161如果61616监听在0.0.0.0说明OpenWire对外网卡可见。如果只监听在127.0.0.1C#客户端只能本机连。这个问题在云服务器上特别常见需要修改conf/activemq.xml里的transportConnectortransportConnector nameopenwire uritcp://0.0.0.0:61616/修改后必须重启brokerActiveMQ不会热加载这个配置。在Windows上部署时还要额外检查防火墙是否放行了61616以及杀毒软件是否拦截了activemq进程。我在一个客户现场就遇到过控制台8161能打开、C#程序无论如何超时的情况最后定位就是云安全组只放行了8161没放行61616。所以端口验证这一步一定不要跳过。3.2 创建C#控制台项目并引入Apache.NMS.ActiveMQ包用.NET 6或更高版本创建控制台项目很简单dotnet new console -n ActivemqDemo cd ActivemqDemo dotnet add package Apache.NMS.ActiveMQNuGet上同时存在Apache.NMS和Apache.NMS.ActiveMQ两个包你只需要加后者它会把Apache.NMS作为依赖自动装进来。如果项目原来已经引用了Apache.NMS安装后先运行dotnet restore看到程序集冲突就手动把老的NMS包移除。这里不要抱有侥幸心理版本不一致会在运行时抛FileLoadException而且模糊指向某个未知程序集非常难查。安装完成后先写一个最简连接测试不急着发消息。这一步的目标是确认OpenWire链路通using Apache.NMS; using Apache.NMS.ActiveMQ; class Program { static void Main() { // 连接工厂地址必须是 tcp:// 前缀8161是HTTP控制台不是消息端口 ConnectionFactory factory new ConnectionFactory(tcp://localhost:61616); using IConnection connection factory.CreateConnection(); connection.Start(); // 创建连接后必须Start否则消费者不会接收消息 using ISession session connection.CreateSession(); Console.WriteLine(Connected to ActiveMQ successfully.); } }这段代码的流程是创建工厂、创建连接、启动连接、创建会话。CreateSession()没传参数时默认使用AutoAcknowledge用于连通性测试完全够用。如果代码在CreateConnection处报超时优先查61616端口和防火墙如果报Name or service not known检查BrokerUri是不是把localhost拼错了如果看到SecurityException需要在工厂上补认证参数ConnectionFactory factory new ConnectionFactory(tcp://localhost:61616); factory.UserName admin; factory.Password admin;ActiveMQ默认没有开启认证的话这两个参数可不填。但如果你连接的broker是运维同事带认证配置的就必须填否则永远卡在连接阶段。这一步通过后你的环境就具备了跑通完整Demo的条件。3.3 连接测试里最常见的三种状态超时、拒绝、认证失败连接测试跑不通时系统会给出不同错误不要只盯着“连接异常”四个字看。超时最常见说明请求发出了但没人应答原因基本是防火墙拦截或地址绑定了127.0.0.1。连接拒绝Connection refused说明端口没服务broker没启动或者61616端口被其他进程占用用lsof -i:61616可以快速查到占用情况。认证失败会明确提示SecurityException这时去broker的conf/activemq.xml看authenticationUser配置或者向broker管理员确认账号。把连接测试这一步跑扎实后面所有Demo代码都不会因为环境问题卡壳。4. 写一个能跑通的生产者与消费者队列模式和主题模式的完整代码4.1 队列模式点对点消息只被一个消费者处理队列模式是消息队列最基础的使用方式。生产者向demo.queue这个队列发送消息多个消费者存在时ActiveMQ会把消息分发给其中一个处理同一个消息不会被多个消费者同时消费。下面先写生产者using Apache.NMS; using Apache.NMS.ActiveMQ; ConnectionFactory factory new ConnectionFactory(tcp://localhost:61616); using IConnection connection factory.CreateConnection(); using ISession session connection.CreateSession(); // 声明目标通道队列名必须和消费者完全一致 IDestination destination session.GetQueue(demo.queue); using IMessageProducer producer session.CreateProducer(destination); producer.DeliveryMode MsgDeliveryMode.Persistent; for (int i 0; i 5; i) { ITextMessage message session.CreateTextMessage($Hello ActiveMQ #{i}); producer.Send(message); Console.WriteLine($Sent: {message.Text}); }MsgDeliveryMode.Persistent表示broker会持久化这条消息重启后不丢。如果业务场景允许丢失改成NonPersistent能得到更好的吞吐。producer.Send默认没有设置过期时间消息会一直留在队列里直到有消费者处理。如果希望消息只在有限时间内有效可以设置TimeToLiveproducer.TimeToLive TimeSpan.FromMinutes(10);设置过期时间后超时的消息会被broker标记为过期消费者正常情况下不会收到。这个参数特别适合做验证码、临时状态这类短生命周期消息。队列名称可以包含层级比如queue.sales.order但ActiveMQ不会自动创建目录结构它只是一个字符串标识。要注意的是生产者和消费者的队列名必须一字不差之前见过有人一个写demo.queue另一个写demo_queue结果两边都以为对方有问题实际是命名不一致。消费者用同步方式接收适合演示和测试using Apache.NMS; using Apache.NMS.ActiveMQ; ConnectionFactory factory new ConnectionFactory(tcp://localhost:61616); using IConnection connection factory.CreateConnection(); connection.Start(); using ISession session connection.CreateSession(AcknowledgementMode.AutoAcknowledge); IDestination destination session.GetQueue(demo.queue); using IMessageConsumer consumer session.CreateConsumer(destination); IMessage message consumer.Receive(TimeSpan.FromSeconds(5)); if (message is ITextMessage textMessage) { Console.WriteLine($Received: {textMessage.Text}); } else { Console.WriteLine(No message within 5 seconds.); }同步Receive(TimeSpan)适合测试但实际生产环境更多用异步监听避免主线程被阻塞。异步消费写法只需要把最后的接收循环替换成Listenerconsumer.Listener message { if (message is ITextMessage textMessage) { Console.WriteLine($Listener received: {textMessage.Text}); } }; Thread.Sleep(TimeSpan.FromSeconds(10));Listener回调在NMS内部线程执行所以回调里不能阻塞太久更不能不捕获异常否则回调线程的异常会传给连接层导致整个连接被关闭。这一点细节会放到第5章单独展开。4.2 主题模式发布订阅用ClientId和持久订阅保消息主题模式和队列模式的区别在于一条主题消息会发给所有在线订阅者而不是竞争消费。但ActiveMQ默认的Topic消费者有一个很坑的行为消费者离线后就收不到离线期间发布的消息。要让离线消费者恢复必须做持久订阅。在C#里需要两步连接上设置ClientId然后使用CreateDurableConsumer。下面是持久订阅消费者的骨架ConnectionFactory factory new ConnectionFactory(tcp://localhost:61616); using IConnection connection factory.CreateConnection(); // 持久订阅必须设置唯一ClientIdbroker通过它识别订阅者身份 connection.ClientId demo-durable-consumer-001; connection.Start(); using ISession session connection.CreateSession(AcknowledgementMode.AutoAcknowledge); IDestination topic session.GetTopic(demo.topic); // 订阅名也要固定重启后ActiveMQ依然记得这个订阅关系 using IMessageConsumer consumer session.CreateDurableConsumer(topic, demo-subscriber, null); consumer.Listener message { if (message is ITextMessage textMessage) { Console.WriteLine($Received: {textMessage.Text}); } }; Thread.Sleep(TimeSpan.FromSeconds(10));生产者发送到主题的代码和队列差别不大只是把GetQueue换成GetTopicIDestination topic session.GetTopic(demo.topic); using IMessageProducer producer session.CreateProducer(topic); producer.DeliveryMode MsgDeliveryMode.Persistent; ITextMessage msg session.CreateTextMessage(persistent topic message); msg.Properties[category] A; producer.Send(msg);CreateDurableConsumer的第三个参数是选择器可以用消息属性过滤比如上面设置了category消费者就可以用category A作为选择器只接收匹配的消息using IMessageConsumer consumer session.CreateDurableConsumer(topic, demo-subscriber, category A);选择器字段必须是持久化消息中的Properties属性不能过滤body文本。还有一个要求是ClientId和订阅名一旦确定就不要随意改动。ActiveMQ会以“ClientId加订阅名”作为持久订阅的标识任何一边变化broker都会注册成新订阅旧的离线消息会因订阅失效被清理从而产生“重启后消息丢了”的诡异现象。4.3 事务Session在C#里的正确姿势如果Session创建时使用了AcknowledgementMode.Transactional那所有消息的发送和接收都在事务里。发送方如果不调用Commit消息不会真正从broker上放出去队列消费者如果不调用Commit消息会停留在pending状态。代码表现非常隐蔽因为Producer和Consumer都不会主动报错using ISession session connection.CreateSession(AcknowledgementMode.Transactional); using IMessageProducer producer session.CreateProducer(destination); producer.Send(session.CreateTextMessage(transaction message)); session.Commit(); // 忘了这行消息永远不会被消费者看到接收端默认使用AutoAcknowledge的话事务模式下每条消息也只有提交后才算被消费。这种设计避免了一批消息处理一半崩溃导致数据不一致的问题。但代价是代码必须保证在合适的时机提交或回滚否则很容易出现积压或重复消费。我的建议是事务只在你确实有批量发消息需求时才开启单条消息就别给自己找麻烦。5. 避坑与排查ActiveMQ C# Demo跑不通的六个常见问题5.1 现象8161控制台能打开但C#程序一直连接超时原因运维通常只给防火墙放行了HTTP控制台端口8161而C#客户端使用的OpenWire端口61616根本没有对外开放。还有一种情况是ActiveMQ绑定了127.0.0.1只能本机访问。解决先检查netstat -tlnp | grep 61616确认监听地址是0.0.0.0再检查云安全组或本机防火墙是否放行61616。修改conf/activemq.xml里的transportConnector监听地址后重启broker。我每次部署完都强行让自己先做端口检查不再相信控制台能打开就等于客户端能连上。5.2 现象生产者发送成功消费者运行但还是收不到队列消息原因最常见的是漏了connection.Start()。NMS里连接对象默认不启动消费者虽然创建了但不会主动拉消息。第二个原因是Session开启了事务但没提交消息只在事务内可见。第三个原因可能是有多个ActiveMQ实例生产者和消费者分别连了不同的broker。解决在消费者代码里显式调用connection.Start()如果用了事务模式在每次接收完成后调用session.Commit()核对两个连接的BrokerUri是否完全一致包括主机名和端口。5.3 现象持久订阅消费者重启后离线期间的消息没收到原因Connection.ClientId或订阅名在重启后变了。ActiveMQ根据ClientId加订阅名判断一个持久订阅关系两边只要有一处变化broker就当成新订阅旧订阅积压的消息会被清理。解决把ClientId和订阅名写死在配置里不要用随机字符串。还要注意持久订阅消费者必须至少成功启动过一次broker才会记录订阅关系第一次启动之前发布的主题消息无论怎么订阅都不会补收。5.4 现象队列Pending数量一直居高不下消费者明明在网络中有连接原因Session使用了ClientAcknowledge或Transactional模式代码却没有调用Acknowledge()或Commit()。ActiveMQ把消息投递给消费者后只要没收到确认就会一直视为未消费同时不断重新投递。解决在接收循环里手动调用message.Acknowledge()或者事务模式在整批处理结束后调用Commit()。还有一点要注意Listener回调模式中如果回调立即返回消息可能还没来得及确认又被再次投递这取决于确认时机。建议在回调内先完成业务处理再返回反正AutoAcknowledge模式下回调结束后才会自动确认。5.5 现象Listener回调里处理消息时抛异常导致连接直接被关闭原因NMS的Listener回调运行在内部线程池上回调内未捕获的异常会穿透到连接层导致连接断开。解决在Listener内部用try-catch把业务处理包起来异常记录日志但不要抛出consumer.Listener message { try { ITextMessage textMessage (ITextMessage)message; Console.WriteLine($Handle: {textMessage.Text}); } catch (Exception ex) { Console.WriteLine($Handle failed: {ex.Message}); } };同理在Listener里新开线程做异步处理时要小心Session不是线程安全的多线程同时使用同一个Session会引发难以定位的乱序或消息丢失。这个现象让不少新人觉得ActiveMQ有“玄学”实际上就是线程模型没处理干净。5.6 现象NuGet安装Apache.NMS.ActiveMQ失败或运行时报程序集加载异常原因项目里手动引用了旧版本的Apache.NMS DLL再通过NuGet安装新版后出现版本冲突。解决先把项目里所有Apache.NMS相关的DLL引用移除再执行dotnet add package Apache.NMS.ActiveMQ让NuGet统一管理依赖版本。如果公司内网无法访问nuget.org用dotnet nuget list source检查源配置或者配置一个内网镜像源。不要从网上下载一个单独的DLL塞进项目里NMS的依赖关系比看起来复杂手工管理版本一定会翻车。6. 从Demo到落地用监控控制台验证消息流转并让代码更接近生产6.1 用ActiveMQ Web控制台确认消息真的送达到位Demo跑通后不要只看控制台输出就完事更可靠的验证方式是打开ActiveMQ Web控制台http://localhost:8161默认账号密码是admin/admin。进入Queues页面你会看到队列名、Enqueue数、Dequeue数和Pending数。跑一次生产者后Enqueue应该增加跑一次消费者后Dequeue也随之增加Pending归零。如果生产者和消费者同时运行Pending数一直不为零说明消费速度跟不上或者消息没有被正确确认。Topic页面里看到的数字是瞬时值因为主题消息不持久化。想验证持久订阅是否生效可以先停掉消费者发几条主题消息再重启消费者观察Dequeue是否恢复了离线期间的消息。这一步验证的价值比看任何日志都直接我现在每个Demo交付前都会在控制台上把这条路走一遍。6.2 三个让Demo更接近生产的小习惯超时设置、连接复用、日志记录第一个习惯是给连接设置合理超时。NMS的默认行为在某些异常网络环境下会一直挂住导致排查时分不清是消息丢了还是Broker卡了ConnectionFactory factory new ConnectionFactory(tcp://localhost:61616); factory.RequestTimeout TimeSpan.FromSeconds(10);第二个习惯是连接复用。不要每次发消息都new一个ConnectionFactory和Connection测试代码这么写没问题生产环境每次建立连接都要做TCP握手加OpenWire握手成本很高。正确的做法是把Connection和Session提升为单例由服务容器管理生命周期只创建一次。第三个习惯是在日志里记录消息关联字段。NMS的消息对象自带NMSCorrelationID和NMSTimestamp收到消息时打印这两个值加消息正文之后在控制台或日志系统里能按消息ID精准定位一条消息从生产到消费的完整链路。这个习惯能帮你省掉大量“消息到底去哪了”的追问时间。从那以后我每次写完ActiveMQ C# Demo都会强制自己走一遍“端口检查、控制台确认、超时设置、日志打印”四步再交给同事复现。消息队列这东西黑匣子越多越容易被坑把验证工具用起来比看十篇理论文章都管用。希望帮到你。本文还有配套的精品资源点击获取
