Мой SQSListener возвращает тело в виде messageID, и как мне получить SendId MessageId с помощью Spring Boot QueueMessagingTemplate? - PullRequest
Новая
       81

Мой SQSListener возвращает тело в виде messageID, и как мне получить SendId MessageId с помощью Spring Boot QueueMessagingTemplate?

0 голосов
/ 03 февраля 2020

У меня есть 2 проблемы, касающиеся Spring AWS -SQS sdk, (или, возможно, я делаю это неправильно).

Во-первых, раньше я использовал CLI и мне удалось получить идентификатор сообщения отправленного пример сообщения: 

aws sqs send-message --queue-url https://sqs.us-west-2.amazonaws.com/testqueue --message-body hooray   

'{
    "MD5OfMessageBody": "d3101ad", 
    "MessageId": "jdhj-933"
}

Теперь я попытался с помощью spring-cloud-starter-aws-messaging, и я настроил шаблон обмена сообщениями очереди, подобный этому 

private final QueueMessagingTemplate queueMessagingTemplate;

  public SqsQueueService(@Qualifier("amazonSQSAsync") final AmazonSQSAsync amazonSQS) {
    this.queueMessagingTemplate = new QueueMessagingTemplate(amazonSQS);
  }

  public void sendMessage(String queueName, String queueMessage) {
    Map<String, Object> headers = new HashMap<>();
    queueMessagingTemplate.convertAndSend(queueName, queueMessage, headers);
  }

Может показаться, что идентификатор сообщения отправленного сообщения был получен с помощью queueMessagingTemplate.convertAndSend(queueName, queueMessage, headers);

Мне нужен messageId для выполнения бизнес-логики c.

Вторая проблема заключается в том, что мой слушатель может принимать сообщения, однако messageID также является нулевым; 

  @Async
  @SqsListener(value = "${notification.sqs-queue-url}", deletionPolicy = SqsMessageDeletionPolicy.NEVER)
  public void listen(Acknowledgment acknowledgment, String message, String messageId) {

    //messageId is equal to message here. which is wrong for me
  }

Сообщение всегда равно messageId, что сбивает с толку. Любой совет, где я могу пойти не так?

1 Ответ

0 голосов
/ 13 февраля 2020

Я изменил подпись метода listner на 

 @Async
  @SqsListener(value = "${queue-url}", deletionPolicy = SqsMessageDeletionPolicy.NEVER)
  public void listen(Acknowledgment acknowledgment, String message, @Headers MessageHeaders headers) throws ExecutionException, InterruptedException {
    String messageId = (String) headers.get("MessageId");
    acknowledgment.acknowledge().get();
  }

Затем извлек messageId из карты заголовков

...