2

在 localstack 中测试我的应用程序时遇到问题。要求很简单。我想通过 SQS 订阅订阅 AWS-SNS 中的主题。在 localstack 中，当我配置 @SQSListener 时，它会尝试使用 AWS 验证凭证，我收到以下错误消息：

org.springframework.beans.factory.BeanCreationException：在类路径资源 [org/springframework/cloud/aws/messaging/config/annotation/SqsConfiguration.class] 中定义名称为“simpleMessageListenerContainer”的 bean 创建错误：调用 init 方法失败；嵌套异常是 com.amazonaws.SdkClientException: Unable to load AWS credentials from any provider in the chain:

我尝试为 localstack 设置一个虚拟凭据，但也无济于事。我尝试在网上搜索，但找不到任何相关的帖子。

代码

@Slf4j
@Configuration
@Profile("local")
@EnableSns
@EnableSqs
public class LocalStackConfiguration {


    @Value("${sqsClassifierQueueName}")
    private String sqsClassifierQueueName;

    @Value("${sqsConsQueueNames}")
    private List<String> sqsConsQueueNames;

    @Bean(name = "amazonSqs")
    public AmazonSQS configureSQSClient()
            throws MalformedURLException
    {
        log.info("Loading local SQS client");
        BasicAWSCredentials awsCreds = new BasicAWSCredentials("user", "password");

        AmazonSQS sqsClient = AmazonSQSClientBuilder.standard()
                .withCredentials(new AWSStaticCredentialsProvider(awsCreds))
                .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration("http://localhost:4576", "us-east-1"))
                .build();

        sqsClient.createQueue(sqsClassifierQueueName);
        sqsConsQueueNames.forEach(queueName -> sqsClient.createQueue(queueName));
        //log.info("Device Registry queue Created: " + sqsClient.createQueue(REGISTRY_QUEUE_URL.substring(REGISTRY_QUEUE_URL.lastIndexOf('/') + 1)));
        return sqsClient;
    }

    @Bean(name = "amazonSns")
    public AmazonSNS configureSNSClient() {
        AmazonSNS snsClient= AmazonSNSClientBuilder.standard()
                .withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials("user", "password")))
                .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration("http://localhost:4575", "us-east-1"))
                .build();
        return snsClient;
    }



}


@Component
@DependsOn(value = {"amazonSns", "amazonSqs"})
public class SNSSubscription {

    @Qualifier("amazonSns")
    @Autowired
    private AmazonSNS sns;

    @Qualifier("amazonSqs")
    @Autowired
    private AmazonSQS sqs;

    @Value("${snsClassifierTopicName}")
    private String snsClassifierTopicName;

    @Value("${sqsClassifierQueueName}")
    private String sqsClassifierQueueName;

    @PostConstruct
    public void subscribeToTopics() {
        subscribeToClassifierTopics();
    }

    private void subscribeToClassifierTopics() {
        SNSSubscriptionUtils.subscribeToSQS(sns, sqs, snsClassifierTopicName, sqsClassifierQueueName);

    }

    @SqsListener(value="${snsClassifierTopicName}")
    public void receiveMessage(Object obj){
        System.out.println(obj);
    }
}



public class SNSSubscriptionUtils {

    public static void subscribeToSQS(AmazonSNS sns, AmazonSQS sqs, String topicName, String queueName) {
        String snsTopicArn = getSnsTopicArn(sns, topicName);
        String queueUrl = SQSUtils.getQueueUrl(sqs, queueName);
        Topics.subscribeQueue(sns, sqs, snsTopicArn, queueUrl);
    }

    public static String getSnsTopicArn(AmazonSNS sns, String topicName) {
        String topicArn = sns.createTopic(topicName).getTopicArn();
        return topicArn;
    }
}


public class SQSUtils {

    public static String getQueueUrl(AmazonSQS sqs, String queueName) {
        GetQueueUrlRequest queueUrlRequest = new GetQueueUrlRequest(queueName);
        return sqs.getQueueUrl(queueUrlRequest).getQueueUrl();
    }
}
4

1 回答 1

0

使用主要注释来注释您配置的 bean。

@Bean(name = "amazonSqs")
@Primary
public AmazonSQS configureSQSClient()
于 2020-02-12T09:24:58.513 回答