从java客户端中的多个RabbitMQ交换中读取无轮询

Question

请解释如何配置Java客户端，使其在不进行轮询的情况下从两个不同的RabbitMQ交换中读取。我希望客户端在消息到达时唤醒，然后再次阻塞。

在我的小系统集成问题中，一个RabbitMQ交换携带使用多种路由密钥的工作消息（我知道如何使用通配符捕捉它们），另一个交换携带控制消息（例如，“stop”）。所以我的当事人要收听两地的留言。这是一个相对低容量的系统问题，我不是问负载分担或公平性等等。

当然，我可以运行一个线程来轮询每个交换、Hibernate、调度，直到永远。但我想避免投票。

不知怎的，我想起了Unix select（）系统调用，当任何文件描述符上的数据准备就绪时，它就会被唤醒。RabbitMQ也有类似的东西吗？

我当前的解决方案是一个适配器，它在每个输入交换上旋转一个线程来阻塞；接收到后，每个线程将写入java.util.concurrent集合；我还使用另一个线程来阻止该集合，并在消息到达最终使用者时将其传递出去。它工作很好，但如果我能删掉这种复杂性，那就太好了。

这些帖子围绕着这个问题跳来跳去，如果我在这些帖子中忽略了这个问题，请随时给我打听一下解决方案：

For Java：RabbitMQ示例：多线程、通道和队列

对于C#:从多个队列读取，RabbitMQ

提前道谢。

Answer 1

谢谢robthewolf的评论。是的，我读过教程，我知道我需要一个线程每个消费者。

结果表明，用一个线程读取多个交换非常简单，根本不需要轮询：获取一个新队列，并将其绑定到所有相关的交换。适用于话题和扇出。用SSCE对此进行了测试，见下文。

我对RabbitMQ javadoc中缺少细节感到遗憾，在channel#queueBind(String，String，String)方法中使用几个选择词会有很大帮助。

HTH

package rabbitExample;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

/**
 * Demonstrates reading messages from two exchanges via a single queue monitored
 * by a single thread.
 * 
 */
public class MultiExchangeReadTest implements Runnable {

private final String exch1 = "my.topic.exchange";
private final String exch2 = "my.fanout.exchange";
private final Channel channel;
private final QueueingConsumer consumer;

public MultiExchangeReadTest(final String mqHost) throws Exception {

    // Connect to server
    System.out.println("Connecting to host " + mqHost);
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost(mqHost);
    Connection connection = factory.newConnection();
    channel = connection.createChannel();

    // Declare exchanges; use defaults for durable etc.
    channel.exchangeDeclare(exch1, "topic");
    channel.exchangeDeclare(exch2, "fanout");

    // Get a new, unique queue name
    final String queue = channel.queueDeclare().getQueue();

    // Bind the queue to the exchanges; topic gets non-empty routing key
    channel.queueBind(queue, exch1, "my.key");
    channel.queueBind(queue, exch2, "");

    // Configure the channel to fetch one message at a time, auto-ACK
    channel.basicQos(1);
    consumer = new QueueingConsumer(channel);
    channel.basicConsume(queue, true, consumer);
}

public void run() {
    // Reads messages until interrupted
    try {
        while (true) {
            // Wait for a message
            System.out.println("Awaiting message");
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            // Show contents using default encoding scheme
            String body = new String(delivery.getBody());
            System.out.println("Message from exch "
                    + delivery.getEnvelope().getExchange() + ", key '"
                    + delivery.getEnvelope().getRoutingKey() + "':\n"
                    + body);
        } // while
    } catch (Exception ex) {
        ex.printStackTrace();
    }
}

public static void main(String[] args) throws Exception {
    if (args.length != 1) {
        System.err
                .println("Usaage: MultiExchangeReadTest.main mq-host-name");
    } else {
        MultiExchangeReadTest multiReader = new MultiExchangeReadTest(
                args[0]);
        multiReader.run();
    }
}
}