java - Jetty - 使用 websockets 和 ByteBuffer 时可能发生内存泄漏

Question

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

在一定的小时数/天数后，根据客户端的数量，我注意到堆内存增加并且 GC 无法恢复它。 当堆(设置为 512MB)几乎满时，jvm 使用的内存约为 700-800 MB，CPU 为 100%(看起来 GC 经常尝试清理)。一开始，当我启动 Jetty 时，调用 GC 时内存大约为 30MB，但一段时间后，这个数字越来越大。最终进程被杀死。

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

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

我基本上有一个方法可以为需要在一条消息中发送的所有事件创建一个 byte[](全字节):

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];
}

然后是另一个方法(在单独的线程中调用)在 websockets 上发送字节

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

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

请指教!

编辑:

我做了一些更多的测试，我注意到当向未连接的客户端发送消息时出现问题，但是 jetty 没有收到 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使用效率极低。

不要创建所有那些次要/微小的 ByteBuffers 只是为了获得一个字节数组，然后将其丢弃。恶心。

Note: you don't even use the .array() call correctly, as not all ByteBuffer allocations have a backing array you can access like that.

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

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

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

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