4

我正在使用Netty执行大文件上传。它工作正常，但客户端使用的 RAM 似乎随着文件的大小而增加。这不是预期的行为，因为一切都是从读取源文件到写入目标文件的管道。

起初，我想到了一种自适应缓冲区，直到达到 Xmx，但将 Xmx 设置为合理的值（50M）会在开始上传后很快导致 OutOfMemoryError 。在使用 Eclipse Memory Analyzer 进行一些研究后，似乎保留堆内存的对象是：

org.jboss.netty.channel.socket.nio.NioSocketChannel$WriteRequestQueue

是否有任何选项可以设置此队列的限制，或者我是否必须使用 ChannelFutures 编写自己的队列来控制字节数并在达到限制时阻塞管道？

谢谢你的帮助，

问候，

雷诺

4

2 回答 2

2

Netty Github 上@normanmaurer 的回答

你应该使用

Channel.isWritable()

检查“队列”是否已满。如果是这样，您将需要检查是否有足够的空间来写更多。因此，如果您将数据快速写入以将其发送给客户端，则可能会发生您看到的效果。尝试通过 DefaultFileRegion 或 ChunkedFile 写入文件时，您可以解决此类问题。

@normanmaurer 谢谢我错过了频道的这种方法！我想我需要阅读里面发生的事情：

org.jboss.netty.handler.stream.ChunkedWriteHandler

更新：2012/08/30 这是我为解决我的问题而编写的代码：

public class LimitedChannelSpeaker{
    Channel channel;
    final Object lock = new Object();
    long maxMemorySizeB;
    long size = 0;
    Map<ChannelBufferRef, Integer> buffer2readablebytes = new HashMap<ChannelBufferRef, Integer>();

    public LimitedChannelSpeaker(Channel channel, long maxMemorySizeB) {
        this.channel= channel;
        this.maxMemorySizeB = maxMemorySizeB;
    }

    public ChannelFuture speak(ChannelBuffer buff) {
        if (buff.readableBytes() > maxMemorySizeB) {
            throw new IndexOutOfBoundsException("The buffer is larger than the maximum allowed size of " + maxMemorySizeB + "B.");
        }
        synchronized (lock) {
            while (size + buff.readableBytes() > maxMemorySizeB) {
                try {
                    lock.wait();
                } catch (InterruptedException ex) {
                    throw new RuntimeException(ex);
                }
            }
            ChannelBufferRef ref = new ChannelBufferRef(buff);
            ref.register();
            ChannelFuture future = channel.write(buff);
            future.addListener(new ChannelBufferRef(buff));
            return future;
        }
    }

    private void spoken(ChannelBufferRef ref) {
        synchronized (lock) {
            ref.unregister();
            lock.notifyAll();
        }
    }

    private class ChannelBufferRef implements ChannelFutureListener {

        int readableBytes;

        public ChannelBufferRef(ChannelBuffer buff) {
            readableBytes = buff.readableBytes();
        }

        public void unregister() {
            buffer2readablebytes.remove(this);
            size -= readableBytes;
        }

        public void register() {
            buffer2readablebytes.put(this, readableBytes);
            size += readableBytes;
        }

        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            spoken(this);
        }
    }
}
于 2012-08-29T13:11:59.957 回答
0

对于桌面后台应用程序

Netty 专为高度可扩展的服务器而设计，例如大约 10,000 个连接。对于少于几百个连接的桌面应用程序，我会使用普通 IO。您可能会发现代码要简单得多，并且应该使用不到 1 MB。

于 2012-08-29T09:08:07.950 回答