我最近對我的框架進行了基準測驗，發現它分配了大量垃圾。
我正在使用 a Channel<T>and the TryReadorReadAsync操作在每次呼叫時分配記憶體。所以我將它與 a 交換，BlockingCollection<T>它也在TryTake.
我使用了一個無限通道和一個作者/讀者。和一個正常的BlockingCollection<T>。
// Each thread runs this, jobmeta is justa struct
while (!token.IsCancellationRequested)
{
var jobMeta = await Reader.ReadAsync(token); // <- allocs here
jobMeta.Job.Execute();
jobMeta.JobHandle.Notify();
}
探查器告訴我所有分配都是由該ChannelReader.ReadAsync方法引起的。不幸的是我不能展示完整的代碼，但是因為我在熱路徑中使用它們，所以我需要不惜一切代價避免分配。
是否有任何替代方案在讀/寫/獲取期間不分配記憶體并且行為相同（生產者/消費者多執行緒的并發類）？我怎么能自己實作一個？
uj5u.com熱心網友回復：
System.Threading.Channels庫目前具有三個內置Channel<T>實作：
UnboundedChannel<T>SingleConsumerUnboundedChannel<T>BoundedChannel<T>
從這些實作中，分配越少的是BoundedChannel<T>. 如果你不想要邊界，你可以配置它capacity: Int32.MaxValue。內部UnboundedChannel<T>基于 a ConcurrentQueue<T>，這是一個非常高效且無爭議的集合（它是無鎖的）。分配是無鎖的必要折衷。TheBoundedChannel<T>基于內部Deque<T>集合，與locks 同步。它僅在必須擴展其支持陣列的容量時才分配記憶體，這在通道的生命周期中只會發生幾次。
BlockingCollection<T>default也是基于a的ConcurrentQueue<T>，所以也有相同的優缺點。如果您想減少分配（同時降低性能并增加爭用），您可以IProducerConsumerCollection<T>基于 synchronized實作 an Queue<T>，并將其作為引數傳遞給BlockingCollection<T>建構式。您可以使用此答案作為起點。
CancellationToken無論如何，最終將s 傳遞給這些 API 中的任何一個都將導致分配。CancellationToken必須注冊一個回呼才能立即生效，沒有分配的回呼是不可能的。我的建議是擺脫CancellationToken，并找到其他一些優雅完成的方式。喜歡使用ChannelWriter<T>.Complete或BlockingCollection<T>.CompleteAdding方法。
uj5u.com熱心網友回復：
您不需要替代型別，而是使用通道的替代方法。不是取消讀者，而是關閉作者。這是 .NET RabbitMQ 客戶端中使用的技術。
此外，確保有效載荷不需要大量分配更為重要。如果您使用 ArrayPool 來減少分配，您不能只是中止作業方法，您需要檢索訊息并釋放它們。
RabbitMQ 中的通道
RabbitMQ 在最熱的路徑中使用通道——將訊息幀寫入網路流時以及接收位元組時。很少有庫比 RabbitMQ 處理更多的流量。Kestrel 本身就是其中之一，用例也類似。兩者都有望在沒有開銷的情況下處理數百萬個請求/訊息。
寫入由WriteLoop在后臺任務中運行的方法處理：
_writerTask = Task.Run(WriteLoop);
WriteLoop方法本身不會取消。如果這樣做，它可能會留下未完成的訊息。
private async Task WriteLoop()
{
try
{
while (await _channelReader.WaitToReadAsync().ConfigureAwait(false))
{
while (_channelReader.TryRead(out ReadOnlyMemory<byte> memory))
{
MemoryMarshal.TryGetArray(memory, out ArraySegment<byte> segment);
#if NETSTANDARD
await _writer.WriteAsync(segment.Array, segment.Offset, segment.Count).ConfigureAwait(false);
#else
await _writer.WriteAsync(memory).ConfigureAwait(false);
#endif
RabbitMqClientEventSource.Log.CommandSent(segment.Count);
ArrayPool<byte>.Shared.Return(segment.Array);
}
await _writer.FlushAsync().ConfigureAwait(false);
}
}
catch (Exception ex)
{
ESLog.Error("Background socket write loop has crashed", ex);
throw;
}
}
在這段代碼_writer中是一個緩沖的網路流：
_writer = new BufferedStream(netstream, _socket.Client.SendBufferSize);
Write方法只是一個呼叫，因此_channelWriter.TryWrite它不必檢查撰寫器是否已關閉
public void Write(ReadOnlyMemory<byte> memory)
{
_channelWriter.TryWrite(memory);
}
Cancelling意味著呼叫Close完成撰寫器，一旦管道完成，關閉套接字：
try
{
_channelWriter.Complete();
_writerTask?.GetAwaiter().GetResult();
}
catch
{
// ignore, we are closing anyway
}
try
{
_socket.Close();
}
catch
{
// ignore, we are closing anyway
}
finally
{
_closed = true;
}
這可確保仍處理任何待處理的訊息。
確保有效載荷不需要大量分配更為重要。RabbitMQ 使用一個Channel.CreateUnbounded<ReadOnlyMemory<byte>>. 緩沖區本身來自ArrayPool<byte>.Shared. 發送幀后，緩沖區將釋放回池中
ArrayPool<byte>.Shared.Return(segment.Array);
在不泄漏緩沖區的情況下對管道進行核爆
在正常操作中，您無論如何都不需要取消。只有當你不得不破壞管道而不是停止發布者時，你才需要它。一種這樣的情況可能是下游作業人員無法再處理訊息，因為網路連接已終止。在這種情況下，呼叫Complete發布者仍然可以在 Channel 中留下無法再處理的訊息。
處理此問題的協作方法是在發出取消信號時丟棄訊息，而不是取消Read操作本身。如果我們使用 ArrayPool，我們必須接收這些緩沖區才能釋放它們。
RabbitMQ 代碼不是這樣作業的，但是處理這個只需要在內部回圈中做一個簡單的改變：
while (await _channelReader.WaitToReadAsync().ConfigureAwait(false))
{
while (_channelReader.TryRead(out ReadOnlyMemory<byte> memory))
{
MemoryMarshal.TryGetArray(memory, out ArraySegment<byte> segment);
if(!token.IsCancellationRequested)
{
// Normal processing
}
ArrayPool<byte>.Shared.Return(segment.Array);
}
await _writer.FlushAsync().ConfigureAwait(false);
}
這確保 worker 方法立即停止處理訊息而不會泄漏緩沖區。
RabbitMQ 中的結構和多執行緒作業者
RabbitMQ 在ConsumerDispatcherChannelBase類中使用相同的模式。在這種情況下，有效載荷是一個只讀結構WorkStruct，它還包含一個緩沖池RentedArray。
這次的不同之處在于同一個 RabbitMQ 連接可以有多個消費者，因此該類可以啟動多個作業任務：
Func<Task> loopStart = ProcessChannelAsync;
if (concurrency == 1)
{
_worker = Task.Run(loopStart);
}
else
{
var tasks = new Task[concurrency];
for (int i = 0; i < concurrency; i )
{
tasks[i] = Task.Run(loopStart);
}
_worker = Task.WhenAll(tasks);
}
關閉作業是一樣的。完成作者并等待工人退出：
protected override Task InternalShutdownAsync()
{
_writer.Complete();
return _worker;
}
轉載請註明出處，本文鏈接：https://www.uj5u.com/gongcheng/535810.html
上一篇：讀寫鎖（作者首選）實作