失去连接后 Spring AWS SQS 重新连接

Question

我正在使用 Spring Cloud AWS (1.0.1.RELEASE) 和 Spring Boot 来运行 SQS 消费者。应用程序运行良好，但是当它失去网络连接时（例如，如果我在笔记本电脑上运行时关闭 WIFI），我在控制台上看到错误并且应用程序永远无法恢复。它只是挂在那里并且在网络可用后不会重新连接。我必须杀了它并把它提出来。如何强制它自行恢复？

// Spring Boot entry point: 
public static void main(String[] args) {
    SpringApplication.run(MyConsumerConfiguration.class, args);
}

// Message Listener (A different class)
@MessageMapping(value = "myLogicalQueueName" )
public void receive(MyPOJO object) {

}

我在控制台看到的错误：

线程“simpleMessageListenerContainer-1”com.amazonaws.AmazonClientException 中的异常：无法执行 HTTP 请求：sqs.us-east-1.amazonaws.com 在 com.amazonaws.http.AmazonHttpClient.executeHelper(AmazonHttpClient.java:473) 在 com.amazonaws.http.AmazonHttpClient.execute(AmazonHttpClient.java:297) 在 com.amazonaws.services.sqs.AmazonSQSClient.invoke(AmazonSQSClient.java:2422) 在 com.amazonaws.services.sqs.AmazonSQSClient.receiveMessage(AmazonSQSClient.java:1130) 在 com.amazonaws.services.sqs.AmazonSQSAsyncClient$23.call(AmazonSQSAsyncClient.java:1678) 在 com.amazonaws.services.sqs.AmazonSQSAsyncClient$23.call(AmazonSQSAsyncClient.java:1676) 在 java.util.concurrent.FutureTask.run(FutureTask.java:266) 在 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 在 java.lang.Thread.run(Thread.java:745

Answer 1

我刚刚发现SQS在网络连接丢失后无法重新连接的问题。

实际上在org.springframework.cloud.aws.messaging.listener.SimpleMessageListenerContainer.java的当前Spring AWS实现中似乎是一个问题

private class AsynchronousMessageListener implements Runnable {

    private final QueueAttributes queueAttributes;
    private final String logicalQueueName;

    private AsynchronousMessageListener(String logicalQueueName, QueueAttributes queueAttributes) {
        this.logicalQueueName = logicalQueueName;
        this.queueAttributes = queueAttributes;
    }

    @Override
    public void run() {
        while (isRunning()) {
            ReceiveMessageResult receiveMessageResult = getAmazonSqs().receiveMessage(this.queueAttributes.getReceiveMessageRequest());
            CountDownLatch messageBatchLatch = new CountDownLatch(receiveMessageResult.getMessages().size());
            for (Message message : receiveMessageResult.getMessages()) {
                if (isRunning()) {
                    MessageExecutor messageExecutor = new MessageExecutor(this.logicalQueueName, message, this.queueAttributes);
                    getTaskExecutor().execute(new SignalExecutingRunnable(messageBatchLatch, messageExecutor));
                } else {
                    break;
                }
            }
            try {
                messageBatchLatch.await();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

上面的代码启动了一个新线程，该线程对 SQS 队列进行轮询以获取消息。一旦网络连接断开getAmazonSqs().receiveMessage(this.queueAttributes.getReceiveMessageRequest()) 就会抛出UnknownHostException，这不会在代码中处理并导致线程终止。 所以以后建立网络连接时，没有线程轮询队列来检索数据。

我已经为此提出了 Spring 的问题。以下是链接：https://github.com/spring-cloud/spring-cloud-aws/issues/82

希望这能解释一切。