RavenDB Stream для неограниченных результатов - устойчивость соединения

Ответ 1

В соответствии с предложением @StriplingWarrior я воссоздал решение, используя Подписки данных.

Используя этот подход, я смог перебрать все 2 миллиона строк (хотя, по общему признанию, с гораздо меньшей обработкой на элемент); 2 пункта здесь, которые помогли бы, когда мы пытались реализовать ту же логику с помощью Streams:

• Пакеты только удаляются из подписной "очереди" после подтверждения (как и большинство стандартных очередей)
  • Подписанный IObserver<T> должен завершиться успешно для этого подтверждения, которое должно быть установлено.
  • Эта информация обрабатывается сервером, а не клиентом, поэтому позволяет перезагрузить клиент, не затрагивая последнюю успешную позицию, обработанную в подписке.
  • Подробнее см. здесь
• Поскольку @StriplingWarrior указан потому, что вы можете создавать подписки с фильтрами вплоть до уровня свойств, можно было бы воспроизвести с меньшим набором результатов в случае исключения в самой подписке.
  • Первая точка действительно заменяет это; но это позволяет нам получить дополнительную гибкость, не проявляемую в Stream API

Среда тестирования - это база данных RavenDB 3.0 (локальная машина, работающая как служба Windows) с настройками по умолчанию для коллекции из 2 миллионов записей.

Код для создания фиктивных записей:

using (IDocumentStore store = GetDocumentStore())
{
    store.Initialize();

    using (var bulkInsert = store.BulkInsert())
    {
        for (var i = 0; i != recordsToCreate; i++)
        {
            var person = new Person
            {
                Id = Guid.NewGuid(),
                Firstname = NameGenerator.GenerateFirstName(),
                Lastname = NameGenerator.GenerateLastName()
            };

            bulkInsert.Store(person);
        }
    }
}

Подписывание этой коллекции тогда является случаем:

using (IDocumentStore store = GetDocumentStore())
{
    store.Initialize();

    var subscriptionId = store.Subscriptions.Create(new SubscriptionCriteria<Person>());

    var personSubscription = store.Subscriptions.Open<Person>(
        subscriptionId, new SubscriptionConnectionOptions()
    {
        BatchOptions = new SubscriptionBatchOptions()
        {
            // Max number of docs that can be sent in a single batch
            MaxDocCount = 16 * 1024,  
            // Max total batch size in bytes
            MaxSize = 4 * 1024 * 1024,
            // Max time the subscription needs to confirm that the batch
            // has been successfully processed
            AcknowledgmentTimeout = TimeSpan.FromMinutes(3)
        },
        IgnoreSubscribersErrors = false,
        ClientAliveNotificationInterval = TimeSpan.FromSeconds(30)
    });

    personSubscription.Subscribe(new PersonObserver());

    while (true)
    {
        Thread.Sleep(TimeSpan.FromMilliseconds(500));
    }
}

Обратите внимание на PersonObserver; это просто базовая реализация IObserver:

public class PersonObserver : IObserver<Person>
{
    public void OnCompleted()
    {
        Console.WriteLine("Completed");
    }

    public void OnError(Exception error)
    {
        Console.WriteLine("Error occurred: " + error.ToString());
    }

    public void OnNext(Person person)
    {
        Console.WriteLine($"Received '{person.Firstname} {person.Lastname}'");
    }
}