一个队列上的多个消费者 RabbitMQ - Java

Question

我是新手 RabbitMQ java 客户端。 我的问题：我创建了 10 个消费者并将它们添加到队列中。每个消费者使用 10 秒来处理我的流程。我检查了兔子的页面，我看到我的队列有 4000 条消息没有发送给客户端。我检查了日志客户端，结果是为一位消费者获取了一条消息，10 秒后我为一位消费者获取了一条消息，依此类推..我想同时为所有消费者获取 10 条消息（10 条消息-当时 10 条消费者进程) 请帮助我，我没有找到问题的解决方案。 非常感谢。

        while (!isRetry) {
        try {
            isRetry = true;
            connection = mConnectionFactory.newConnection(addresses.toArray(new Address[addresses.size()]));
            String queueName = "webhook_customer";
            String exchangeName = "webhook_exchange";
            String routingKey = "customer";
            System.out.println("step2");

            Channel channel = connection.createChannel();
            channel.exchangeDeclare(exchangeName, "topic", true);
            channel.queueDeclare(queueName, true, false, false, null);
            channel.queueBind(queueName, exchangeName, routingKey);
            channel.basicQos(1);
            for (int i = 0; i < numberWorker; i++) {
                Consumer consumer = new QueueingConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope,
                                               AMQP.BasicProperties properties, byte[] body) throws IOException {
                        long startProcess = System.nanoTime();
                        JSONObject profile = null;
                        try {

                        } catch (IOException ioe) {
                            handleLogError(profile, ioe.getMessage().toString());
                        } catch (Exception e) {
                            handleLogError(profile, e.getMessage());
                        } finally {
                            channel.basicAck(envelope.getDeliveryTag(), false);
                            long endProcess = System.nanoTime();
                            _logger.info("===========######### TIME PROCESS  + " + (endProcess - startProcess) + " Nano Seconds  ========#### " + (endProcess - startProcess) / 1000000 + " Milli Seconds");
                        }
                    }
                };

                channel.basicConsume(queueName, false, consumer);
            }
            System.out.printf("Start Listening message ...");
        } catch (Exception e) {
            System.out.println("exception " + e.getMessage());
            isRetry = closeConnection(connection);
            e.printStackTrace();
        } finally {
        }
        if (!isRetry) {
            try {
                System.out.println("sleep waiting retry ...");
                Thread.sleep(30000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        //END
    }

Answer 1

我确实找到了解决方案。当消息进入并在其中处理时，我在消费者中使用新线程。我创建了多个频道以便同时发送多个消息。我使用线程池来控制线程

Answer 2

从您的代码示例看来，您可以使用QueueingConsumer 而不是DefaultConsumer。这将从 RabbitMQ 拉出更多消息给消费者并将它们排队，直到它们被处理。

然后，在您的 for (int i = 0; i < 10; i++) 循环中，您使用相同的消费者实例消费了 10 次。相反，您应该创建 10 个消费者：

for (int i = 0; i < 10; i++) {
    Consumer consumer = new DefaultConsumer(channel) {
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
          channel.basicAck(envelope.getDeliveryTag(),false);
        }
    };

    channel.basicConsume(queueName, false, consumer);
}

理想情况下，创建另一个类并在循环中正确创建新实例而不是匿名实例。

注意：您的消费者应该在后台（单独的线程）执行他们的进程，否则他们会互相阻塞。虽然，您提供的信息并没有真正表明您将如何实际处理这些消息。