awaitable Tâche de la file d'attente

Je ne sais pas de lock-solution gratuite, mais vous pouvez prendre un coup d'oeil à la nouvelle Bibliothèque de flux de données, une partie de la Async CTP. Un simple BufferBlock<T> devrait suffire, par exemple:

BufferBlock<int> buffer = new BufferBlock<int>();

De Production et de consommation sont plus facilement mises en œuvre via des méthodes d'extension sur le flux de données de types de bloc.

De Production est aussi simple que:

buffer.Post(13);

et de la consommation est asynchrone de prêt:

int item = await buffer.ReceiveAsync();

Je ne vous recommandons l'utilisation de Flux de données si possible; la fabrication d'un tel tampon à la fois efficace et correcte est plus difficile qu'il n'y paraît au premier abord.

Mon atempt (il a un évènement soulevée lors de la "promesse" est créé, et il peut être utilisé par un producteur externe pour savoir quand produire plus d'articles):

public class AsyncQueue<T>
{
private ConcurrentQueue<T> _bufferQueue;
private ConcurrentQueue<TaskCompletionSource<T>> _promisesQueue;
private object _syncRoot = new object();
public AsyncQueue()
{
_bufferQueue = new ConcurrentQueue<T>();
_promisesQueue = new ConcurrentQueue<TaskCompletionSource<T>>();
}
///<summary>
///Enqueues the specified item.
///</summary>
///<param name="item">The item.</param>
public void Enqueue(T item)
{
TaskCompletionSource<T> promise;
do
{
if (_promisesQueue.TryDequeue(out promise) &&
!promise.Task.IsCanceled &&
promise.TrySetResult(item))
{
return;                                       
}
}
while (promise != null);
lock (_syncRoot)
{
if (_promisesQueue.TryDequeue(out promise) &&
!promise.Task.IsCanceled &&
promise.TrySetResult(item))
{
return;
}
_bufferQueue.Enqueue(item);
}            
}
///<summary>
///Dequeues the asynchronous.
///</summary>
///<param name="cancellationToken">The cancellation token.</param>
///<returns></returns>
public Task<T> DequeueAsync(CancellationToken cancellationToken)
{
T item;
if (!_bufferQueue.TryDequeue(out item))
{
lock (_syncRoot)
{
if (!_bufferQueue.TryDequeue(out item))
{
var promise = new TaskCompletionSource<T>();
cancellationToken.Register(() => promise.TrySetCanceled());
_promisesQueue.Enqueue(promise);
this.PromiseAdded.RaiseEvent(this, EventArgs.Empty);
return promise.Task;
}
}
}
return Task.FromResult(item);
}
///<summary>
///Gets a value indicating whether this instance has promises.
///</summary>
///<value>
///<c>true</c> if this instance has promises; otherwise, <c>false</c>.
///</value>
public bool HasPromises
{
get { return _promisesQueue.Where(p => !p.Task.IsCanceled).Count() > 0; }
}
///<summary>
///Occurs when a new promise
///is generated by the queue
///</summary>
public event EventHandler PromiseAdded;
}

Découvrez https://github.com/somdoron/AsyncCollection, vous pouvez retirer de façon asynchrone et C# 8.0 IAsyncEnumerable.

L'API est très similaire à BlockingCollection.

AsyncCollection<int> collection = new AsyncCollection<int>();
var t = Task.Run(async () =>
{
while (!collection.IsCompleted)
{
var item = await collection.TakeAsync();
//process
}
});
for (int i = 0; i < 1000; i++)
{
collection.Add(i);
}
collection.CompleteAdding();
t.Wait();

Avec IAsyncEnumeable:

AsyncCollection<int> collection = new AsyncCollection<int>();
var t = Task.Run(async () =>
{
await foreach (var item in collection)
{
//process
}
});
for (int i = 0; i < 1000; i++)
{
collection.Add(i);
}
collection.CompleteAdding();
t.Wait();

Vous pouvez simplement utiliser un BlockingCollection ( à l'aide de la valeur par défaut ConcurrentQueue ) et l'enveloppe de l'appel à Take dans un Task de sorte que vous pouvez await c':

var bc = new BlockingCollection<T>();
T element = await Task.Run( () => bc.Take() );