Spring Boot 中具有不同凭证的多个 AWS SQS 队列

Question

我有一个 Spring Boot 应用程序，想要从多个 AWS SQS 队列接收消息。这些队列都有自己的凭据（遗憾的是我对此无能为力）。这些凭证都不能访问其他队列之一，它们都仅限于一个队列。只需一个队列和凭据，这很简单。我只需提供 Bean 形式的凭据并用\AWSCredentialsProvider注释我的方法。 但我不知道如何使用多个凭据来做到这一点。@SqsListener@EnableSqs注释@SqsListener无法提供凭据、预配置AmazonSqs对象或其他任何有帮助的内容。我寻找一种通过扩展CredentialsProvider或AmazonSqs客户端来将队列映射到凭据的方法，但无济于事。我什至尝试将凭证注入 AmazonHttpClient 的标头中，但这也是不可能的。我尝试创建手动监听 SQS 队列所需的一切。但我坚持为SimpleMessageListenerContainer.所需的QueueMessageHandler仅在使用应用程序上下文创建为 bean 时才有效。否则它不会查找用 注释的方法@SqsListener。遗憾的是，我能找到的唯一教程或示例要么使用 JMS（我想避免使用 JMS），要么仅使用@SqsListener一个队列的注释。还有其他方法可以为多个队列提供不同的凭据吗？

Answer 1

在花了一些时间寻找更好的解决方案后，我坚持以下几点：

package test;

import com.amazonaws.auth.AWSCredentialsProvider;

import com.amazonaws.regions.AwsRegionProvider;

import com.amazonaws.services.sqs.AmazonSQSAsync;

import com.amazonaws.services.sqs.AmazonSQSAsyncClientBuilder;

import test.TestDto;

import test.CustomQueueMessageHandler;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.cloud.aws.messaging.config.SimpleMessageListenerContainerFactory;

import org.springframework.cloud.aws.messaging.config.annotation.NotificationMessage;

import org.springframework.cloud.aws.messaging.config.annotation.NotificationSubject;

import org.springframework.cloud.aws.messaging.listener.SimpleMessageListenerContainer;

import org.springframework.cloud.aws.messaging.listener.SqsMessageDeletionPolicy;

import org.springframework.cloud.aws.messaging.listener.annotation.SqsListener;

import org.springframework.stereotype.Component;

@Component

public class TestQueue {

  private static final String QUEUE_NAME = "TestQueue";

  private static final Logger log = LoggerFactory.getLogger(TestQueue.class);

  public TestQueue(AWSCredentialsProvider credentialsProvider, AwsRegionProvider regionProvider) {

    AmazonSQSAsync client = AmazonSQSAsyncClientBuilder.standard()

        .withCredentials(credentialsProvider)

        .withRegion(regionProvider.getRegion())

        .build();

    // custom QueueMessageHandler to initialize only this queue

    CustomQueueMessageHandler queueMessageHandler = new CustomQueueMessageHandler();

    queueMessageHandler.init(this);

    queueMessageHandler.afterPropertiesSet();

    SimpleMessageListenerContainerFactory factory = new SimpleMessageListenerContainerFactory();

    factory.setAmazonSqs(client);

    factory.setQueueMessageHandler(queueMessageHandler);

    SimpleMessageListenerContainer simpleMessageListenerContainer = factory.createSimpleMessageListenerContainer();

    simpleMessageListenerContainer.setMessageHandler(queueMessageHandler);

    try {

      simpleMessageListenerContainer.afterPropertiesSet();

    } catch (Exception e) {

      throw new RuntimeException(e);

    }

    simpleMessageListenerContainer.start();

  }

  @SqsListener(value = QUEUE_NAME, deletionPolicy = SqsMessageDeletionPolicy.ON_SUCCESS)

  public void receiveMessage(@NotificationMessage TestDto dto, @NotificationSubject String subject) {

    log.info("Received SQS Message: \nSubject: {} \n{}", subject, dto);

  }

}

和习俗QueueMessageHandler：

package test;

import java.util.Collections;

import org.springframework.cloud.aws.messaging.listener.QueueMessageHandler;

import org.springframework.messaging.converter.MappingJackson2MessageConverter;

public class CustomQueueMessageHandler extends QueueMessageHandler {

  public void init(Object handler) {

    detectHandlerMethods(handler);

  }

}

的唯一目的CustomQueueMessageHandler是传递一个应该扫描 SQS 注释的对象。由于我不使用 Spring Context 启动它，因此它不会在每个 bean 中搜索注释@SqsListener。但所有的初始化都隐藏在受保护的方法后面。这就是为什么我需要覆盖该类以访问这些 init 方法。

我不认为这是一个非常优雅的解决方案，手动创建所有 AWS 客户端内容并调用 bean init 方法。但这是我能找到的唯一仍然可以访问 AWS SQS 库的所有功能的解决方案，例如转换传入消息和通知、删除策略、队列轮询（包括故障处理）等。

Answer 2

@Bean您可以为您希望使用 custom 的帐户声明不同的s @Qualifier。假设您在两个不同的帐户中有两个 SQS 队列。然后声明两个类型为 的 bean AmazonSQS。

@Qualifier("first")

@Bean public AmazonSQS amazonSQSClient() {

    return AmazonSQSClient.builder()

            .withCredentials(credentialsProvider())

            .withRegion(Regions.US_EAST_1)

            .build();

}

@Qualifier("second")

@Bean public AmazonSQS amazonSQSClient() {

    return AmazonSQSClient.builder()

            .withCredentials(anotherCredentialsProvider())

            .withRegion(Regions.US_EAST_1)

            .build();

}

然后在你的服务中，你可以为@Autowired他们服务。

@Autowired @Qualifier("second") private AmazonSQS sqsSecond;