SQS异步消费者连接异常检测与恢复方案问询

问题背景 #

我在应用中使用了异步SQS Consumer，基于官方文档实现的代码如下：

消息监听器实现 #

class MyListener implements MessageListener {
 
    @Override
    public void onMessage(Message message) {
        try {
            // 将收到的消息转换为TextMessage并打印内容
            System.out.println("Received: " + ((TextMessage) message).getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

消费者启动代码 #

// 创建基于配置的连接工厂
SQSConnectionFactory connFactory = new SQSConnectionFactory(new ProviderConfiguration(),
        AmazonSQSClientBuilder.defaultClient());
try {
    // 创建连接
    SQSConnection sqsConn = connFactory.createConnection();

    // 创建会话
    Session session = sqsConn.createSession(false, Session.CLIENT_ACKNOWLEDGE);
    MessageConsumer consumer = session.createConsumer(session.createQueue(pdfResponseQueueName));

    MyListener sqsMessageListener = new MyListener();
    consumer.setMessageListener(sqsMessageListener);

    sqsConn.start();
} catch (JMSException e) {
    LOG.error("Can't start SQSMessageListener.", e);
}

近期发现消费者会长时间闲置，不处理任何消息（AWS控制台可见队列消息堆积），疑似连接中断但消费者无任何报错提示。

查阅文档得知SQS是JMS的封装，JMS的Connection接口提供setExceptionListener方法用于连接出现严重问题时通知监听器，但SQSConnection的JavaDoc标注“不支持连接上的异常监听器”，且该方法未抛出UnsupportedOperationException，仅实现了赋值逻辑，无法起到预期作用。

请问在SQS异步监听器中，正确检测并解决连接问题的方式是什么？

解决方案 #

1. 强化SQS客户端的重试策略 #

AWS SQS Java客户端内置了重试机制，但默认配置可能无法覆盖所有连接异常场景。初始化客户端时可自定义重试策略，针对连接超时、Socket异常等场景触发重试：

AmazonSQS sqsClient = AmazonSQSClientBuilder.standard()
        .withRetryPolicy(new RetryPolicy(
                PredefinedRetryPolicies.DEFAULT_RETRY_CONDITION,
                PredefinedRetryPolicies.DEFAULT_BACKOFF_STRATEGY,
                5, // 最大重试次数
                true)) // 是否重试请求
        .build();

2. 主动定时检测连接有效性 #

由于SQSConnection的异常监听器无效，可通过定时任务主动验证连接状态：

• 定期调用sqsClient.getQueueAttributes()获取队列基础信息（如ApproximateNumberOfMessages），若调用失败则判定连接异常。
• 检测到异常时，销毁现有连接、会话和消费者，重新初始化整个消费者链路。

3. 增强消息监听器的异常捕获 #

在onMessage方法中扩展异常捕获范围，捕获底层SDK抛出的连接类异常，并触发连接重建：

@Override
public void onMessage(Message message) {
    try {
        TextMessage textMessage = (TextMessage) message;
        System.out.println("Received: " + textMessage.getText());
    } catch (JMSException e) {
        LOG.error("JMS处理消息失败", e);
        triggerReconnect();
    } catch (RuntimeException e) {
        if (e.getCause() instanceof SdkClientException) {
            LOG.error("SQS连接异常", e);
            triggerReconnect();
        }
        throw e;
    }
}

4. 封装连接生命周期管理 #

将消费者的初始化、销毁、重启逻辑封装成工具类，便于异常时快速重建：

public class SQSConsumerManager {
    private SQSConnection sqsConn;
    private Session session;
    private MessageConsumer consumer;
    private final String queueName;
    private final MessageListener listener;
    private final AmazonSQS sqsClient;
    private static final Logger LOG = LoggerFactory.getLogger(SQSConsumerManager.class);

    public SQSConsumerManager(String queueName, MessageListener listener, AmazonSQS sqsClient) {
        this.queueName = queueName;
        this.listener = listener;
        this.sqsClient = sqsClient;
    }

    public void start() throws JMSException {
        SQSConnectionFactory connFactory = new SQSConnectionFactory(new ProviderConfiguration(), sqsClient);
        sqsConn = connFactory.createConnection();
        session = sqsConn.createSession(false, Session.CLIENT_ACKNOWLEDGE);
        consumer = session.createConsumer(session.createQueue(queueName));
        consumer.setMessageListener(listener);
        sqsConn.start();
    }

    public void stop() {
        try {
            if (consumer != null) consumer.close();
            if (session != null) session.close();
            if (sqsConn != null) sqsConn.close();
        } catch (JMSException e) {
            LOG.error("关闭消费者资源失败", e);
        }
    }

    public void restart() {
        stop();
        try {
            start();
            LOG.info("SQS消费者已重启");
        } catch (JMSException e) {
            LOG.error("重启SQS消费者失败", e);
        }
    }
}

检测到异常时调用restart()方法即可完成链路重建。

5. 基于CloudWatch指标监控告警 #

通过AWS CloudWatch监控队列的ApproximateNumberOfMessages指标，当该指标持续增长且消费者无消息处理日志时，触发告警，手动或自动执行消费者重启操作。

内容的提问来源于stack exchange，提问作者Ihor M.