java OutOfMemoryError:无法使用ExecutorService创建新的本机线程

Question

我在一夜之间启动了我的实例，看看它是如何处理事情的。当我今天早上来的时候，我面临着一个巨大的挑战

Exception in thread "pool-535-thread-7" java.lang.OutOfMemoryError: unable to create new native thread
    at java.lang.Thread.start0(Native Method)
    at java.lang.Thread.start(Thread.java:691)
    at java.util.concurrent.ThreadPoolExecutor.addWorker(ThreadPoolExecutor.java:943)
    at java.util.concurrent.ThreadPoolExecutor.processWorkerExit(ThreadPoolExecutor.java:992)[info] application - Connecting to server A
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:603)
    at java.lang.Thread.run(Thread.java:722)

我的代码的目的很简单：每5分钟，我连接一个远程服务器列表，发送一个请求（通过socket），就这样

这是我的密码：

我的“cron”任务：

/** will create a new instance of ExecutorService every 5 minutes, loading all the websites in the database to check their status **/
/** Maybe that's where the problem is ? I need to empty (GC ?) this ExecutorService ? **/
Akka.system().scheduler().schedule(
    Duration.create(0, TimeUnit.MILLISECONDS), // Initial delay 0 milliseconds
    Duration.create(5, TimeUnit.MINUTES),     // Frequency 5 minutes
    new Runnable() {
        public void run() {
            // We get the list of websites to check
            Query<Website> query = Ebean.createQuery(Website.class, "WHERE disabled = false AND removed IS NULL");
            query.order("created ASC");
            List<Website> websites = query.findList(); // Can be 1, 10, 100, 1000. In my test case, I had only 9 websites.

            ExecutorService executor = Executors.newFixedThreadPool(NTHREDS);
            for (Website website : websites) {
                CheckWebsite task = new CheckWebsite(website);
                executor.execute(task);
            }

            // This will make the executor accept no new threads
            // and finish all existing threads in the queue
            executor.shutdown();
        }
    },
    Akka.system().dispatcher()
);

我的支票网站课程：

public class CheckWebsite implements Runnable {
    private Website website;

    public CheckWebsite(Website website) {
        this.website = website;
    }

    @Override
    public void run() {
        WebsiteLog log = website.checkState(); // This is where the request is made, I copy paste the code just after
        if (log == null) {
            Logger.error("OHOH, WebsiteLog should not be null for website.checkState() in CheckWebsite class :s");
            return;
        }

        try {
            log.save();
       catch (Exception e) {
            Logger.info ("An error occured :/");
            Logger.info(e.getMessage());
            e.printStackTrace();
        }
    }
}

我在Website.class中的checkState()方法：

public WebsiteLog checkState() {
    // Since I use Socket and the connection can hang indefinitely, I use an other ExecutorService in order to limit the time spent
    // The duration is defined via Connector.timeout, Which will be the next code.

    ExecutorService executor = Executors.newFixedThreadPool(1);

    Connector connector = new Connector(this);
    try {
        final long startTime = System.nanoTime();

        Future<String> future = executor.submit(connector);
        String response = future.get(Connector.timeout, TimeUnit.MILLISECONDS);

        long duration = System.nanoTime() - startTime;

        return PlatformLog.getLastOccurence(this, response, ((int) duration/ 1000000));
    }
    catch (Exception e) {
        return PlatformLog.getLastOccurence(this, null, null);
    }
}

这是Connector.class。我在这里删除了无用的部分（如捕捉）：

public class Connector implements Callable<String> {
    public final static int timeout = 2500; // WE use a timeout of 2.5s, which should be enough

    private Website website;

    public Connector(Website website) {
        this.website = website;
    }

    @Override
    public String call() throws Exception {
        Logger.info ("Connecting to " + website.getAddress() + ":" + website.getPort());
        Socket socket = new Socket();

        try {
            socket.connect(new InetSocketAddress(website.getIp(), website.getPort()), (timeout - 50));
            BufferedReader input = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            String response = input.readLine();
            socket.close();

            return response;
        }
        catch (Exception e) {
            e.printStackTrace();
            throw e;
        }
        finally {
            // I take the precaution to close the socket here in order to avoid a memory leak
            // But if the previous ExecutorService force the close of this thread before
            // I can't guarantee it will be closed :/
            if (socket != null && !socket.isClosed()) {
                socket.close();
            }
        }
    }
}

我是Java多线程新手，所以我可能犯了一个很大的错误。我怀疑某些方面可能是原因，但我缺乏知识，因此需要寻求您的帮助：）

作为总结，以下是潜在领域：

1. 每5分钟创建一个新的ExecutorService。也许我可以重复使用旧的？或者我是否需要在完成时关闭当前一个（如果需要，如何关闭？）
2. 事实上，我创建了一个ExecutorService，它将创建一个ExecutorService（在checkstate()方法中）
3. 如果运行连接器类的ExecutorService花费的时间太长，导致socket未关闭（然后导致内存泄漏），那么连接器类可能会（猛烈地）停止

另外，正如您所看到的，线程“pool-535-thread-7”发生了异常，这意味着它不会很快发生

我将上次发生的检查存储在数据库中，创建日志条目（在{}）时，增量大约为5小时（因此，每5分钟，线程在大约60次调用后崩溃一次）

更新：以下是重新访问的checkState方法，其中包括关闭调用：

public PlatformLog checkState() {
    ExecutorService executor = Executors.newFixedThreadPool(1);

    Connector connector = new Connector(this);
    String response = null;
    Long duration = null;

    try {
        final long startTime = System.nanoTime();

        Future<String> future = executor.submit(connector);
        response = future.get(Connector.timeout, TimeUnit.MILLISECONDS);

        duration = System.nanoTime() - startTime;
    }
    catch (Exception e) {}

    executor.shutdown();
    if (duration != null) {
        return WebsiteLog.getLastOccurence(this, response, (duration.intValue()/ 1000000));
    }
    else {
        return WebsiteLog.getLastOccurence(this, response, null);
    }
}

Answer 1

我不确定这是唯一的问题，但是您正在checkState()方法中创建一个ExecutorService，但是您没有关闭它

根据^{}的JavaDocs：

The threads in the pool will exist until it is explicitly shutdown.

线程保持活动状态将导致ExecutorService{a2}（它将代表您调用shutdown()）。因此，每次调用它时，您都会泄漏一个线程