使用 LinkedBlockingQueue 时的 Java 内存泄漏答案

【问题标题】：Java Memory Leak in using LinkedBlockingQueue使用 LinkedBlockingQueue 时的 Java 内存泄漏
【发布时间】：2016-02-20 06:37:52
【问题描述】：

亲爱的，

我的java应用程序运行一段时间后抛出异常

线程“Thread-6”java.lang.OutOfMemoryError 中的异常：超出 GC 开销限制

使用 jprofiler 监控 jvm 内存后，我发现我的代码中存在内存泄漏。

在我的程序中，我向某些设备发送了很多 snmp 请求并获得它们的响应并将它们推送到两个 LinkedBlockingQueue 对象中。
有一些线程从队列中轮询并处理它们并从队列中删除对象。但似乎这些对象并没有从队列中删除。我在我的代码中看不到任何问题。任何意见和建议将不胜感激

ResponseListener.java
public class ResponseListener{
    public void receiveResponse(){
                SNMPResultMessage resultMessage;
                    resultMessage = new SNMPResultMessage(
                            userObject.getDeviceId(),
                            oid,
                            deviceServiceId,
                            sensorId,
                            response.getVariable(oid).toInt(),
                            userObject.getTimestamp(),
                            Utils.getCreateDay(dt),
                            Utils.getQuarterOfDay(dt),
                            slaDisabled
                    );
                    Logger.debug("SNMP Request Response:" + resultMessage);                    

                PubSub.publish(resultMessage);

     }
}

PubSub.java

公共类 PubSub {

public static final BlockingQueue<ResultMessage> average = new LinkedBlockingQueue<>();
public static final BlockingQueue<ResultMessage> cassandra = new LinkedBlockingQueue<>();
public static void publish(ResultMessage message) {
    average.add(message);
    cassandra.add(message);
}



private PubSub() {
    /* prevent instantiation */
}

}

AverageWatcher.java

public class AverageWatcher implements Runnable{
    @Override
    public void run() {
        try {
            while (true) {
                ResultMessage message = PubSub.average.take();
                if (message == null) {
                    if (numberOfChanges > 1000) {
                        insertBatch();
                    }
                    continue;
                }
                numberOfChanges++;
                if (changes.get(message.getSensorId()) == null) {
                    changes.put(message.getSensorId(), new HashSet<>());
                }
                changes.get(message.getSensorId()).add(new CreateDayQuarterOfDay(message.getCreateDay(), message.getQuarterOfDay()));
                if (numberOfChanges > 5000) {
                    insertBatch();
                }
            }
        } catch (Exception ex) {
            Logger.error(ex.getCause());
        }
    }
}

CassandraQueueWatcher.java

public class CassandraQueueWatcher implements Runnable{
    @Override
    public void run() {
        try {
            while (true) {
                ResultMessage message = PubSub.cassandra.take();
                if (message == null) {
                    if (batchSize > 1000) {
                        insertBatch();
                    }
                    continue;
                }

                boolean violated = checkViolated(message);

                batch.add(
                        Cassandra.insertRawReportStmt
                                .bind()
                                .setInt("sensor_id", message.getSensorId())
                                .setInt("create_day", message.getCreateDay())
                                .setLong("create_time", message.getTimestamp() / 1000)
                                .setFloat("value", message.getValue())
                                .setBool("violated",violated)
                );
                batchSize++;
                if (batchSize > 2000) {
                    insertBatch();
                }
                message =  null;
            }
        } catch (Exception ex) {
            Logger.error(ex.getCause());
        }
    }

}

在我的代码中，仅在 ResponseListener.java 中创建新的 SNMPResultMessage 对象。请看一下附图。 jprofiler

【问题讨论】：

insertBatch 是做什么的？
您使用的是什么版本的 Java？一些早期版本没有正确解除队列头部的链接，导致节点无法被清理。 bugs.java.com/view_bug.do?bug_id=6805775
insert batch in cassandraWatcher 将数据插入 db 并在 averageWatcher 中将数据插入 mysql java 版本“1.8.0_60”Java(TM) SE Runtime Environment (build 1.8.0_60-b27) 我正在运行它在 docker 容器上
您确定它来自队列吗？您应该进行堆转储并找出所有数据所在的位置。错误不太可能出现在队列中，而不是您的实现中。
@jo 你看到截图了吗？ jprofiler screenshoot，为什么要生成 LinkedBlockingQueue 实例？我只有一个 LinkedBlockingQueue 类型的对象

标签： java multithreading memory-leaks

【解决方案1】：

我猜想队列正在填满，因为您插入的新消息比阅读它们的速度更快。

尝试检查队列的大小或使用bounded version。

【讨论】：