Блог

Контекст

В моем консольном приложении .NET 5 мой потребитель запускается как поток (задача). В настоящее время существует только один, но в будущем возможно одновременное использование пользователей в одном и том же экземпляре BufferBlock.

Текущая реализация работает нормально. Я застрял с реализацией следующей добавленной функции:

Я хотел бы просыпаться в настроенный период, даже если буфер пуст, и все равно оставаться в основном цикле. Этот вариант использования должен отличаться от варианта пробуждения по IsCancellationRequested токена, но необязательно отличать его от «продукта, доступного для использования», не обязательно.

Я вижу перегрузку ReceiveAsync с таймаутом, но неясно, как OutputAvailableAsync это происходит, которая не принимает тайм-аут.

Вопрос

Как реализовать пробуждение в течение заданного периода времени и оставаться в цикле. Только в том случае, если IsCancellationRequested должен прервать цикл

 public class MyConsumer
{
    private readonly BufferBlock<MyProduct> _products;

    public void Start(CancellationToken token)
    {
        Task.Factory.StartNew(() => Run(token), token);
    }

    private async Task Run(CancellationToken token)
    {
        await ConsumeAsync(token);
    }

    private async Task ConsumeAsync(CancellationToken token)
    {
        while (await _products.OutputAvailableAsync(token))
        {
            var product = await _products.ReceiveAsync(token);
            // Consume product goes here:...

            // I would like wake up here in a configured period, even the buffer is empty.
            //
            // How to implement this timeout based wake up? (then still remain in the loop)
            // I do not even understand clearly why are we using the two waiting operations, 
            // the 1) OutputAvailableAsync(token) then 2) ReceiveAsync(token)
        }
    }
}
 

Я не очень хорошо знаком с BufferBlock , но в общем смысле: если async API не предоставляет тайм-аут, вы можете имитировать то же самое с помощью токена отмены с тайм-аутом — и при этом уважать существующий токен:

 using var cts = new CancellationTokenSource();
// link the existing CancellationToken so that *it* can propagate cancellation
using var linked = token.Register(
    static s => ((CancellationTokenSource)s).Cancel(), cts, false);
// add a timeout
cts.CancelAfter(yourTimeoutHere);
// use this new combined token to do the magic
await DoSomethingAsync(cts.Token);
 

(если вы не используете C # 9 или выше, удалите static ; это просто проверяет, что мы не вызываем дополнительных выделений из-за захваченных переменных в обратном вызове)

Как только у вас это будет на месте, вы можете просто реагировать на свой тайм-аут как на дополнительную работу, перехватывать OperationCanceledException и делать то, что вам нужно. Небольшое предостережение заключается в знании того, в каком состоянии вещи останутся в DoSomethingAsync случае отмены. Это будет зависеть от конкретного сценария.