producer-consumer 相关问题

生产者 - 消费者问题（也称为有界缓冲问题）是多进程同步问题的经典示例。该问题描述了两个进程，生产者和使用者，他们共享一个用作队列的通用固定大小缓冲区。

I使用Confluent-kafka v1.3.0，我在消费者组会话超时方面存在以下问题。我的配置看起来像： c ['kafka'] = { 'bootstrap.servers'：'host.docker.internal：9104'，， '缺点...

python apache-kafka kafka-consumer-api producer-consumer confluent-platform

回答 1 投票 0

如何运行平行。

c# asynchronous concurrency producer-consumer parallel.foreachasync

回答 1 投票 0

使用有界通道<T>使用严格的内存分配要求

Channel<T> 在通道传播的TS是巨大的（每个1GB），这需要将任何给定时刻的数组总数限制为最小。因此，两个生产商在创建byte[]之前一直在等待，直到他们知道频道中有空的空间为止。这是第一个生产者： Channel<byte[]> channel = Channel.CreateBounded<byte[]>(2); 第二生产商是相同的。不幸的是，即使频道中只有一个空插槽，这都使两个生产商都可以开始创建一个新数组。因此，在某个时候，系统可能同时拥有4个巨大的阵列：1被消费者消耗，其中1个存储在频道中，而2个同时创建的是两个生产商（试图填充一个空的一个空插槽）。我想将托管内存中的数组总数限制为3。我有什么办法可以驯服我的生产者，这样他们才能开始创建新的byte[]渠道？换句话说，在创建数组后，生产者应该能够在这样的频道中立即写入它：new byte[1_000_000_000] ...并且Task producer1 = Task.Run(async () => { while (true) { await channel.Writer.WaitToWriteAsync(); // The channel has space available. Let's create the array. byte[] array = new byte[1_000_000_000]; // Here initialize the array (mainly I/O bound, time consuming) // At this moment the channel might be full, // because the other producer filled the gap. await channel.Writer.WriteAsync(array); } }); 始终应该是byte[]。这种情况是人为的。它的灵感来自。 clarification：字节阵列的构建和初始化是生产者的专有责任，应该保持这种状态。部分或全部分配其他地方的施工工作是不可取的。一个选项，尽管它需要一些手动管理，但它将是一个限制大小的频道的简单包装器。 bool success = channel.Writer.TryWrite(array); 该“裸机”实现的最大问题是，您必须在发布该项目时手动管理。作为替代方案，您可以返回一个“包装”实例：success有这些锁定的返回实例。 OFC然后您将分配跟踪对象，但这是围绕分配的组成和手动代码之间的权衡。 Edit每评论： true //This is for the sake of simplicity in example SingletonHolder.LockedGibiByteArrayPool = new LockingArrayPool<byte[]>(new [] { new byte[1_000_000_000], new byte[1_000_000_000] }); Task producer1 = Task.Run(async () => { while (true) { await channel.Writer.WaitToWriteAsync(); // Grab an array from our pool. byte[] array = await SingletonHolder.LockedGibiByteArrayPool.GetItemAsync(); // Here initialize the array (mainly I/O bound, time consuming) // The channel should not be full, // But the reader -must- make sure to release the array // when it is done. // alternatively, use the 'LockPooledItem` pattern suggested, // and then at least it's just a `Dispose()` call... await channel.Writer.WriteAsync(array); } }); buffered生产者/消费者系统（如通道和数据流）在最大缓冲尺寸周围有些“模糊”（我永远不记得数据流是否在输出缓冲区中计数项目是否计数）。正如您指出的那样，它们不计算生产者或消费者持有的任何物品。，为了随时限制total对象数量，您需要自己的分配器。 Disposable usage： public sealed class TokenAllocator { private readonly SemaphoreSlim _mutex; public TokenAllocator(int maxTokens) => _mutex = new(maxTokens); public async Task<IDisposable> AllocateAsync() { await _mutex.WaitAsync(); return Disposable.Create(() => _mutex.Release()); } } 注：生产者示例假定例外将导致应用程序故障。如果必须从中恢复异常，则生产商中的var allocator = new TokenAllocator(3); var channel = Channel.CreateBounded<(IDisposable token, byte[] item)>(2); var consumer = Task.Run(async () => { await foreach (var (token, item) in channel.Reader.ReadAllAsync()) using (token) { ... // Do something with `item` } }); var producer1 = Task.Run(async () => { while (true) { var token = await allocator.AllocateAsync(); try { var item = new byte[1_000_000_000]; ... // Do something with `item` } catch { token.Dispose(); throw; } await channel.Writer.WriteAsync((token, item)); } }); 需要围绕它的public sealed class LimitedAllocator<T> { private readonly SemaphoreSlim _mutex; public LimitedAllocator(int maxItems) => _mutex = new(maxItems); public async Task<AllocatedItem> AllocateAsync(Func<T> create) { await _mutex.WaitAsync(); return new(this, create()); } private void Free() => _mutex.Release(); public sealed class AllocatedItem : IDisposable { public AllocatedItem(LimitedAllocator<T> allocator, T item) { Item = item; _disposer = Disposable.Create(() => allocator.Free()); } public T Item { get; } public void Dispose() => _disposer.Dispose(); private readonly IDisposable _disposer; } } /var allocator = new LimitedAllocator<byte[]>(3); var channel = Channel.CreateBounded<LimitedAllocator<byte[]>.AllocatedItem>(2); var consumer = Task.Run(async () => { await foreach (var allocatedItem in channel.Reader.ReadAllAsync()) using (allocatedItem) { ... // Do something with allocatedItem.Item } }); var producer1 = Task.Run(async () => { while (true) { var allocatedItem = await allocator.AllocateAsync(() => new byte[1_000_000_000]); ... // Do something with allocatedItem.Item await channel.Writer.WriteAsync(allocatedItem); } }); ，仅在例外情况下处理分配的项目。在...中仅是内存（例如try）的情况下，您可以考虑使用catch而不是T。 LimitedAllocator<T>本质上是一种与内存结合的一次性。制片人不再等待查看渠道中是否有空间；它只是在等待查看loscator是否有空间。如果分配器中有可用的空间，那么该生产商将成为将创建一个要发送到渠道的项目的空间。如果您非常不喜欢byte[]with-intem的配对，那么使用连接的属性是可以使用的。但这倾向于更多的魔术和较低的可维护代码。这里是另一种方法，它也使用了IMemoryOwnerStephenCleary和To11mtm的答案。不同之处在于，信号量的AllocatedItem和IMemoryOwner被初始化为通道的确切容量，并且信号量是在从通道中取出AllocatedItem后立即释放的，而不是在完全处理该通道时。： IDisposable 本质上，SemaphoreSlim成为限制渠道能力的后卫。如果需要，您不妨使用无限频道。这种方法可确保任何时间分配的最大数量为3，不包括符合垃圾收集的资格。

c# optimization producer-consumer memory-efficient system.threading.channels

回答 2 投票 0

具有优先级的电流收集我有许多线程从服务器列表中检索数据。每5分钟从服务器解析器下载服务器列表。我用于处理数据的线程应仅使用

BlockingCollection<T>

c# multithreading collections concurrency producer-consumer

回答 1 投票 0

在生产者/消费者场景中延误IASYNCENCENOMETION 我有一个方案，我正在阅读数据库中的一些数据。这些数据以iAsyncenumerable的形式返回。阅读数据后，我想将其发送给消费者。这个

// C# IAsyncEnumerable<MyData> enumerable = this.dataSource.Read(query); await foreach (var data in enumerable) { await this.consumer.Write(data); }

c# linq async-await producer-consumer iasyncenumerable

回答 1 投票 0

使用执行权限和管道阅读器/pipedWriter（或PipedInputStream/pipedOutputStream）用于消费者生产商

我找不到使用新的并发软件包和任何一个管道类别的示例