OutOfMemoryError в SQSMessageConsumerPrefetch - PullRequest
Новая
       28

OutOfMemoryError в SQSMessageConsumerPrefetch

0 голосов
/ 08 января 2020

У меня есть приложение SpringBoot, которое отправляет сообщения в 3 отдельные очереди SQS после предварительной обработки и одновременно использует их для дальнейшей их обработки отдельно в зависимости от очереди.

Мы периодически получаем эту ошибку, которая в конечном итоге приводит к сбою приложения. 

2020-01-07 19:44:32.484 [ConsumerPrefetchThread-1] ERROR c.a.s.j.SQSMessageConsumerPrefetch - Unexpected exception when prefetch messages: java.lang.OutOfMemoryError: Java heap space

Вот и все сообщение.

Мы настроили количество слушателей очереди как 5; у нас есть 3 очереди; Во время профилирования я вижу, что создано 7 ConsumerPrefetchThread с. Ожидается ли это?

Также с точки зрения SQS или даже обобщенной c, что может быть виновником ошибок OOM? Я открыт для любой теории!

Соответствующие зависимости в build.gradle:

  • org.springframework: spring-jms: 5.1.1.RELEASE
  • com .amazon aws: aws - java -sdk-sqs: 1.11.428
  • com.amazon aws: amazon-sqs- java -messaging-lib: 1.0.4

Фрагмент кода из SQSConfiguration. java

@Bean
public SQSConnectionFactory sqsConnectionFactory() {
    try {
        AmazonSQSClientBuilder.defaultClient()
                .getQueueUrl({This is the queue url});
    }
    catch (Exception ex) {
        throw new ApplicationLoadingException(
                "SQS settings error: " + ex.getStackTrace());
    }
    return new SQSConnectionFactory(new ProviderConfiguration(),
            AmazonSQSClientBuilder.defaultClient());
}

@Bean
public JmsTemplate jmsTemplate() {
    JmsTemplate template = new JmsTemplate();
    template.setConnectionFactory(sqsConnectionFactory());
    return template;
}

А Вот фрагмент кода из SQSClient. java

@PostConstruct
public void init() {
    sqsClient = AmazonSQSClientBuilder.defaultClient();
}

@Retryable(value = {
        SdkClientException.class }, maxAttempts = 3, backoff = @Backoff(delay = 30000))
public void sendMessage(String queue, String message, int delayInMinutes) {
    SendMessageRequest request = new SendMessageRequest(
            sqsClient.getQueueUrl(queue).getQueueUrl(), message);
    request.withDelaySeconds(delayInMinutes * 60);
    sqsClient.sendMessage(request);
}
...