等待 RabbitTemplate 相关确认

问题描述

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

    @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

解决方法

您需要从 CorrelationData 返回 publish 并将它们收集到列表中；然后，在发送完成后，遍历它们并使用 data.getFuture().get( <some timeout> ) 等待每个确认。