Rabbitmq

Однажды я вел вебинар про то, как принимать 10 000 ивентов в секунду. Показал вот такую картинку, зрители увидели сиреневый слой, и началось: Ребят, а зачем нам все эти кафки и рэббиты, неужели без них не обойтись? Мы и ответили: Зачем-зачем, чтобы пройти собес!

Очень смешно, но давайте я все-таки объясню.

Мы можем принимать ивенты сразу в зеленой области и заставить наши приложения писать их в кликхаус.

Но кликхаус любит, когда в него пишут сообщения пачками

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

Как бывает: в одну секунду пришло 10 тысяч ивентов, в следующую тысяча, в другую 50 тысяч. Это нормально, пользователи рандомно включают свои мобильные приложения. В таком случае в кликхаус напрямую будет заходить то 2 тысячи, то 10 тысяч сообщений. Но с помощью буфера вы можете подкопить сообщения, потом достать из этой копилки миллион ивентов и направить в кликхаус. И вот она желанная стабильная нагрузка на ваш кластер.

Это все история про очереди

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

Например, для бэкграунд задач. Вы заходите в админку магазина и генерируете отчет по продажам за год. Задача трудоемкая: нужно прочитать миллионы строк из базы, это хлопотно и очень долго. Если клиент будет висеть постоянно с открытым http-коннектом 5, 10 минут связь может оборваться, и он не получит файл.

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

Второй кейс про кучу микросервисов, которые общаются через шину.

Например, один сервис принимает ивенты от пользователей, передает их в очередь. Следующий сервис вытаскивает ивенты и нормализует их, к примеру, проверяет, чтобы у них был валидный e-mail или телефон. Если все хорошо, он перекладывает сообщение дальше, в следующую очередь, из которой данные будут записываться в базу.

Еще один поинт это падение дата-центра, в котором хостится база.

Конец, ничего не работает. Что будет с сообщениями? Если писать без буфера, сообщения потеряются. Кликхаус недоступен, клиенты отвалились. До какого-то предела выручит память, но с буфером безопаснее вы просто взяли и записали сообщения в очередь (например, в кафку). И сообщения будут храниться там, пока не закончится место или пока их не прочитают и не обработают.

Как автоматически добавлять новые виртуалки при увеличении нагрузки

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

Мы создаем инстанс-группу. Задаем ей имя и указываем сервисный аккаунт. Он будет использоваться для создания виртуалок.

resource "yandex_compute_instance_group" "events-api-ig" {  name               = "events-api-ig"  service_account_id = yandex_iam_service_account.instances.id

Затем указываем шаблон виртуалки. Указываем CPU, память, размер диска и т.д.

instance_template {    platform_id = "standard-v2"    resources {      memory = 2      cores  = 2    }    boot_disk {      mode = "READ_WRITE"      initialize_params {        image_id = data.yandex_compute_image.container-optimized-image.id        size = 10      }

Указываем, к какому сетевому интерфейсу его подрубить.

}    network_interface {      network_id = yandex_vpc_network.internal.id      subnet_ids = [yandex_vpc_subnet.internal-a.id, yandex_vpc_subnet.internal-b.id, yandex_vpc_subnet.internal-c.id]      nat = true    }

Самое интересное это scale_policy.

Можно задать группу фиксированного размера fixed scale с тремя инстансами A, B, C.

scale_policy {    fixed_scale {      size = 3    }  }  allocation_policy {    zones = ["ru-central1-a", "ru-central1-b", "ru-central1-c"]  }

Либо использовать auto_scale тогда группа будет автоматически масштабироваться в зависимости от нагрузки и параметров.

scale_policy {auto_scale {    initial_size = 3    measurment_duration = 60    cpu_utilization_target = 60    min_zone_size = 1    max_size = 6    warmup_duration = 60    stabilization_duration = 180}

Главный параметр, на который надо обратить внимание, это cpu utilization target. Можно выставить значение, при превышении которого Яндекс.Облако автоматически создаст нам новую виртуалку.

Теперь протестируем автомасштабирование при увеличении нагрузки

Наше приложение принимает различные ивенты, проверяет джейсонку и направляет в кафку.

Перед нашей инстанс-группой стоит load-балансер. Он принимает все запросы, которые приходят на адрес 84.201.147.84 на порту 80, и направляет их на нашу инстанс-группу на порт 8080.

У меня есть виртуалка, которая с помощью Yandex.Tank делает тестовую нагрузку. Для теста я установил 20 тысяч запросов в течение 5 минут.

Итак, нагрузка пошла.

Сначала все ноды будут загружены во всех трех зонах (A, B и C), но когда мы превысим нагрузку, Яндекс.Облако должно развернуть дополнительные инстансы.

По логам будет видно, что нагрузка выросла и в каждом регионе количество нод увеличилось до двух. В балансировку добавилась еще одна машина, количество инстансов тоже везде увеличилось.

При этом у меня был интересный момент. Один инстанс, который находится в регионе С, записывал данные (от момента приема данных до записи) за 23 миллисекунды, а у инстанса из региона А было 12,8 миллисекунд. Такое происходит из-за расположения кафки. Кафка находится в регионе А, поэтому в нее записи идут быстрее.

Ставить все инстансы кафки в одном регионе не надо.

Когда добавилась еще одна машина, новая нагрузка спала, показатель CPU вернулся к норме. Полную аналитику по тестовому запуску можно посмотреть по ссылке: overload.yandex.net/256194.

Как написать приложение для работы с очередями и буферами обмена

Приложение написано на golang. Сначала мы импортируем встроенные модули.

package mainimport (    "encoding/json"    "flag"    "io"    "io/ioutil"    "log"    "net/http"    "strings")

Затем подключаем github.com/Shopify/sarama это библиотека для работы с кафкой.

Прописываем github.com/prometheus/client_golang/prometheus, чтобы метрики передавались в API Metrics.

Также подключаем github.com/streadway/amqp для работы с rabbitmq.

Затем следуют параметры бэкендов, в которые мы будем записывать.

var (    // Config options    addr     = flag.String("addr", ":8080", "TCP address to listen to")    kafka    = flag.String("kafka", "127.0.0.1:9092", "Kafka endpoints")    enableKafka    = flag.Bool("enable-kafka", false, "Enable Kafka or not")amqp    = flag.String("amqp", "amqp://guest:guest@127.0.0.1:5672/", "AMQP URI")enableAmqp    = flag.Bool("enable-amqp", false, "Enable AMQP or not")sqsUri    = flag.String("sqs-uri", "", "SQS URI")sqsId    = flag.String("sqs-id", "", SQS Access id")sqsSecret    = flag.String("sqs-secret", "", "SQS Secret key")enableSqs    = flag.Bool("enable-sqs", false, "Enable SQS or not")        // Declaring prometheus metrics    apiDurations = prometheus.NewSummary(        prometheus.SummaryOpts{            Name:       "api_durations_seconds",            Help:       "API duration seconds",            Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},        },    )

Адрес кафки (строка).

Включить кафку или нет поскольку приложение может писать в несколько разных бэкендов.

В приложении реализована возможность работы с тремя очередям.

Первое это кафка.

Второе amqp для рэббита.

И третья очередь sqs для Яндекс.Кью.

Дальше мы открываем и задаем общие глобальные переменные для работы с нашим бэкендом. Прописываем настройки prometheus для отображения и визуализации.

В main мы включаем кафку, рэббит и создаем очередь с названием Load.

И если у нас включен sqs, мы создаем клиент для Яндекс.Кью.

Дальше наше приложение по http принимает несколько инпоинтов:

/status просто отдает okey, это сигнал для load-балансера, что наше приложение работает.

Если вы кидаете запрос на /post/kafka, ваша джейсонка попадет в кафку. Также работают /post/amqp и /post/sqs.

Как работает кафка

Кафка простой, некапризный и очень эффективный инструмент. Если вам нужно быстро принять и сохранить много сообщений, кафка к вашим услугам.

Как-то на одном из проектов важно было уложиться в маленький бюджет. И вот представьте, мы берем самые дешевые машины без SSD (а кафка пишет последовательно и читает последовательно, так что можно не тратиться на дорогие диски), ставим кафку и zookeeper. Наше скромное решение на три ноды спокойно выдерживает нагрузку 200 тысяч сообщений в секунду! Кафка это про поставил и забыл, за пару лет работы кластер ни разу нас не потревожил. И стоил 120 евро в месяц.

Единственное, что нужно запомнить кафка очень требовательна к CPU, и ей очень не нравится, когда кто-то рядом поджирает проц. Если у нее будет сосед под боком, она начнет тормозить.

Кафка устроена так: у вас есть topic, можно сказать, что это название очереди. Каждый topic бьется на части до 50 partitions. Эти партиции размещаются на разных серверах.

Как вы видите на схемке, topic load разбит на 3 партиции. Partition 1 оказывается на Kafka 1, вторая партиция на кафка 2, третья на 3. Тем самым нагрузка полностью распределяется. Когда кластер начинает принимать нагрузку, сообщения пишутся в один топик, а кафка раскидывает их по партициям, гоняет их по кругу. В итоге все ноды нагружаются равномерно.

Можно заморочиться и разбить топик на 50 партиций, поставить 50 серверов и расположить на каждом сервере по 1 партиции нагрузка распределится на 50 нод. И это очень круто.

Партиции могут реплицироваться благодаря zookeeper. Кафке необходимо минимум 3 ноды зукипера. Например, вы хотите, чтобы ваша партиция реплицировались на 2 ноды. Указываете репликейшн фактор 2 и каждая партиция будет закинута 2 раза на рандомные хосты. И если ваша нода упадет, то благодаря зукиперу кафка это увидит: ага, первая нода в дауне, кафка 2 заберет себе первую партицию.

Как я разворачивал кафку с помощью Terraform

В репозитории у нас есть terraform-файл, он называется kafka.tf .

Вначале мы поднимем 3 зукипера: resource yandex compute instance zookeeper count = 3.

Потом находим zookeeper_deploy, который деплоит наш зукипер. Хорошо, если он будет вынесен на отдельные машины, где кроме него ничего нет. Далее собираем айдишники нод и генерируем файл. Запускаем ansible для настройки зукипера.

Кафку поднимаем аналогично зукиперу и, что важно, после него.

Как работает RabbitMQ

Благодаря тому, что кафка по сути просто сохраняет сообщения на диск и по запросу отдает клиенту с нужного места, она очень и очень шустрая. Производительность рэббита значительно ниже, но он напичкан фичами под завязку! Рэббит пытается сделать очень многое, что естественным образом влечет за собой потребление ресурсов.

Рэббит уже не так прост тут вам и exchanges с роутингом, и куча плагинов для delayed messages, deadletter и прочего хлама. За сообщениями следит сам кролик. Как только консьюмер подтвердил обработку сообщения, оно удаляется. Если консьюмер отвалился посередине рэббит вернет сообщение в очередь. В общем, хороший комбайн, когда нужно перекидывать сообщения между сервисами. Цена этого производительность.

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

Если вам нужно перекидывать сообщения между сервисами в небольшом количестве ваш выбор однозначно RabbitMQ. Если вам необходимо быстро сохранять кучу событий метрики от клиентов, логи, аналитика и т.д. ваш выбор kafka. Подробнее о сравнении двух инструментов можно прочитать в моей статье.

И еще: рэббиту не нужен зукипер.

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

В большинстве случаев всяenterpriseразработка сводится к выполнению одних и тех же требований: создается заявка, в зависимости от типа заявки у нее есть какой-то жизненный цикл, по завершению жизни заявки мы получаем (или не получаем) желаемое. Под заявкой мы можем подразумевать все что угодно, начиная с покупки в интернет-магазине товара, денежного перевода или расчета траектории баллистической ракеты. У каждой заявки есть свой жизненный путь и что важно отметить -время жизни, и чем меньше это время, тем лучше. Иными словами, чем быстрее мой банковский перевод осуществится, тем лучше. Требования тоже схожи, побольшеRPCoperationspersecond, поменьшеLatency, система должна быть отказоустойчивой, масштабируемой и должна быть готовавчера. Есть миллион инструментов, сотни баз данных, различные подходы и паттерны. И все уже давно написано, нам остается лишь правильно использовать готовые технологии в наших проектах.

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

Давайте сначала поймем, что значит бесконечная задача и где в природе такое вообще может встречаться. Бесконечная задача это некий процесс (Job), который выполняет работу до тех пор, пока ему не скажут прекратить это. Аналогию можно провести с бесконечными циклами. В природе же подобное встречается, когда нам нужны наблюдатели, которые следят и реагируют на определённые события. Например: нам необходимо следить за изменениями цен на бирже, повышением или понижением цены актива. Представим нам нужно следить за всеми валютами, всеми активами, на разных биржах, тогда количество наблюдателей может превышать десятки тысяч единиц. Что же из себя может представлять наблюдатель- это может быть отдельноеWebSocketсоединение, которое должно быть постоянноconnected. Этот наблюдатель, может получать данные,денормализовывать, производить расчеты, сохранять и много чего еще. Для удобства, наблюдателем я буду называть неObserverиз известного паттерна, а модуль, который постоянно в работе и бесконечно долго выполняет полезную работу.

Очевидно, нам нужна распределенная система, так как на ноутбуке ее явно не запустишь. Сформулируемтребование длянаших наблюдателей:

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

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

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

• Сервис/приложение, должно иметь ограничение на количество наблюдателей, которые могут быть в работе и задаваться данное значение должно через файл конфигурации или рассчитываться исходя из мощности сервера (количество ядер, RAM и т.д.), на котором сервис работает.

Упрощенно: имеем N Сервисов, каждый сервис имеет несколько наблюдателей. Пока не будем задаваться вопросом о том, как это все работает и каким образом компоненты взаимодействуют, разберем это чуть позднее.

Статья описана в 3 актах. Все листинги с кодом на С#, но в процессе написания старался уделять меньше внимания примерам с кодом и больше самой идеи. Поэтому листинги должны быть понятны даже тем людям, которые вообще не писали на C# и не знакомы с .Net.

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

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

3. Очередь, которая думает, что она планировщик. Финальная часть статьи, где мы реализуем системуоркестрациизадач через очереди сообщений. Я использовалRabbitMq, и какFramework-MassTransit, поэтому все примеры будут тесно связаны с данными инструментами. Но принцип будет оставаться тот же.

Всё естьTask

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

Рассмотрим на простом примере. Возьмём метод, которыйпишет HelloWord в консольотправляет письмо:

public async Task SendEmailAsync(Email email, CancellationToken token) {    // отправляем письмо }

Чтобы отправить письмо, не дожидаясь получения результата, нам достаточно просто забыть поставитьawaitперед вызовомSendEmailAsync.

foreach (var email in emails {    if(token.IsCancellationRequested)         break;     _emailSender.SendEmailAsync(email, token); //нет await } 

Минусов у данного подхода много:

• Мы никак не гарантируем выполнение отправки письма.

• FireAndForgetи как следствие о возникновенииExceptionмы не узнаем.

• Так же не узнаем и о выполнении.

• Многие считают, что это грех большой, вообщеантипаттерни я с ними согласен.

Более детально о том почему желательно рано или поздноawait-ить таску, можно почитатьпро async/await антипаттерны.

Наша задача во многом похожа на отправкуemail, только внутри у нас будет подобие бесконечного цикла и метод закончит работу естественным путем только тогда, когда будет вызванCancellationToken. Мы можем, конечно, написать свои костыли, которые позволят нам отслеживать состояние задачи и уведомлять, когда она завершилась. У нее будутRetryPolicyи много чего ещё, но зачем?! Когда есть уже готовые планировщики задач, которые заточены под данные требования.

Schedulers

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

• HangFire

• Quartz.net

Тутесть неплохое сравнение планировщиков. Больше всего нас интересует возможность иметь неограниченное количествосерверов, (тут может быть недопонимание, сервер это не физическая машина, где выполняется наше приложение, этоinstanceпланировщика)где будут исполнятся наши задачи/Tasks. Лично я отдал предпочтениеHangfire, по большей части из-за хорошо описанной документации и встроенного UI, который позволяет не только отображать метрики по задачам, но и вручную запускать их. Всё это весьма приятные бонусы.

А теперь посмотрим на то, как отправить наше письмо с использованиемHangfire. В этом нам поможет статический методBackgroundJob.Enqueue(Expression<Action>methodCall).

var jobIds = new List<string>(); foreach (var email in emails) {    if(token.IsCancellationRequested)       break;    jobIds.Add(BackgroundJob.Enqueue(       async () => await _emailSender.SendEmailAsync(email, token))); } 

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

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

_observer.DoWork(observerArg,newCancellationToken())

Мы передаем какие-то аргументы для работы и главное, передаем токен отмены.Для этого нам потребуется указать еще имя очереди в созданномBackgroundJobClient.

var client = new BackgroundJobClient(JobStorage.Current);//задаем имя очереди, где будет хоститься задача.var state = new EnqueuedState(unique-queue-name); client.Create(() =>_observer.DoWork(observerArg,newCancellationToken()), state);

И конечно же мы должны иметь сервис, который займется обработкой данной очереди. В настройках которого будет указано имя той самой очереди-unique-queue-name.

// Настраиваем instance hangfire сервера. _server = new BackgroundJobServer(new BackgroundJobServerOptions() {       WorkerCount = 10,     Queues = new[] { unique-queue-name },     ServerName = _serverOptions.ServerName }); 

WorkerCount- отвечает за то, сколько сервер может одновременно обрабатывать задач. Запомним ее, так как в последствии мы о ней будем много говорить.

Теперь у нас есть возможность запускать любое количество задач на разных серверах, указывая очередь, которую слушает сервис. Пока нам не хватает только одного: возможности мониторинга. Мы должны понимать какой сервер свободен, а какой нет и запускать задачи на свободном. Для этого вHangfireесть статический класс, которой предоставляет все метрики, начиная с того сколько серверов сейчас активно и заканчивая информированием о том сколько раз задача была выпалена с ошибкой.

_monitoringApi = JobStorage.Current.GetMonitoringApi(); 

Наша система с планировщиком теперь будет выглядеть следующим образом:

Observer-service - сервис, который может выполнять одновременно несколько задач, количество задается через конфиг или рассчитывается с учетом количества ядер и мощности сервера (ВHangFilreэтоWorkerCount).

Observer-manager- сервис, который отвечает за... наблюдателей.Валидируетзапросы, решает на каком сервисе будет запущен наблюдатель, а также имеет возможность удалить его. Он знает сколько сейчас доступно сервисов и на сколько каждый из них загружен.

Schedulercommondbпсевдо-очередьи хранилище всей информации по задачам,Hangfireподдерживает какMsSql, такPostgreSqlи дажеRedis.

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

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

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

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

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

3)Добавление задачи в очередь возвращает числовой идентификатор и идентифицировать задачу можно только по нему. Нельзя задать свойcustom-id, например поиск по названию. Поэтому получив идентификатор задачи нужно сохранить связку идентификатор-название в собственное хранилище.

4)В случае ошибки во время исполнения задачи, она автоматически будет перемещена в default очередь. Крайне неприятный момент, о котором узнал уже на этапе тестирования, так как в документации о таком не рассказали. Решается черезjob-filtersили черезатрибуты. Второй вариант делает код более связанным и не подходит, так как значение атрибута не может задаваться динамически.

5)В случае если сервер откажет, задачи, которые на нем исполнялись, не будут перераспределены между работающими. Можно, конечно, реализовать данную логику в менеджере и сделать его ответственным за это, но хотелось бы чтобыframeworkумел это из коробки.

6)Отсутствие транзакционности, ВедьHangfireуниверсален как дляMsSql, так и дляRedis, а в нем транзакции не предусмотрены.

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

Очередь, которая думает, что она планировщик

Намучившись, пытаясь использовать верблюда как лошадь, перейдем к описанию того, как это можно сделать через очередь. Но сперва зададимся вопросом. Что мы знаем о планировщиках? Мы знаем, что планировщики это системы, которые выполняют задачи с учетом заданных правил, расписания. А что мы знаем об очередях, шинах данных? Мы используем очереди как транспорт данных, как средство доставки сообщений. Конечно же это все очень абстрактно, и тут можно говорить часами, но пока ограничимся этим. Давай те изменим шаблонное мышление и на время представим, что очередь тоже может быть планировщиком.

Как же сделать из очереди сообщений планировщик задач? Хотя тут корректней был бы термин оркестратор.Весь вышеописанный функционал решается использованием только одной настройки-PrefetchCountи особенностью обработки сообщений.

• Когда сообщение попадает в очередь оно имеет состояниеReady.

• КогдаConumerобрабатывает сообщение, оно переходит в состояниеUnacked. И другойConsumerможет взять следующие сообщение из очереди.

• Если в момент обработки сообщения происходит ошибка, оно помещается в _Errorочередь.

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

И теперь главное -PrefetchCountэто количество одновременно обрабатываемых сообщений в очереди, а если сообщение никогда не будет дочитано (бесконечно обрабатывается), то его можно воспринимать какWorkerCount, прям как уHangfire.

Разберем на пальцах:

На данной схеме у нас есть триObserver-services, каждый из них слушает очередь в ожидании поступления сообщения.PrefetchCountу каждого стоит 1. Это значит, что за раз каждый сервис будет обрабатывать одно сообщение. А так как мы знаем, что сообщение это запуск бесконечной задачи, то оно никогда не прочитается и всегда будет в состоянииUnacked.

Дадим команду на создание двух "наблюдателей, таким образом в очереди у нас окажется два сообщения:

Так какObserver-servicesслушают одну и ту же очередь, то сообщения между ними будут распределятся равномерно, черезRound-robin.

• msg1поступает в очередь. Его начинает обрабатывать один из свободныхконсьюмеров, допустим Observer1. Сообщение переходит в состояниеUnackedи теперь новые сообщения, которые поступят в очередь будут доступны для другихконсьюмеров.

• msg2поступает в очередь. Observer1у нас уже занят, и поэтому сообщение на обработку достанется всем свободнымконсьюмерам, в данном случае оно достаетсяObserver2.

Давайте теперь представим, что Observer-service1у нас сломался, например он находится на отдельном сервере и сервер вышел из строя (самый популярный контраргумент - а что... если свет вырубили?).

Ошибка произошла не в самомконсьюмере, и поэтому сообщение, по которому не былоacknowledgementпопадает обратно в очередь и переходит из статусаUnackedвReady. Теперь его может считать любой свободныйконсьюмер. Второй у нас занят, первый умер, свободен только третий и поэтому ему задача и достаётся.

Резонным будет замечание - что будет если ошибка произойдет в самомконсьюмере, в процессе обработки сообщения. Оно в таком случае прейдёт в очередь с пометкой _Error, чтобы этого избежать можем настроитьRetryPolicy. И тогда в случае ошибки сообщение не попадет обратно в очередь, аконсьюмерпопытается заново обработать это сообщение.
Правила дляRetryPolicyмогут быть гибкими:

• Попробовать 1000 раза и положить в очередь ошибок.

• Попробовать 5 раз с интервалом 1,4,10...минут и потом положить в очередь ошибок.

• Вообще попробоватьint.MaxValueраз.

Что же мы имеем в итоге? Мы можем иметь абсолютно любое количество наблюдателей, каждый из которых смотрит на одну очередь и каждый обрабатывает свою задачу/сообщение. Мы можем увеличитьPrefetchCount, допустим до 10, и тогда у сервиса будет 10 свободных консьюмеров, которые будут ждать команды на работу. Сервисы можно распределять по разным серверамиесли мы допускаем что кой-то сервер может выйти из строя, нужно просто иметь свободный сервис, который в случае поломки возьмет задачибольного.Например, если у нас есть 10 серверов, мощностей каждого из которых достаточно для обработки 5 наблюдателей, и шанс того, что один процент из них может выйти из строя, нужнозадеплоитьодин 11-ый сервер с той же мощностью, который будет на подстраховку.

А как жеконсистентность? Да и как вообще всем этим управлять? Да, мы можем добавить сообщения в очередь, но как убрать их оттуда... не очищать же очередь вручную?! Тем более, в идеале, наши "наблюдатели" должны закончить свою жизнь естественным образом, то есть через вызовCancellationToken.

И тут нам снова потребуетсяManager. Менеджеру неплохо бы знать об активных сервисах в системе. Это позволит понимать, перед запуском задачи, сколько свободных сервисов и может ли один из них взять в работу новую задачу. Так же это даст возможность отображать сколько их в системе, на сколько каждый из них загружен и какие задачи обрабатывает. Поэтому, когда сервис только поднимается он отправляет сообщение о своем рождении, которое содержит:

• Id(Идентификатор) -Guidгенерируемый при рождении.

• Name(Имя), которое мы сами дали ему, когда сервис деплоили, оно уникальное для каждого сервиса.

• CreatedAt/ModifyAt(Дата создания/Дата изменения).

• WorkersCount, это будетPrefetchCount- его мощность, сколько он может обрабатывать задач одновременно.

Managerпринимает эти сообщения с делает записи в базу данных о новых активных сервисах.

И нам не важно работает ли менеджер или мы вообще забыли егозадеплоить. В тот момент, когда он заработает ему сразу придет информация о том, что в системе есть 3 сервиса с такими-то параметрами.

Перед отключением сервисы так же отправят сообщения о том, что они закончили работать и большенедоступны, менеджербудет знать, что теперь на N сервисов у него меньше в строю. Тем самым он сделает пометку в базе данных и проставит каждому удаленному значениеIsDeleted=true.

Есть вероятность с тем, что сервис может не успеть отправить свое последнее сообщение о прекращении работы (Kill9, все тот же свет вырубили). За работоспособность компонентов у нас должна отвечать инфраструктура, напримерDocker. Мы должны быть уверены, если сервис непредвиденно прекратил работу, контейнер пере поднимется и сервис заново начнет работу. В таком случае при рождении, он заново отправит сообщение, но уже с новым идентификатором, но старым именем. Менеджеру достаточно будет данной информации чтобы привести данные в консистентное состояние и понять, что со старым сервисом случилось что-то страшное.

А теперь попробуем создать нового наблюдателя через API. Отправляем команду на создание (Мы должны позаботиться о том, что менеджер в процессе инициализации прочитает все сообщения из Statequeue и будет содержать последние актуальные данные о состоянии сервисов). Менеджер проверяет есть ли наблюдатель с таким именем уже, если нет, он проверяет наличие свободных сервисов, а они пока все свободны, далее он дает команду на создание - кладет сообщение в очередь.

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

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

Менеджер, дождавшись ответа от сервиса, которому досталась задача, изменяет статус наProcessingи связывает задачу с сервисом.

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

Перечень статусов:

• Created

• Processing

• OnDeleting

• Deleted

Разберем теперь как удалить "наблюдателя", тут можно пойти двумя путями:

1) направить конкретному сервису сообщение о том, что нужно найти у себя наблюдателя с указанным идентификатором и вызватьCancellationToken.

2) Направить сообщение всем доступным сервисам, черезFanOut.Сервис,у которого есть наблюдатель с нужным идентификатором будет удален, а все остальные сервисы просто проигнорируют это сообщение.

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

КаждыйObserver-serviceимеет свою очередь, где он ожидает получить команду на остановку работы наблюдателя. Когда приходит такая команда, сервис проверяет наличие такого наблюдателя и в случае обнаружения вызываетCancellationToken. Тем самым завершая работу наблюдателя естественным путем.

Пошагово завершение работы Наблюдателя выглядит следующим образом. Пользователь отправляет команду на завершение работы, для этого ему нужно знать лишьidнаблюдателя. Менеджер проверят если ли такой наблюдатель в системе, и его статус.

• ЕслиCreated, пользователю возвращается ответ что наблюдатель еще не активирован. Из-за гонки условий сообщений об удалении может прийти раньше, чем сервису придет сообщение о запуске наблюдателя.

• ЕслиOnDeletingилиDeleted, то возвращается ответ - запрос на удаление уже был отправлен или наблюдатель удален, соответственно.

• ЕслиProcessing, то менеджер переводит наблюдателя в статусOnDeletingи отправляет сообщения на удаление в очередь. Сообщениеброадкаститсявсем сервисам. Сервис, у которого был нужный наблюдатель, вызываетCancellationTokenи оправляет сообщение в statequeue. Менеджер же, получив данное сообщение актуализирует данные и делает пометку переводя изOnDeletingвDeleted.

Рассмотрим критичные сценарии:

1) Отказала шина данных.

Вся инфраструктура, будь то шина данных или база данных должна находится опосредованно от нашей системы и бытькластеризированной. От себя добавлю следующий тезис, который как бритва Оккама отсечет ряд критичных сценариев -MsSql,RabbitMq,Kafka, дажеKubernetesсюда можно добавить, все это надежные системы, и при соблюдении SLA будут работать без отказа. За спиной у них огромные компании или комьюнити, сотни разработчиков. А вот собственную систему нужно воспринимать как что-то ненадежное, где любой компонент в любой момент времени может выйти из строя.

2) Полныйblackout, везде нет света.

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

3) Вылетел один сервер.

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

4) Отказал менеджер. В процессе того пока он был неактивен, ломались сервера с "наблюдателями.

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

5) Попытка удалить конкретного наблюдателя, в том момент, пока он перераспределяется.

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

Итог

Мы реализовали оркестратор задач, на базе механизма отправки сообщений. Где сообщение это задача, с двумя статусами, в работе -Unacked, и в ожидании работы -Ready. Очередь сама распределяет задачи между исполнителями, делая это событийно, а не черезpollingсостояния, как это делают планировщики. Система масштабируемая - мы можем иметь неограниченно количество "наблюдателей, которые могут быть распределены на разных серверах. Более того мы можем масштабировать как горизонтально, так и вертикально, увеличивая количество одновременно обрабатываемых сервисом задач, просто увеличиваяPrefetchCount. И последнее, время на разработку оказалось меньше, чем время на изучение и внедрение планировщика.