Jetty - 使用 websockets 和 ByteBuffer 时可能存在内存泄漏

Question

我正在使用 Jetty 9.3.5.v20151012 向使用 websocket 的客户端传递大量事件。事件由 3 部分组成：数字、事件类型和时间戳，每个事件都被序列化为 byte[] 并使用 ByteBuffer 发送。

在一定的小时数/天数之后，根据客户端的数量，我注意到堆内存增加了，GC 没有任何可能恢复它。 当堆（设置为 512MB）快满时，jvm 使用的内存约为 700-800 MB，CPU 为 100%（看起来就像 GC 经常尝试清理）。一开始，当我启动 Jetty 时，调用 GC 时内存在 30MB 左右，但一段时间后，这个数字越来越大。最终进程被杀死。

我正在使用 jvisualvm 作为内存泄漏调试的分析器，并且我附上了一些磁头转储的屏幕截图：

以下是使用 ByteBuffer 处理消息发送的主要代码：

我基本上有一种方法可以为需要在一条消息中发送的所有事件创建一个字节[]（fullbytes）：

byte[] numberBytes = ByteBuffer.allocate(4).putFloat(number).array();
byte[] eventBytes = ByteBuffer.allocate(2).putShort(event).array();
byte[] timestampBytes  = ByteBuffer.allocate(8).putDouble(timestamp).array();

for (int i = 0; i < eventBytes.length; i++) {
    fullbytes[i + scount*eventLength] = eventBytes[i];
}

for (int i = 0; i < numberBytes.length; i++) {
    fullbytes[eventBytes.length + i + scount*eventLength] = numberBytes[i];
}

for (int i = 0; i < timestampBytes.length; i++) {
    fullbytes[numberBytes.length + eventBytes.length + i + scount*eventLength] = timestampBytes[i];
}

然后是另一种方法（在单独的线程中调用）在 websocket 上发送字节

ByteBuffer bb = ByteBuffer.wrap(fullbytes);
wsSession.getRemote().sendBytesByFuture(bb);
bb.clear();

正如我在几个地方（在文档或here 和here）上读到的，这个问题不应该出现，因为我没有使用直接的 ByteBuffer。这可能是与 Jetty / websockets 相关的错误吗？

请指教！

编辑：

我进行了更多测试，我注意到在向未连接的客户端发送消息时出现问题，但码头没有收到 onClose 事件（例如，用户将他的笔记本电脑置于待机状态）。因为 on close 事件没有被触发，所以服务器代码不会取消注册客户端并继续尝试将消息发送到该客户端。我不知道为什么，但关闭事件是在 1 或 2 小时后收到的。此外，有时（尚不知道上下文）虽然接收到事件并且客户端（套接字）未注册，但对 WebSocketSession 对象（对于该客户端）的引用仍然挂起。我还没有弄清楚为什么会发生这种情况。

在那之前，我有 2 种可能的解决方法，但我不知道如何实现它们（还有其他好的用途）：

1. 始终检测连接何时未打开（或暂时关闭，例如用户将笔记本电脑置于待机状态）。我尝试使用 sendPing() 并实现 onFrame() 但我找不到解决方案。有没有办法做到这一点？
2. 定期“刷新”缓冲区。如何丢弃未发送给客户端的消息，以免它们继续排队？

编辑 2

这可能将话题指向另一个方向，所以我发了另一个帖子here。

编辑 3

我已经对发送的大量消息/字节进行了更多测试，我发现了为什么“它接缝”了内存泄漏有时只出现：在与 sevlet 时使用的线程不同的线程上异步发送字节时.configure() 被调用，在大量积累之后，客户端断开连接后内存没有被释放。此外，我无法在使用 sendBytes(ByteBuffer) 时模拟内存泄漏，只能使用 sendBytesByFuture(ByteBuffer) 和 sendBytes(ByteBuffer, WriteCallback)。

这看起来很奇怪，但我不相信我在测试中做错了什么。

代码：

@Override
public void configure(WebSocketServletFactory factory) {
    factory.getPolicy().setIdleTimeout(1000 * 0);
    factory.setCreator(new WebSocketCreator() {

    @Override
    public Object createWebSocket(ServletUpgradeRequest req,
            ServletUpgradeResponse resp) {
        return new WSTestMemHandler();
    }
}); 
}

@WebSocket
public class WSTestMemHandler {
    private boolean connected = false;
    private int n = 0;

    public WSTestMemHandler(){
    }

    @OnWebSocketClose
    public void onClose(int statusCode, String reason) {
        connected = false;
        connections --;
        //print debug
    }

    @OnWebSocketError
    public void onError(Throwable t) {
    //print debug
    }

    @OnWebSocketConnect
    public void onConnect(final Session session) throws InterruptedException {

        connected = true;
        connections ++;
    //print debug

        //the code running in another thread will trigger memory leak 
        //when to client endpoint is down and messages are still sent 
        //because the GC will not cleanup after onclose received and
        //client disconnects

        //if the "while" loop is run in the same thread, the memory 
        //can be released when onclose is received, but that would
        //mean to hold the onConnect() method and not return. I think
        //this would be bad practice.

        new Thread(new Runnable() { 

            @Override
            public void run() {

                while (connected) {

                    testBytesSend(session);
                    try {
                        Thread.sleep(4);
                    } catch (InterruptedException e) {
                    }

                }
                //print debug
            }
        }).start();


    }



    private void testBytesSend(Session session) {

        try {
            int noEntries = 200;
            ByteBuffer bb = ByteBuffer.allocate(noEntries * 14);
            for (int i = 0; i < noEntries; i++) {
                n+= 1.0f;
                bb.putFloat(n);
                bb.putShort((short)1);
                bb.putDouble(123456789123.0);
            }
            bb.flip();


            session.getRemote().sendBytes(bb, new WriteCallback() {

                @Override
                public void writeSuccess() {

                }

                @Override
                public void writeFailed(Throwable arg0) {

                }
            });


        //print debug
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


}   

Answer 1

您使用ByteBuffer 的效率非常低。

不要仅仅为了获得一个字节数组而创建所有这些次要/微小的 ByteBuffer，然后将其丢弃。哎呀。

注意：您甚至没有正确使用 .array() 调用，因为并非所有 ByteBuffer 分配都有一个您可以这样访问的支持数组。

字节数组的numberBytes、eventBytes、timestampBytes 和fullbytes 不应存在。

创建一个ByteBuffer，代表您要发送的整个消息，将其分配为您需要的大小或更大的大小。

然后将您想要的单个字节放入其中，翻转它，并为 Jetty 实现提供单个 ByteBuffer。

Jetty 将使用标准的ByteBuffer 信息（例如position 和limit）来确定应该实际发送ByteBuffer 的哪一部分。