memory-leaks - 使用带有 Windows 换行符 (CR LF) 的行时，Camel netty TCP 客户端中的内存泄漏

Question

我的使用文本行的 Camel netty tcp 客户端似乎存在内存泄漏，但前提是测试数据行以 Windows (CR LF) 换行符结尾。我没有遇到 Unix (LF) 换行符的问题。

我做了一个简短的测试来演示模拟 tcp 服务器连续发送测试数据线的问题。

在测试数据中使用 Unix (LF) 换行符时，我看到大约 3.500 条消息/秒的吞吐量和稳定的 180 MB ram 使用。没有问题。

在测试数据中使用 Windows (CR LF) 换行符时，我看到吞吐量从 380.000(哇!)消息/秒开始，直到在大约 30 秒后达到我的 -Xmx4G 堆限制，并且可能由于过度的 GC 而大大减慢；如果给定更多堆，它会稳步增长，直到达到该限制(尝试使用 -Xmx20G)。

唯一的区别是我的测试数据中的换行符......
我在这里错过了什么吗？

在带有 OpenJDK 1.8.0_192 的 Linux 上使用 Camel 2.24.0(使用 netty 4.1.32-Final)。最新的 netty 4.1.36.Final 也会出现此问题。 OpenJ9 JVM 也会发生，因此似乎不是特定于 JVM 的。

public abstract class MyRouteBuilderTestBase extends CamelTestSupport {
    private final int nettyPort = AvailablePortFinder.getNextAvailable();

    private ServerSocket serverSocket;
    private Socket clientSocket;
    private PrintWriter out;

    @Override
    protected RouteBuilder createRouteBuilder() {
        return new RouteBuilder() {
            public void configure() {
                from("netty4:tcp://localhost:" + nettyPort + "?clientMode=true&textline=true&sync=false")
                    .to("log:throughput?level=INFO&groupInterval=10000&groupActiveOnly=false");
            }
        };
    }

    protected void startServerStub(String testdata) throws Exception {
        serverSocket = new ServerSocket(nettyPort);
        clientSocket = serverSocket.accept();
        out = new PrintWriter(clientSocket.getOutputStream(), true);
        for (;;) {
            out.print(testdata);
        }
    }

    @After
    public void after() throws Exception {
        if (out != null) out.close();
        if (clientSocket != null) clientSocket.close();
        if (serverSocket != null) serverSocket.close();
    }
}

public class MyRouteBuilderTestUnixLineBreaks extends MyRouteBuilderTestBase {
    @Test
    public void testUnixLineBreaks() throws Exception {
        startServerStub("my test data\n");  // Unix LF
    }
}

public class MyRouteBuilderTestWindowsLineBreaks extends MyRouteBuilderTestBase {
    @Test
    public void testWindowsLineBreaks() throws Exception {
        startServerStub("my test data\r\n");  // Windows CR LF
    }
}

Answer 1

堆转储分析显示内存由 io.netty.util.concurrent.DefaultEventExecutor 的一个实例分配，该实例在内部使用具有无限大小的 LinkedBlockingQueue。此队列在导致问题的负载下无限增长。

由于参数 ，Camel 创建了 DefaultEventExecutor usingExecutorService 默认情况下这是正确的(可能不是一个好的选择)。设置 usingExecutorService=false 使 Netty 使用其事件循环而不是效果更好的执行器。

我现在得到 每秒 600.000 条消息吞吐量 使用 Windows 换行符 (CR NL) 的数据，稳定的 ram 使用量约为 200mb (-Xmx500M)。好的。

尽管使用 Unix 换行符 (NL) 的数据吞吐量仅为每秒 6.500 条消息，慢了两个数量级，这仍然令人费解。

原因是 Camel 通过继承 Netty 的 io.netty.handler.codec.DelimiterBasedFrameDecoder 创建了自己的 org.apache.camel.component.netty4.codec.DelimiterBasedFrameDecoder 类——我不知道为什么，因为 Camel 的类没有添加任何功能.但是通过子类化，Camel 阻止了 Netty 的 DelimiterBasedFrameDecoder 内部的某种优化，它在内部切换到 io.netty.handler.codec.LineBasedFrameDecoder，但前提是没有子类化。

为了克服这个问题，除了设置 usingExecutorService=false 之外，我还需要使用 Netty 的类显式声明解码器和编码器。

现在，我也使用 Unix 换行符 (NL) 获得了每秒 600.000 条消息的吞吐量，并且看到稳定的 ram 使用量约为 200mb。那看起来好多了。

public abstract class MyRouteBuilderTestBase extends CamelTestSupport {
    private final int nettyPort = AvailablePortFinder.getNextAvailable();

    private ServerSocket serverSocket;
    private Socket clientSocket;
    private PrintWriter out;

    @Override
    protected JndiRegistry createRegistry() throws Exception {
        JndiRegistry registry = super.createRegistry();

        List<ChannelHandler> decoders = new ArrayList<>();
        DefaultChannelHandlerFactory decoderTextLine = new DefaultChannelHandlerFactory() {
            @Override
            public ChannelHandler newChannelHandler() {
                return new io.netty.handler.codec.DelimiterBasedFrameDecoder(1024, true, Delimiters.lineDelimiter());
                // Works too:
                // return new LineBasedFrameDecoder(1024, true, true);
            }
        };
        decoders.add(decoderTextLine);
        ShareableChannelHandlerFactory decoderStr = new ShareableChannelHandlerFactory(new StringDecoder(CharsetUtil.US_ASCII));
        decoders.add(decoderStr);
        registry.bind("decoders", decoders);

        List<ChannelHandler> encoders = new ArrayList<>();
        ShareableChannelHandlerFactory encoderStr = new ShareableChannelHandlerFactory(new StringEncoder(CharsetUtil.US_ASCII));
        encoders.add(encoderStr);
        registry.bind("encoders", encoders);

        return registry;
    }

    @Override
    protected RouteBuilder createRouteBuilder() {
        return new RouteBuilder() {
            public void configure() {
                from("netty4:tcp://localhost:" + nettyPort + "?clientMode=true&textline=true&sync=false&usingExecutorService=false&encoders=#encoders&decoders=#decoders")
                .to("log:throughput?level=INFO&groupInterval=10000&groupActiveOnly=false");
            }
        };
    }

    protected void startServerStub(String testdata) throws Exception {
        serverSocket = new ServerSocket(nettyPort);
        clientSocket = serverSocket.accept();
        out = new PrintWriter(clientSocket.getOutputStream(), true);
        for (;;) {
            out.print(testdata);
        }
    }

    @After
    public void after() throws Exception {
        if (out != null) out.close();
        if (clientSocket != null) clientSocket.close();
        if (serverSocket != null) serverSocket.close();
    }
}

更新 :内存使用问题不是内存泄漏(我很遗憾这样表达我的问题)，而是关于缓冲。请引用用户 Bedla 和 Claus Ibsen 对此答案的评论，以更好地了解上述解决方案的后果。另请咨询CAMEL-13527