应该传输一个字节流，并且有一个生产者线程和一个消费者线程。生产者的速度大多数时候都高于消费者，我需要足够的缓冲数据来满足我的应用程序的 QoS。我读到了我的问题，并且有共享缓冲区、PipeStream .NET 类等解决方案……这个类将在服务器上多次实例化，所以我需要优化解决方案。使用 ByteArray 的队列是个好主意吗？
如果是，我将使用优化算法来猜测队列大小和每个 ByteArray 容量，理论上它适合我的情况。
如果没有，我最好的方法是什么？
请让我知道在 C# 或 VB 中是否有良好的无锁线程安全实现 ByteArray 队列。
提前致谢
如果不是逐个字节地生产和消费，你可能会获得更多的加速，你可以分块工作。在这种情况下，代码的“无锁”可能根本不重要——事实上，传统的锁定解决方案可能更可取。我会尝试证明。
C# 中给出了一个无锁、单一生产者、单一消费者、有界队列。（清单 A）
没有深奥的互锁操作，甚至没有显式的内存屏障。比方说，乍一看，它的速度和无锁一样快。不是吗？现在让我们将它与Marc Gravell 给出
的锁定解决方案进行比较，这里。
我们将使用在内核之间没有共享 L3 缓存的双 CPU 机器。我们预计最多 2 倍的加速。2 倍的加速确实意味着无锁解决方案在理论范围内表现理想。
为了给无锁代码创造一个理想的环境，我们甚至会使用这里的实用程序类来设置生产者和消费者线程的 CPU 亲和性。
测试的结果代码在（清单 B）中。
它正在生产约。一个线程上 10MBytes，而在另一个线程上消耗它。
队列大小固定为 32KBytes。如果它已满，则生产者等待。
在我的机器上运行的典型测试如下所示：
LockFreeByteQueue：799
毫秒字节队列：1843 毫秒
无锁队列更快。哇，它的速度是原来的 2 倍以上！这是值得吹嘘的。:)
让我们看看发生了什么。Marc 的锁定队列就是这样做的。它锁定。它对每个字节都这样做。
我们真的需要锁定每个字节并逐字节推送数据吗？它肯定会以块的形式到达网络上（例如一些大约 1k 的数据包）。即使它真的从内部源逐字节到达，生产者也可以轻松地将其打包成漂亮的块。
让我们这样做——而不是逐个字节地产生和消费，让我们分块工作并将另外两个测试添加到微基准测试（清单 C，只需将其插入到基准测试体中）。
现在典型的运行如下所示：
LockFreePageQueue：33ms
PageQueue：25ms
现在，它们实际上都比原来的无锁代码快了 20 倍——Marc的添加分块的解决方案现在实际上比带有分块的无锁代码快！
我们没有采用会导致 2 倍加速的无锁结构，而是尝试了另一种解决方案，它可以很好地处理锁定并导致 20 倍（！）加速。
许多问题的关键不是避免锁定，而是避免共享和最小化锁定。在上述情况下，我们可以避免在字节复制期间共享。
我们可以在大部分时间处理私有结构，然后将单个指针排入队列，从而将共享空间和时间缩减为将单个指针插入队列中。
清单 A，一个无锁、单生产者、单消费者队列：
public class BoundedSingleProducerSingleConsumerQueue<T>
{
T[] queue;
volatile int tail;
volatile int head;
public BoundedSingleProducerSingleConsumerQueue(int capacity)
{
queue = new T[capacity + 1];
tail = head = 0;
}
public bool TryEnqueue(T item)
{
int newtail = (tail + 1) % queue.Length;
if (newtail == head) return false;
queue[tail] = item;
tail = newtail;
return true;
}
public bool TryDequeue(out T item)
{
item = default(T);
if (head == tail) return false;
item = queue[head];
queue[head] = default(T);
head = (head + 1) % queue.Length;
return true;
}
}
清单 B，一个微基准：
class Program
{
static void Main(string[] args)
{
for (int numtrials = 3; numtrials > 0; --numtrials)
{
using (ProcessorAffinity.BeginAffinity(0))
{
int pagesize = 1024 * 10;
int numpages = 1024;
int totalbytes = pagesize * numpages;
BoundedSingleProducerSingleConsumerQueue<byte> lockFreeByteQueue = new BoundedSingleProducerSingleConsumerQueue<byte>(1024 * 32);
Stopwatch sw = new Stopwatch();
sw.Start();
ThreadPool.QueueUserWorkItem(delegate(object state)
{
using (ProcessorAffinity.BeginAffinity(1))
{
for (int i = 0; i < totalbytes; i++)
{
while (!lockFreeByteQueue.TryEnqueue((byte)(i & 0xFF))) ;
}
}
});
for (int i = 0; i < totalbytes; i++)
{
byte tmp;
while (!lockFreeByteQueue.TryDequeue(out tmp)) ;
}
sw.Stop();
Console.WriteLine("LockFreeByteQueue: {0}ms", sw.ElapsedMilliseconds);
SizeQueue<byte> byteQueue = new SizeQueue<byte>(1024 * 32);
sw.Reset();
sw.Start();
ThreadPool.QueueUserWorkItem(delegate(object state)
{
using (ProcessorAffinity.BeginAffinity(1))
{
for (int i = 0; i < totalbytes; i++)
{
byteQueue.Enqueue((byte)(i & 0xFF));
}
}
});
for (int i = 0; i < totalbytes; i++)
{
byte tmp = byteQueue.Dequeue();
}
sw.Stop();
Console.WriteLine("ByteQueue: {0}ms", sw.ElapsedMilliseconds);
Console.ReadKey();
}
}
}
}
清单 C，分块测试：
BoundedSingleProducerSingleConsumerQueue<byte[]> lockfreePageQueue = new BoundedSingleProducerSingleConsumerQueue<byte[]>(32);
sw.Reset();
sw.Start();
ThreadPool.QueueUserWorkItem(delegate(object state)
{
using (ProcessorAffinity.BeginAffinity(1))
{
for (int i = 0; i < numpages; i++)
{
byte[] page = new byte[pagesize];
for (int j = 0; j < pagesize; j++)
{
page[j] = (byte)(i & 0xFF);
}
while (!lockfreePageQueue.TryEnqueue(page)) ;
}
}
});
for (int i = 0; i < numpages; i++)
{
byte[] page;
while (!lockfreePageQueue.TryDequeue(out page)) ;
for (int j = 0; j < pagesize; j++)
{
byte tmp = page[j];
}
}
sw.Stop();
Console.WriteLine("LockFreePageQueue: {0}ms", sw.ElapsedMilliseconds);
SizeQueue<byte[]> pageQueue = new SizeQueue<byte[]>(32);
ThreadPool.QueueUserWorkItem(delegate(object state)
{
using (ProcessorAffinity.BeginAffinity(1))
{
for (int i = 0; i < numpages; i++)
{
byte[] page = new byte[pagesize];
for (int j = 0; j < pagesize; j++)
{
page[j] = (byte)(i & 0xFF);
}
pageQueue.Enqueue(page);
}
}
});
sw.Reset();
sw.Start();
for (int i = 0; i < numpages; i++)
{
byte[] page = pageQueue.Dequeue();
for (int j = 0; j < pagesize; j++)
{
byte tmp = page[j];
}
}
sw.Stop();
Console.WriteLine("PageQueue: {0}ms", sw.ElapsedMilliseconds);
在 .NET 4 中System.Collections.Concurrent.Queue<T>，这些东西可以是无锁的（虽然仍然是通用的）。
Dobbs 博士在 C++ 中实现了一个无锁队列，您可以相对轻松地将其应用到 C# 中。当只有一个生产者（可以有任意数量的消费者）时，它可以工作。
基本思想是使用双向链表作为底层结构以及可移动的头部和尾部引用。当一个项目被生成时，它被添加到末尾，并且从列表的开头到当前“头”之间的所有内容都被删除。吃东西，试着抬起头；如果它碰到尾部，则失败，如果没有，则成功并返回新元素。特定的操作顺序使其本质上是线程安全的。
但是，在这里使用这种“无锁”设计有两个主要问题：
没有办法强制队列大小的上限，如果你的生产者比你的消费者快，这可能是一个严重的问题；
按照设计，Consume如果没有产生任何内容，该方法必须根本无法检索元素。这意味着您需要为消费者实现自己的锁定，并且这种锁定总是忙等待（这比锁定性能范围内的锁定要糟糕得多）或定时等待（这会进一步减慢消费者的速度）。
由于这些原因，我建议您认真考虑是否真的需要无锁结构。很多人来到这个站点时认为它会比使用锁定的等效结构“更快”，但对于大多数应用程序而言，实际差异是如此微不足道，以至于通常不值得增加复杂性，在某些情况下它实际上可以性能更差，因为等待状态（或警报等待）比忙等待便宜得多。
多核机器和内存屏障的需求使得有效的无锁线程更加复杂；在正常操作下，您仍然可以无序执行，并且在 .NET 中，抖动可以进一步决定重新排序指令，因此您可能需要在代码中添加volatile变量和Thread.MemoryBarrier调用，这可能再次有助于锁定- 免费版本比基本同步版本更昂贵。
首先使用普通的旧同步生产者-消费者队列，然后分析您的应用程序以确定它是否可以满足您的性能要求，怎么样？在Joseph Albahari 的网站上有一个很棒、高效的 PC 队列实现。或者，正如 Richard 所提到的，如果您使用的是 .NET 4.0 框架，那么您可以简单地使用ConcurrentQueue或更可能使用BlockingCollection。
先测试 - 负载测试同步队列，这很容易实现 - 并观察实际花费了多少时间锁定。不是等待，无论如何您都必须这样做，而是在发出信号后实际获取和释放锁。如果它超过你程序执行时间的 1%，我会非常惊讶；但如果是这样，那么开始研究无锁实现——并确保你也对它们进行分析，以确保它们实际上表现得更好。
节流在这里很重要，听起来，这篇杂志文章中的 BoundedBuffer 类符合要求。.NET 4.0 中将提供一个与BlockingCollection 类类似的类。调整缓冲区大小仍然取决于您。
Julian M Bucknall用 C# 编写了一个。
最重要的部分是共享对象的设计。在我的场景中，读取器和写入器可以独立使用单独的缓冲区（大数据块），然后，只有访问像队列这样的共享 FIFO 对象才应该同步。这样可以最大限度地减少锁定时间，并且线程可以并行完成工作。使用.NET framewok 4.0实现这一概念变得容易：
System.Collections.Concurrent 命名空间中有一个 ConcurrentQueue(Of T) 类，arrayByte 是我的场景中用作队列类型的好类型。命名空间中还有其他线程安全的集合。
http://msdn.microsoft.com/en-us/library/system.collections.concurrent.aspx