Когда издатель подтверждает, что включены, предел длины очереди установлен, а переполнение установлено, чтобы отклонить публикацию, почему причина в обратном вызове подтверждения, который я получил, является нулевой? - PullRequest
Новая
       11

Когда издатель подтверждает, что включены, предел длины очереди установлен, а переполнение установлено, чтобы отклонить публикацию, почему причина в обратном вызове подтверждения, который я получил, является нулевой?

1 голос
/ 04 октября 2019

Я изучаю ограничение длины очереди (https://www.rabbitmq.com/maxlength.html),, как говорится, очередь установлена ​​на 'x-max-length:10' и 'x-overflow:reject-publish', а также, я включаю издателя, подтверждает, поэтому, когда количество сообщений вочередь достигает 10, издатель будет проинформирован об отклонении через сообщение basic.nack, и это: мой обратный вызов подтверждения получил ложное подтверждение, но причина равна нулю, мне интересно, не должно ли он вернуть что-то, чтобы я могРазличают эту ситуацию. Часть кода выглядит следующим образом: 

  @Bean
  public AmqpTemplate amqpTemplate(@Autowired CachingConnectionFactory amqpConnectionFactory) {
    amqpConnectionFactory.setPublisherReturns(true);
    amqpConnectionFactory.setPublisherConfirms(true);
    RabbitTemplate rabbitTemplate = new RabbitTemplate(amqpConnectionFactory);
    rabbitTemplate.setMessageConverter(jsonMessageConverter());
    rabbitTemplate.setConfirmCallback(confirmCallback);
    rabbitTemplate.setReturnCallback(returnCallback);
    return rabbitTemplate;
  }

  static RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
      System.out.println(ack);  // when number of messages reach 10, print false
      System.out.println(cause); // when number of messages reach 10, print null
    }
  };

 @Bean
  public Queue queue() {
    return QueueBuilder.durable(DURABLE_QUEUE).withArgument("x-max-length", 10).withArgument("x-overflow", "reject-publish").build();
  }

 @Scheduled(fixedDelay = 1000L)
  public void produce() {
    Message msg = new Message(UUID.randomUUID().toString(), "sth");
    amqpTemplate.convertAndSend("sth", "sth", msg );
  }

1 Ответ

0 голосов
/ 04 октября 2019

К сожалению, протокол AMQP и клиент Java не предоставляют информации о том, почему публикация не удалась. Только ack / nack и подтверждение наличия нескольких сообщений: 

/**
 * Implement this interface in order to be notified of Confirm events.
 * Acks represent messages handled successfully; Nacks represent
 * messages lost by the broker.  Note, the lost messages could still
 * have been delivered to consumers, but the broker cannot guarantee
 * this.
 * For a lambda-oriented syntax, use {@link ConfirmCallback}.
 */
public interface ConfirmListener {
    void handleAck(long deliveryTag, boolean multiple)
        throws IOException;

    void handleNack(long deliveryTag, boolean multiple)
        throws IOException;
}

Мы добавили cause, потому что в некоторых случаях платформа синтезирует nack (например, когда канал закрыт, пока мыв ожидании подтверждений, где мы добавляем Channel closed by application в качестве cause.

. Фреймворк не может предположить причину, по которой мы получили от брокера отрывок.

...