Оркестратор бесконечных задач

[lair]

Что-то мне это болезненно напоминает модель акторов (см Akka.net или Orleans).

[oleg_gavrilov]

А если отступить на шаг назад, то и вообще можно просто взять Nomad.

[dkzkv]

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

А акторная модель, в частности Orleans, предназначена больше для обеспечения statefull бизнес-логики в облачных серверах.

[lair]

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

Ну так вроде бы и в модели акторов то же самое: актор умирает, сообщение не обработано, оно попадает обратно в диспетчер, тот перезапускает актора, переотправляет сообщение. Ничего в акторе для этого писать не надо.

А акторная модель, в частности Orleans, предназначена больше для обеспечения statefull бизнес-логики в облачных серверах.

Ээээ… с чего бы вдруг? Модель акторов предназначена для обработки сообщений, а облака там вообще не при чем.

Ну то есть да, конечно, я с этим копался года два или больше назад, и мог все забыть и перепутать, врать не буду. Но для меня схожесть очевидна.

[dkzkv]

По поводу облаков прокомментирую, Orleans показывает всю свою мощь только в кластерах/облаках. То есть его нет смысла использовать если у вас всего один сервер и нет необходимости распределять нагрузку.

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

Но смотрите, в чем мне кажется основное различие, задача/task в статье берется из очереди сообщений (важно отметить что сообщения должны не храниться в памяти, а быть записаны на диск), таким образом мы можем не переживать за то, что она будет выполняться. Активные сервисы просто будут «арендовать» задачу.
С орлеанс же подход немного другой. Диспетчер отправляет сообщение на исполнение, но отправляет выстраивая WS соединение. Весь мессаджинг происходит по TCP. То есть информация о задаче существует только inMemory (vs записана на диск.)

В дополнение. Подобный подход для управления бесконечных задач, на мой взгляд более безопасный. По личному опыту добавлю, имплементирование акторов, как по мне затратнее. И еще хочу отметить: похожая оркестрация задач есть у Kafka, разница лишь в том что там мы не можем динамически менять количество исполнителей.

[stristan]

Статья очень интересная. спасибо!
Вопрос — нет ли у Вас исходников готового примера для использования, чтобы не городить самому из описания в статье?

[dkzkv]

Здравствуйте, подобный подход реализован в проекте:
github.com/dkzkv/ArbitralSystem
Обратите внимание на:
ArbitralSystem.Distributor.MQDistributor.MQManagerService — Это Observer manager
ArbitralSystem.Distributor.MQDistributor.MQOrderBookDistributorService — а это Observer service

Если кристаллизировать все что было написано, то для оркестрации задач понадобится всего пара строчек кода:
Observer service
ObserverConsumer .cs

public class ObserverConsumer : IConsumer<IStartJob> //IConsumer - MassTransit
{
    private readonly ObserverJob _observerJob;
    public JobConsumer(observerJob ourJob)
    {
        _observerJob = observerJob;
    }

    public async Task Consume(ConsumeContext<IStartJob> context)
    {
        var jobArg = context.Message;
        //Запуск (бесконечной) задачи.
        //Важно сохранить токен отмены в памяти, тогда другим сообщением мы можем вызвать его и завершить задачу.
        _observerJob.Execute(jobArg, new CancellationTokenSource().Token);
    }
}

Program.cs

cfg.ReceiveEndpoint("observer-queue-name", e =>
{
    e.UseMessageRetry(r => r.Immediate(int.MaxValue));
    //а тут мы задаем сколько сервис может обрабатывать задач.
    e.PrefetchCount = serviceOptions.MaxWorkersCount;
    //Подключаем обработчик сообщений.
    e.ConfigureConsumer<ObserverConsumer>(provider);
});

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

[stristan]

Спасибо большое! Поковыряю на досуге