使用Spring Boot侦听消息队列SQS不能与标准配置一起工作

Question

我无法使用Spring Boot和SQS创建works队列侦听器（消息已发送并显示在SQS ui中）

@messagemapping或@sqslistener不工作

Java:11
Spring Boot:2.1.7
依赖关系：spring-cloud-aws-messaging

@Configuration
@EnableSqs
public class SqsConfig {

    @Value("#{'${env.name:DEV}'}")
    private String envName;

    @Value("${cloud.aws.region.static}")
    private String region;

    @Value("${cloud.aws.credentials.access-key}")
    private String awsAccessKey;

    @Value("${cloud.aws.credentials.secret-key}")
    private String awsSecretKey;

    @Bean
    public Headers headers() {
        return new Headers();
    }

    @Bean
    public MessageQueue queueMessagingSqs(Headers headers,
                                          QueueMessagingTemplate queueMessagingTemplate) {
        Sqs queue = new Sqs();
        queue.setQueueMessagingTemplate(queueMessagingTemplate);
        queue.setHeaders(headers);
        return queue;
    }

    private ResourceIdResolver getResourceIdResolver() {
        return queueName -> envName + "-" + queueName;
    }

    @Bean
    public DestinationResolver destinationResolver(AmazonSQSAsync amazonSQSAsync) {
        DynamicQueueUrlDestinationResolver destinationResolver = new DynamicQueueUrlDestinationResolver(
                amazonSQSAsync,
                getResourceIdResolver());
        destinationResolver.setAutoCreate(true);
        return destinationResolver;
    }

    @Bean
    public QueueMessagingTemplate queueMessagingTemplate(AmazonSQSAsync amazonSQSAsync,
                                                         DestinationResolver destinationResolver) {
        return new QueueMessagingTemplate(amazonSQSAsync, destinationResolver, null);
    }

    @Bean
    public QueueMessageHandlerFactory queueMessageHandlerFactory() {
        QueueMessageHandlerFactory factory = new QueueMessageHandlerFactory();
        MappingJackson2MessageConverter messageConverter = new MappingJackson2MessageConverter();
        messageConverter.setStrictContentTypeMatch(false);
        factory.setArgumentResolvers(Collections.singletonList(new PayloadArgumentResolver(messageConverter)));
        return factory;
    }

    @Bean
    public SimpleMessageListenerContainerFactory simpleMessageListenerContainerFactory(AmazonSQSAsync amazonSqs) {
        SimpleMessageListenerContainerFactory factory = new SimpleMessageListenerContainerFactory();
        factory.setAmazonSqs(amazonSqs);
        factory.setMaxNumberOfMessages(10);
        factory.setWaitTimeOut(2);
        return factory;
    }

}

@RunWith(SpringJUnit4ClassRunner.class)
public class ListenTest {

    @Autowired
    private MessageQueue queue;

    private final String queueName = "test-queue-receive";

    private String result = null;

    @Test
    public void test_listen() {
        // given
        String data = "abc";

        // when
        queue.send(queueName, data).join();

        // then
        Awaitility.await()
                .atMost(10, TimeUnit.SECONDS)
                .until(() -> Objects.nonNull(result));

        Assertions.assertThat(result).equals(data);
    }

    @MessageMapping(value = queueName)
    public void receive(String data) {
        this.result = data;
    }
}

你觉得有什么不对劲吗？

我为示例创建了一个repo:(https://github.com/mmaryo/java-sqs-test)
在测试文件夹中，在'application.yml'中更改aws凭据
然后运行测试

Answer 1

我在使用spring-cloud-aws-messaging包时也遇到了同样的问题，但随后我在@sqslistener注释中使用了队列URL而不是队列名称，结果它起了作用。

@SqsListener(value = { "https://full-queue-URL" }, deletionPolicy = SqsMessageDeletionPolicy.ON_SUCCESS)
public void receive(String message) {
     // do something
}

在使用spring-cloud-starter-aws-messaging包时似乎可以使用队列名。如果您不想使用starter包，我相信有一些配置允许使用队列名而不是URL。

编辑：我注意到该区域被默认为us-west-2，尽管我在属性文件中列出了us-east-1。然后，我创建了一个RegionProvider bean，并在其中将区域设置为us-east-1，现在，当我在@SQSMessaging中使用队列名时，就会在框架代码中找到该队列名并将其正确解析为URL。