等待 RabbitTemplate 相关确认

Question

我在 RabbitMQ class 中有这个函数用于发布消息：

    @Retryable(value = { AmqpIOException.class, AmqpTimeoutException.class, AmqpConnectException.class, AmqpReplyTimeoutException.class },
            maxAttempts = 3,
            backoff = @Backoff(delay = 1000))
    public void publish(Message msg, SuccessCallback<? super CorrelationData.Confirm> successCallback, FailureCallback failureCallback, String customRoutingKey) {
        CorrelationData data = new CorrelationData();
        data.getFuture().addCallback(successCallback, failureCallback);
        rabbitTemplate.convertAndSend(exchange.getName(), customRoutingKey, msg, data);
    }

下面是我如何使用上面的这个函数：

public void sendMessageToRMQ(MyEntity msgObj, String routingKey) throws ServiceUnavailableException {

        try {
            rmqPublisher.publish(
                    MessageBuilder
                            .withBody(RMQPayloadBuilder.buildPayload(msgObj))
                            .build(),
                    confirmation -> {
                        if (confirmation != null && confirmation.isAck()) {
                            msgObj.setSentDate(OffsetDateTime.now(ZoneOffset.UTC));

                            repository.save(msgObj);
                        } else {
                            StructuredLogger.logErrorMessage("Message un-ack", logValues);
                        }
                    },
                    failure -> StructuredLogger.logErrorMessage("un able to send", logValues),
                    routingKey
            );
        } catch (AmqpException amqpException) {
            StructuredLogger.logErrorMessage("Unable to publish message", logValues);
        }
    }

上面的函数在for循环中用于为大量的msgObjects发布消息。

我想要实现的是这样的：

// msgObjArray is a chunk of the whole msgObjArray

for (MyEntity msgObj : msgObjArrayChunk) {
            try {
                
                this.service.sendMessageToRMQ(msgObj, this.routingKey);


            } catch (ServiceUnavailableException e) {

                StructuredLogger.logErrorMessage("Failed to send ");
            }
        }

// What Should I write here to wait for the above chunk of msgObj to be acked and finished processing

Answer 1

您需要 return 来自 publish 的 CorrelationData 并将它们收集到一个列表中；然后，在发送完成后，迭代它们并使用 data.getFuture().get( <some timeout> ) 等待每个确认。

等待 RabbitTemplate 相关确认

Wait for RabbitTemplate Correlated confirms

java

rabbitmq

spring-rabbit

spring-boot