如果 amazonSqsAsync 无效，则阻止加载 SqsListener 和 AwsConfiguration

Question

我有一个 MySQSListener 和一个 AmazonSQSConfig 类。 （参见下面的课程）

如果由于某种原因凭据不正确，我会收到此异常：

org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'simpleMessageListenerContainer' defined in class path resource [org/springframework/cloud/aws/messaging/config/annotation/SqsConfiguration.class]: Invocation of init method failed; nested exception is com.amazonaws.services.sqs.model.AmazonSQSException: The security token included in the request is invalid. (Service: AmazonSQS; Status Code: 403; Error Code: InvalidClientTokenId; Request ID: 729e468d-0c99-56d9-a1db-1ed18e322319)

在这种情况下，我想停止加载 SqsListener 和 AmazonSQSConfig，而不是让整个应用程序崩溃。

有没有办法在加载 bean 之前测试凭据？

谢谢！

@Slf4j
@Component
@ConditionalOnExpression("${aws.sqs.enabled:false}")
public class MySQSListener {

  @SqsListener(value = "${aws.sqs.queueName}", deletionPolicy = SqsMessageDeletionPolicy.NEVER)
  public void receiveMessage(MyCustomObject customObject, Acknowledgment acknowledgment) {
     //process custom object
  }
}

和 SQSConfig：

@Slf4j
@EnableSqs
@Configuration
@ConditionalOnExpression("${aws.sqs.enabled:false}")
public class AmazonSQSConfig {

  @Bean
  public AWSCredentialsProvider awsCredentialsProvider(
      @Value("${aws.sqs.credentials.accessKey:\"UNKNOWN\"}") final String awsAccessKey,
      @Value("${aws.sqs.credentials.secretKey:\"UNKNOWN\"}") final String awsSecretKey) {
    return new AWSStaticCredentialsProvider(new BasicAWSCredentials(awsAccessKey, awsSecretKey));
  }

  @Bean
  public SimpleMessageListenerContainerFactory simpleMessageListenerContainerFactory(
      AmazonSQSAsync amazonSQSAsync,
      @Value("${aws.sqs.maxNumberOfMessages:10}") int maxNumberOfMessages,
      @Value("${aws.sqs.waitTimeOut:20}") int waitTimeOut,
      @Value("${aws.sqs.visibilityTimeOut:300}") int visibilityTimeOut) {
    SimpleMessageListenerContainerFactory factory = new SimpleMessageListenerContainerFactory();
    factory.setAmazonSqs(amazonSQSAsync);
    factory.setMaxNumberOfMessages(maxNumberOfMessages);
    factory.setWaitTimeOut(waitTimeOut);
    factory.setVisibilityTimeout(visibilityTimeOut);
    factory.setAutoStartup(true);
    return factory;
  }
 
  @Bean
  @Primary
  public AmazonSQSAsync amazonSqsAsync(AWSCredentialsProvider awsCredentialsProvider,
                                       @Value("${aws.sqs.local.endpointOverride:#{null}}")
                                       final String endpointOverride) {
    AmazonSQSAsyncClientBuilder amazonSQSAsyncClientBuilder = AmazonSQSAsyncClientBuilder.standard();
    amazonSQSAsyncClientBuilder.withCredentials(awsCredentialsProvider);
    if (endpointOverride != null) {
      LOG.info(String.format("Found endpoint override in application properties, new endpoint: %s", endpointOverride));
      amazonSQSAsyncClientBuilder.withEndpointConfiguration(
          new AwsClientBuilder.EndpointConfiguration(endpointOverride, Regions.GovCloud.getName()));
    } else {
      amazonSQSAsyncClientBuilder.withRegion(Regions.GovCloud);
    }
    return amazonSQSAsyncClientBuilder.build();
  }

  @Bean
  public QueueMessageHandlerFactory queueMessageHandlerFactory(MessageConverter messageConverter,
                                                               AmazonSQSAsync amazonSQSAsync) {
    QueueMessageHandlerFactory factory = new QueueMessageHandlerFactory();
    factory.setAmazonSqs(amazonSQSAsync);
    AcknowledgmentHandlerMethodArgumentResolver acknowledgmentResolver =
        new AcknowledgmentHandlerMethodArgumentResolver("Acknowledgment");
    PayloadArgumentResolver payloadArgumentResolver = new PayloadArgumentResolver(messageConverter);
    factory.setArgumentResolvers(Arrays.asList(acknowledgmentResolver, payloadArgumentResolver));
    return factory;
  }
  
  @Bean
  public QueueMessagingTemplate queueMessagingTemplate(AmazonSQSAsync amazonSQSAsync) {
    return new QueueMessagingTemplate(amazonSQSAsync);
  }
 
  @Bean
  protected MessageConverter messageConverter(ObjectMapper objectMapper) {
    MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
    converter.setObjectMapper(objectMapper);
    converter.setStrictContentTypeMatch(false);
    converter.setSerializedPayloadClass(String.class);
    return converter;
  }
}

Answer 1

您可以通过添加 SimpleMessageListenerContainerFactory bean 的条件创建 (@Conditional) 来实现此目的。在该条件实现中，您可以检查此连接性。并使这些 bean 自动装配需要 false