如何在Java SDK中退出SQS读取

Question

我们有一个Java Web服务,用于轮询AWS SQS队列以获取消息.它使用最长20秒的等待时间.在Tomcat中更新服务时,Tomcat会尝试关闭现有服务,然后启动更新后的版本.发生这种情况时,如果正在读取队列,则线程被卡住并且Tomcat无法阻止它,从而导致内存泄漏.从我的实验中,只有当Tomcat在队列中等待时尝试停止服务时,才会发生此内存泄漏.

如何在关机过程中停止Java中的SQS队列？

这里有一些更多的细节.

Jan 03, 2014 7:52:52 PM org.apache.catalina.loader.WebappClassLoader clearReferencesThreads
SEVERE: The web application appears to have started a thread named [pollSQSQueue-1] but has failed to stop it. This is very likely to create a memory leak.
Jan 03, 2014 7:52:52 PM org.apache.catalina.loader.WebappClassLoader checkThreadLocalMapForLeaks
SEVERE: The web application created a ThreadLocal with key of type [com.amazonaws.auth.AWS4Signer$2] (value [com.amazonaws.auth.AWS4Signer$2@71f69c90]) and a value of type [java.text.SimpleDateFormat] (value [java.text.SimpleDateFormat@ef87e460]) but failed to remove it when the web application was stopped. Threads are going to be renewed over time to try and avoid a probable memory leak.
Jan 03, 2014 7:52:52 PM org.apache.catalina.loader.WebappClassLoader checkThreadLocalMapForLeaks
SEVERE: The web application created a ThreadLocal with key of type [com.amazonaws.auth.AWS4Signer$1] (value [com.amazonaws.auth.AWS4Signer$1@75a0ec63]) and a value of type [java.text.SimpleDateFormat] (value [java.text.SimpleDateFormat@6f64295a]) but failed to remove it when the web application was stopped. Threads are going to be renewed over time to try and avoid a probable memory leak.

该服务使用Spring并使用Spring调度程序启动轮询过程.

<task:scheduler id="pollSQSQueue" pool-size="1"/>

我试过创建一个Spring ApplicationListener<ContextClosedEvent>来尝试关闭队列.我的第一次尝试是尝试关闭队列,看看它是否正常处理它.它没.

@Override
public void onApplicationEvent(ContextClosedEvent contextClosedEvent)
{
    amazonSQS.shutdown();
}

我尝试的下一件事是使用AmazonSQSAsync并使用异步检索.然后我获得了Future<ReceiveMessageResult>并试图cancel在Spring关闭期间调用它.这也没有得到很好的处理.

---

此外,我知道需要关闭IdleConnectionReaper并在Spring关闭时执行此操作,因此我认为它与此无关.

Answer 1

您可以创建一个单独的线程来进行轮询工作。沿着这些思路：

class MyService extends/implements ... {
    private SQSPollingThread pollingThread;

    //this is just my guess of what the "service start" method looks like
    @Override
    public void onApplicationEvent(ContextStartedEvent contextStartedEvent)
    {
         pollingThread = new SQSPollingThread();
         pollingThread.start();
    }

    @Override
    public void onApplicationEvent(ContextClosedEvent contextClosedEvent)
    {
        pollingThread.stop();
    }
}

class SQSPollingThread implements Runnable() {
    private volatile boolean stopped = false;

    public void stop() {
        stopped = true;
    }

    @Override
    public void run() {
        while (!stopped) {
            message = pollSqsBlockingFor20seconds();
            process(message);
        }
    }
}

通过此设置，当 TomCat“告诉”您的服务停止时，您的服务将设置停止标志并立即返回。可能仍有待处理的轮询请求，该请求最终将在接下来的 20 秒内完成并得到处理。处理完最后一个事件后，内部 SQSPollingThread 也将结束。