.net实时流处理-需要巨大而快速的RAM缓冲器

Question

我正在开发的应用程序与一个数字音频设备进行通信，它能够同时发送24个不同的语音流。该设备通过USB接口连接，使用FTDI设备(串口仿真器)和D2XX驱动程序(基本的COM驱动程序来慢处理4.5Mbit的传输)。

基本上，应用程序由3个线程组成：

• 主线程-图形用户界面、控件等.
• 总线读取器--在此线程中，数据不断从设备读取并保存到文件缓冲区(此线程中没有逻辑)
• 数据解释器--该线程从文件缓冲区读取数据，转换为示例，进行简单的样本处理，并将样本保存到单独的wav文件中。

我使用文件缓冲区的原因是我希望确保不会丢失任何示例。应用程序并不总是使用记录，所以我选择了这个解决方案，因为它是安全的。

应用程序工作正常，只是缓冲的波形文件生成器非常慢。对于24条1分钟的平行记录，大约需要4分钟才能完成记录。我非常肯定，在这个过程中消除硬盘的使用会大大提高速度。

第二个问题是，文件缓冲区对于长记录来说真的很重，在数据处理结束之前，我无法清除这个缓冲区(这会进一步减慢处理速度)。

对于RAM缓冲区，我至少需要1GB才能使它正常工作。

在.NET中分配这么大的内存的最好方法是什么？我将在两个线程中使用这个内存，因此需要一个快速同步机制。我在考虑一个循环缓冲区:一个大数组，总线阅读器保存数据，数据解释器读取它。你觉得那个怎么样？

现在编辑缓冲，我使用的是基于文件的类BinaryReader和BinaryWriter。

Answer 1

您应该能够将一个包装器类放在一起，该类管理单个byte[]缓冲区，并使用固定大小的内存流为数据范围打包临时锁。本质上，您只需要定义一次缓冲区，只要您想使用它，就需要从ConcurrentBuffer类中请求一个锁对象。该锁包含一个MemoryStream，它保证在请求锁之前对您指定的索引范围具有独占访问权限。

我总结了一个简单的示例，它应该为您提供一个良好的起点：

public class ConcurrentBuffer
{
    private readonly byte[] buffer;

    internal byte[] GetBuffer() { return buffer; }

    private List<BufferLock> locks = new List<BufferLock>();

    public ConcurrentBuffer(int bufferSize)
    {
        buffer = new byte[bufferSize];
    }

    public BufferLock AcquireLock(int startIndex, int endIndex)
    {
        if (startIndex < 0) throw new ArgumentOutOfRangeException("startIndex");
        if (startIndex > endIndex || endIndex >= buffer.Length) throw new ArgumentOutOfRangeException("endIndex");
        lock (buffer)
        {
            foreach (var l in locks)
            {
                if (!(endIndex < l.StartIndex || startIndex > l.EndIndex))
                {
                    return null;
                }
            }
            var bl = new BufferLock(startIndex, endIndex, this);
            locks.Add(bl);
            return bl;
        }
    }

    public void ReleaseLock(BufferLock lck)
    {
        lock (buffer)
        {
            locks.Remove(lck);
        }
    }

    public class BufferLock
    {
        public int StartIndex { get; private set; }
        public int EndIndex { get; private set; }
        public ConcurrentBuffer TargetBuffer { get; private set; }
        public MemoryStream Stream { get; private set; }

        internal BufferLock(int startIndex, int endIndex, ConcurrentBuffer buffer)
        {
            StartIndex = startIndex;
            EndIndex = endIndex;
            TargetBuffer = buffer;
            Stream = new MemoryStream(buffer.GetBuffer(), startIndex, endIndex - startIndex, true);
        }

        public void Release()
        {
            Stream.Dispose();
            TargetBuffer.ReleaseLock(this);
        }
    }
}

class Program
{
    static void Main(string[] args)
    {
        ConcurrentBuffer cb = new ConcurrentBuffer(32000);

        byte[] myData = { 32, 13, 53, 29, 50 };

        // l1 will acquire a lock
        var l1 = cb.AcquireLock(0, 50); 
        if (l1 != null) l1.Stream.Write(myData, 0, myData.Length);

        // l2 will fail because l1 has part of its range locked
        var l2 = cb.AcquireLock(30, 70);
        if (l2 != null) l2.Stream.Write(myData, 0, myData.Length);

        l1.Release();

        // l3 will succeed at locking because l1 has been released
        var l3 = cb.AcquireLock(40, 5000);
        if (l3 != null)
        {
            while (l3.Stream.Position + myData.Length <= l3.Stream.Length)
            {
                l3.Stream.Write(myData, 0, myData.Length);
            }
        }

        l3.Release();

        Console.ReadLine();
    }
}