【问题标题】:Rabbitmq java client consumer handleDelivery method does not get calledRabbitmq java客户端消费者handleDelivery方法没有被调用
【发布时间】:2019-07-20 14:09:38
【问题描述】:

这是一个非常人为的 rabbitmq 应用程序示例,它在一个主要方法中既是消息的生产者又是消息的消费者。问题是覆盖的 handleDelivery 方法中的代码永远不会被执行。我使用 Rabbitmq 仪表板,看到队列已满并消耗。并且 handleConsumeOk 中的行被打印出来。 由于我是 rabbitmq 的新手,我想知道我是否做错了一些根本性的错误,或者我只是认为“当收到此消费者的 basic.deliver 时调用”的想法是错误的。

public class RabbitMain {

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        Connection connection = Utils.getConnection("test");
        String payload = "hello world!";
        try (Channel channel = connection.createChannel()){
            channel.exchangeDeclare("sampleExchange", BuiltinExchangeType.TOPIC, true);
            channel.basicPublish("sampleExchange", "testKey", null, payload.getBytes());
        }

        System.out.println("Consume...");

        try (Channel channel = connection.createChannel()){
            channel.exchangeDeclare("sampleExchange", BuiltinExchangeType.TOPIC, true);
            channel.queueDeclare("testQueue", true, false, false, null);
            channel.queueBind("testQueue", "sampleExchange", "testKey");

            Consumer consumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String message = new String(body);
                    System.out.println("Received: " + message);
                }

                @Override
                public void handleConsumeOk(String consumerTag) {
                    System.out.println("handled consume ok");
                }
            };

            Thread.sleep(2000);

            channel.basicConsume("testQueue", true, consumer);
        }
    }
}

【问题讨论】:

  • 消费发生在另一个线程上。当我使用 IntelliJ idea 调试器运行代码时,代码可以正常工作。通过向主线程添加一些空闲时间(例如 Thread.sleep(1000) 消费线程有时间完成它的工作。我不确定,但似乎消费者线程是一个守护线程并在主线程终止后被杀死。

标签: java rabbitmq messaging


【解决方案1】:

您在任何队列绑定到它之前将消息发布到exchnage。 RabbitMQ 将丢弃任何无法路由到队列的消息。

channel.queueDeclare("testQueue", true, false, false, null);
channel.queueBind("testQueue", "sampleExchange", "testKey");

在您致电channel.basicPublish之前。

【讨论】:

  • 感谢您的回答。我按你说的改了代码。我仍然通过管理控制台看到消息已发布到队列并被使用。甚至 handleConsumeOk 方法中的代码也会被执行。即使我从控制台发布消息,也只会跳过 handleDelivery 方法。
猜你喜欢
  • 2020-09-20
  • 2019-04-21
  • 1970-01-01
  • 2017-01-31
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 2015-09-18
  • 1970-01-01
相关资源
最近更新 更多