1. 项目概述
ActiveMQ作为一款成熟的开源消息中间件，在企业级分布式系统中扮演着重要角色。最近在金融支付系统升级项目中，我们采用ActiveMQ的Topic模式实现了跨系统的实时交易通知功能。本文将基于.NET 6环境，分享Topic模式的核心实现要点和避坑经验。
不同于常见的Queue点对点模式，Topic的发布/订阅模式更适合需要广播通知的场景。比如当支付订单状态变更时，需要同时通知风控系统、会计系统和客户通知系统。这种一对多的消息分发机制，如果配置不当很容易出现消息丢失或重复消费的问题。
2. 环境准备与基础配置
2.1 ActiveMQ服务端部署
推荐使用ActiveMQ 5.16.3以上版本，该版本对.NET Core的支持较为完善。在Linux环境下安装时需特别注意：
# 下载解压 wget https://archive.apache.org/dist/activemq/5.16.3/apache-activemq-5.16.3-bin.tar.gz tar zxvf apache-activemq-5.16.3-bin.tar.gz # 修改内存配置（重要！） vi conf/activemq.xml # 在<systemUsage>节点中调整内存限制 <systemUsage> <memoryUsage limit="512 mb"/> <storeUsage limit="1 gb"/> <tempUsage limit="100 mb"/> </systemUsage>注意：默认配置的内存限制太小，生产环境必须调整，否则在高并发时会出现消息阻塞。
2.2 .NET客户端依赖
通过NuGet安装最新版Apache.NMS和Apache.NMS.ActiveMQ：
dotnet add package Apache.NMS --version 2.0.0 dotnet add package Apache.NMS.ActiveMQ --version 2.0.0建议使用依赖注入方式初始化连接工厂：
services.AddSingleton<IConnectionFactory>(_ => new ConnectionFactory("tcp://activemq-host:61616?wireFormat.maxInactivityDuration=30000"));关键参数说明：
wireFormat.maxInactivityDuration：心跳检测间隔（毫秒）jms.prefetchPolicy.all=100：预取消息数量（影响消费速度）
3. Topic核心实现模式
3.1 生产者最佳实践
public class TopicPublisher : IDisposable { private readonly ISession _session; private readonly IMessageProducer _producer; public TopicPublisher(IConnectionFactory factory, string topicName) { var connection = factory.CreateConnection(); connection.Start(); _session = connection.CreateSession(); var topic = _session.GetTopic(topicName); _producer = _session.CreateProducer(topic); // 关键配置 _producer.DeliveryMode = MsgDeliveryMode.Persistent; _producer.RequestTimeout = TimeSpan.FromSeconds(30); } public void Publish(string message) { var textMessage = _session.CreateTextMessage(message); // 添加消息头便于追踪 textMessage.Properties["MsgId"] = Guid.NewGuid().ToString(); textMessage.Properties["Timestamp"] = DateTime.UtcNow.ToString("o"); _producer.Send(textMessage); } }重要细节：
- 必须设置
DeliveryMode.Persistent保证消息持久化 - 每个消息建议附加唯一标识和时间戳
- 连接启动后需要显式调用
Start()
3.2 消费者可靠实现方案
public class TopicSubscriber : IDisposable { private readonly ISession _session; private readonly IMessageConsumer _consumer; public TopicSubscriber( IConnectionFactory factory, string topicName, string clientId) { var connection = factory.CreateConnection(); connection.ClientId = clientId; // 必须设置唯一客户端ID connection.Start(); _session = connection.CreateSession(); var topic = _session.GetTopic(topicName); _consumer = _session.CreateDurableConsumer(topic, clientId, null, false); _consumer.Listener += OnMessageReceived; } private void OnMessageReceived(IMessage message) { if (message is ITextMessage textMessage) { try { // 业务处理逻辑 Console.WriteLine($"Received: {textMessage.Text}"); // 手动确认（重要！） textMessage.Acknowledge(); } catch (Exception ex) { // 记录失败消息便于重试 _session.Recover(); } } } }关键点：
- 必须设置
ClientId才能创建持久化订阅 - 使用
CreateDurableConsumer防止消费者离线时丢失消息 - 异常时调用
Recover()将未确认消息重新投递
4. 高级配置与优化
4.1 消息选择器（Selector）
通过设置选择条件实现消息过滤：
// 生产者设置属性 textMessage.Properties["Priority"] = "High"; // 消费者按条件订阅 _consumer = _session.CreateConsumer( topic, "Priority = 'High' AND Timestamp > '2023-01-01'");4.2 事务性会话配置
// 创建事务会话 var session = connection.CreateSession(AcknowledgementMode.Transactional); try { // 发送多条消息 producer.Send(message1); producer.Send(message2); // 提交事务 session.Commit(); } catch { // 回滚事务 session.Rollback(); }4.3 性能调优参数
| 参数名 | 推荐值 | 说明 |
|---|---|---|
| wireFormat.maxInactivityDuration | 30000 | 网络超时时间(ms) |
| jms.prefetchPolicy.all | 50-100 | 预取消息数量 |
| jms.redeliveryPolicy.maximumRedeliveries | 3 | 最大重试次数 |
| jms.sendTimeout | 5000 | 发送超时时间(ms) |
5. 常见问题排查
5.1 消息堆积问题
现象：消费者处理速度跟不上生产速度
解决方案：
- 增加消费者实例数量
- 调整
prefetchPolicy降低预取数量 - 优化消费者业务逻辑性能
5.2 重复消费问题
现象：同一条消息被处理多次
排查步骤：
- 检查是否配置了持久化订阅
- 确认消息确认模式（自动/手动）
- 检查网络是否不稳定导致重连
5.3 连接不稳定问题
现象：频繁断开连接
处理方案：
// 在连接工厂添加重试策略 var factory = new ConnectionFactory("failover:(tcp://primary:61616,tcp://backup:61616)" + "?initialReconnectDelay=1000" + "&maxReconnectDelay=30000" + "&maxReconnectAttempts=-1");6. 监控与维护建议
6.1 健康检查实现
public async Task<bool> CheckHealthAsync() { try { using var connection = _factory.CreateConnection(); using var session = connection.CreateSession(); var tempTopic = session.GetTopic(Guid.NewGuid().ToString()); using var producer = session.CreateProducer(tempTopic); var msg = session.CreateTextMessage("healthcheck"); await producer.SendAsync(msg); return true; } catch { return false; } }6.2 监控指标采集
建议监控以下关键指标：
- 待处理消息数量（Pending Message Count）
- 消费者数量（Consumer Count）
- 消息入队/出队速率（Enqueue/Dequeue Rate）
- 内存使用情况（Memory Usage）
可以通过JMX或ActiveMQ Web Console获取这些数据。
6.3 日志配置建议
在log4net配置中添加：
<logger name="Apache.NMS"> <level value="WARN" /> </logger> <logger name="Apache.NMS.ActiveMQ"> <level value="DEBUG" /> </logger>在开发环境可以将日志级别��为DEBUG，生产环境建议设为WARN以上。