1

我正在研究一个基本示例，但无法解决。

我需要通过队列（TestQ）将消息从一台机器（Machine1）转发到另一台（Machine2）。生产者在 Machine1 上运行，消费者在 Machine2 上运行。

我在 Machine1 的 rabbit broker 配置中的设置：

{rabbitmq_shovel, [ {shovels, [
    {shovel_test, [
        {sources, [{broker, "amqp://" }]},
        {destinations, [{broker, "amqp://Machine2" }]},
        {queue, <<"TestQ">>},
        {ack_mode, on_confirm},
        {reconnect_delay, 5}
    ]}
]} ]}

Machine2 有一个默认配置，没有启用铲子插件。

在 Machine1 上运行的生产者代码：

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();       
channel.queueDeclare("TestQ", true, false, false, null);   
channel.basicPublish("", "TestQ", null, "Hello World!".getBytes());

在 Machine2 上运行的消费者代码：

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.queueDeclare("TestQ", true, false, false, null);
QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume("TestQ", true, consumer);

while (true) {
    QueueingConsumer.Delivery delivery = consumer.nextDelivery();
    String message = new String(delivery.getBody());
    System.out.println(" [x] Received '" + message + "'");
}

执行 rabbitmqctl eval 'rabbit_shovel_status:status().' 在机器 1 上：

[{shovel_test,starting,{{2014,1,7},{9,47,38}}}]
...done.

Producer sends ok, but I never get a receive from the consumer on the Machine2.

Where is a problem? Something is missing in the conf of Machine1's broker, or Machine2's broker?

Thank you!

4

1 回答 1

2

你的铲子的状态应该是running，不是starting。如果它停留在该starting阶段，则意味着它无法正确启动（例如，无法连接到目标代理）。

我发现的一个问题是您使用broker而不是brokers指定源列表。尝试这个：

{rabbitmq_shovel,
 [{shovels, [{shovel_test,
              [{sources, [{brokers, ["amqp://"]}]},
               {destinations, [{broker, "amqp://Machine2"}]},
               {queue, <<"TestQ">>},
               {ack_mode, on_confirm},
               {reconnect_delay, 5}
              ]}
            ]}
 ]}.
于 2014-01-18T11:45:01.677 回答