Automati c повторное объявление / восстановление очереди в RabbitMQ - PullRequest
Новая
       63

Automati c повторное объявление / восстановление очереди в RabbitMQ

0 голосов
/ 31 января 2020

У меня есть приложение-потребитель RabbitMQ в java. Ниже приведен фрагмент кода для объявления очереди: 

public static void declareQueue(final String rmqQueueName) {
    try {
        Map<String, Object> args = new HashMap<>();
        args.put("x-max-priority", 10);
        channel.queueDeclare(rmqQueueName, true, false, false, args);
    } catch (final IOException e) {
        log.warn("Could not declare the queue: " + rmqQueueName + " ", e);
    }
}

Код потребителя: 

public void consumeBrokerMessage(final String rmqQueueName) throws IOException {

    declareQueue(rmqQueueName);
    log.info(" [*] Waiting for messages. To exit press CTRL+C");
    final Consumer consumer = new DefaultConsumer(channel) {

        @Override
        public void handleDelivery(
                final String consumerTag,
                final Envelope envelope,
                final AMQP.BasicProperties properties,
                final byte[] body) throws IOException {
            boolean flag = false;
            final String brokerMessage = new String(body, StandardCharsets.UTF_8);
            final long start = System.currentTimeMillis();
            final OffsetDateTime utcStartDateTime = OffsetDateTime.now(ZoneOffset.UTC);
            final String startDateTime =
                utcStartDateTime.format(
                    DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSS"));
            flag = BrokerMessage.handleBrokerMessage(utility);

            if (flag) {
                log.info("Broker Message has processed");
            } else {
                log.warn("Error while processing broker message");
            }
            final long deliveryTag = envelope.getDeliveryTag();
            // positively acknowledge a single delivery, the message will
            // be discarded
            channel.basicAck(deliveryTag, false);
        }
    };
    channel.basicQos(1);
    channel.basicConsume(rmqQueueName, false, consumer);
}

====================== 

public static Channel createChannel(
        final String rmqHost,
        final String rmqPort,
        final String rmqUser,
        final String rmqPassword,
        final String rmqVHost) {

    final ConnectionFactory factory = new ConnectionFactory();

    factory.setHost(rmqHost);
    factory.setPort(Integer.parseInt(rmqPort));
    factory.setUsername(rmqUser);
    factory.setPassword(rmqPassword);
    factory.setVirtualHost(rmqVHost);
    **factory.setAutomaticRecoveryEnabled(true);**

    final Connection connection;
    Channel rmqChannel = null;
    try {
        connection = factory.newConnection();
        rmqChannel = connection.createChannel();
    } catch (final IOException | TimeoutException e) {
        log.warn("Could not establish rabbitmq connection. ", e);
    }

    return rmqChannel;
}

Когда я запускаю свое приложение, очередь создается с максимальным приоритетом 10. Но когда я удаляю очередь из RabbitMQ Management, очередь не восстанавливается сама и не повторно объявляется автоматически. У кого-нибудь есть идеи, как автоматически восстановить и повторно объявить очередь без перезапуска приложения?

...