Сообщения теряются при использовании из BlockingCollection в пакетах - PullRequest
Новая
       30

Сообщения теряются при использовании из BlockingCollection в пакетах

0 голосов
/ 30 августа 2018

Я пытался изобрести способ потребления партии из BlockingCollection и попал в беду.

Вот минимальное повторение: 

internal class Program
{
    private static readonly BlockingCollection<string> _bc = new BlockingCollection<string>(1000);
    private static int _consumed;

    static void Main()
    {
        Task.Run(() => Producer());
        Task.Run(() => Consumer());
        Console.WriteLine("press [ENTER] to check");
        while (true)
        {
            Console.ReadLine();
            Console.WriteLine("consumed: " + _consumed);
        }
    }

    private static void Producer()
    {
        for (var i = 0; i < 5000; i++)
            _bc.Add("msg");
    }

    private static void Consumer()
    {
        foreach (var s in _bc.GetConsumingEnumerable())
        {
            var batchSize = _bc.Count + 1;
            var batch = new List<string>(batchSize) { s };
            while (_bc.TryTake(out var additionalResult) && batch.Count < batchSize)
                batch.Add(additionalResult);
            _consumed = _consumed + batch.Count;
        }
    }
}

Несколько сообщений теряются (но не всегда одно и то же число). Если вы не можете воспроизвести его, попробуйте увеличить количество создаваемых сообщений.

То, чего я пытаюсь добиться, - это использовать метод GetConsumingEnumerable в приемнике (через некоторое время я назову CompleteAdding) и иметь возможность собирать пакеты сообщений некоторого размера, если они уже присутствует.

В чем причина потери сообщений и как правильно их использовать?

Ответы [ 2 ]

0 голосов
/ 30 августа 2018 
 [__DynamicallyInvokable]
    public IEnumerable<T> GetConsumingEnumerable(CancellationToken cancellationToken)
    {
      ...
        while (!this.IsCompleted)
        {
          T obj;
          if (this.TryTakeWithNoTimeValidation(out obj, -1, cancellationToken, linkedTokenSource))
            yield return obj;
        }
      ...
    }

и 

public bool TryTake(out T item)
{
  ...
  return this.TryTakeWithNoTimeValidation(out item, (int) timeout.TotalMilliseconds, CancellationToken.None, (CancellationTokenSource) null);
}

оба метода TryTake и GetConsumingEnumerable используют метод TryTakeWithNoTimeValidation. Я предполагаю, что отсутствующие элементы были удалены из коллекции с помощью GetConsumingEnumerable. рассмотрим следующий пример: 

private static void Producer()
{
    Console.WriteLine($"begin produce isCompleted:{_bc.IsCompleted}");
    for (var i = 0; i < 5000; i++)
        _bc.Add($"msg:{i}");
    _bc.CompleteAdding();
    Console.WriteLine($"end produce isCompleted:{_bc.IsCompleted}");
}
var batch = new List<string>();
foreach (var s in _bc.GetConsumingEnumerable())
{
    batch.Add(s);
    if (_bc.IsCompleted && _bc.Count == 0)
    {
       break;
    }
}
Console.WriteLine($"first:{batch.First()}, last:{batch.Last()}");
Console.WriteLine($"consumed:{batch.Count}");

_bc пусто. Есть несколько способов реализовать ваш алгоритм, один из которых я рекомендую использовать Take и вызывать потребителя перед производителем (который блокирует вызывающий поток).

0 голосов
/ 30 августа 2018

Ничего себе. Это ошибка. Эта строка 

while (_bc.TryTake(out var additionalResult) && batch.Count < batchSize)

должно быть 

while (batch.Count < batchSize && _bc.TryTake(out var additionalResult))

, поскольку первое условие имеет побочный эффект удаления элемента из коллекции.

...