如何在C#中使用Apache.NMS.ActiveMQ客户端向多个队列发送消息？

ActiveMQ Classic没有原生的Fanout Exchange，但可以通过**Topic + 临时队列（Temporary Queue）**组合实现等价功能，完美适配你“消费者动态创建/销毁专属队列、消息广播到所有在线队列”的需求，同时兼容AWS ActiveMQ。

核心实现逻辑 #

利用Topic的广播特性，让每个消费者连接后创建一个临时队列并绑定到Topic，Topic的消息会自动路由到所有绑定的临时队列。临时队列会在消费者断开连接时被ActiveMQ自动删除，不用手动维护队列生命周期。

消费者端代码示例（Java） #

每个消费者连接后创建独立的临时队列，监听消息：

// 初始化连接工厂，替换为你的ActiveMQ地址
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://your-activemq-host:61616");
Connection connection = factory.createConnection();
connection.start();

// 创建非事务性会话，自动确认消息
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

// 定义目标Topic，所有消费者都绑定这个Topic
Topic fanoutTopic = session.createTopic("broadcast.fanout");

// 创建临时队列，断开连接自动销毁
TemporaryQueue tempQueue = session.createTemporaryQueue();

// 将临时队列绑定到Topic，生成唯一订阅ID避免冲突
String subId = "dynamic-sub-" + UUID.randomUUID().toString();
session.createDurableSubscriber(fanoutTopic, subId, "", false);

// 监听临时队列的消息
MessageConsumer consumer = session.createConsumer(tempQueue);
consumer.setMessageListener(message -> {
    try {
        if (message instanceof TextMessage) {
            System.out.println("收到广播消息: " + ((TextMessage) message).getText());
        }
    } catch (JMSException e) {
        e.printStackTrace();
    }
});

生产者端代码示例（Java） #

生产者只需向指定Topic发送消息，所有绑定的临时队列都会收到：

ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://your-activemq-host:61616");
Connection connection = factory.createConnection();
connection.start();

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Topic fanoutTopic = session.createTopic("broadcast.fanout");

MessageProducer producer = session.createProducer(fanoutTopic);
// 非持久化消息适合实时广播，若需离线接收可改为PERSISTENT
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

// 发送消息
TextMessage msg = session.createTextMessage("这是一条Fanout广播消息");
producer.send(msg);

// 关闭资源
producer.close();
session.close();
connection.close();

关键细节说明 #

• 临时队列特性：临时队列与消费者连接绑定，连接关闭后ActiveMQ自动删除队列，完全适配你“队列随消费者动态增减”的需求。
• AWS ActiveMQ适配：AWS托管的ActiveMQ Classic完全支持临时队列和Topic订阅，只需将连接地址替换为AWS提供的端点（注意使用正确的端口和认证信息），代码无需修改。
• 持久化选项：如果需要消费者离线后能接收消息，可改用Virtual Topic方案：生产者发往VirtualTopic.xxx，消费者创建队列Consumer.xxx.VirtualTopic.xxx，但这种方式需要手动管理队列生命周期（比如消费者断开时删除队列）。
• 消息过滤：如果需要部分消费者接收特定消息，可在绑定Topic时添加Selector规则（比如session.createDurableSubscriber(topic, subId, "type='alert'", false)），实现精准路由。

内容的提问来源于stack exchange，提问作者ZedZip