Queue: проектная работа по курсу «Разработчик C++»

На одном из занятий курса «Разработчик C++» мы изучали пул потоков. Ключевым элементом этого шаблона была общая очередь, которая была доступна всем потокам.

Потоки можно поделить на две группы: 1) первая группа потоков, которая могла писать в очередь, формируя таким образом задания для остальных; 2) вторая группа потоков, которая могла читать из этой очереди.

В таком виде наше решение могло быть использовано только на одной машине. Наша задача — позволить распределить потоки, работающие с этой очередью, на разные машины.

Можно сразу ввести новые понятия. То, что создаёт задание в очереди и фактически пишет в очередь называется «Продюсер». То, что читает из очереди, называется «Консьюмер».

Фактически наша задача сводится к написанию сервера, который будет обеспечивать по сети доступ к методам push_back() для продюсеров и front() / pop() для консьюмеров класса std::queue.

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

О клиентской библиотеке

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

Что такое клиентская библиотека? Клиентская библиотека — это, прежде всего, код с нужными классами и функциями, а уж есть там .so или .a уже не имеет значения.

Грубо говоря, если мне хочется поработать с вашей очередью, я должен иметь понятный сценарий её подключения, например, должен написать #include <mgafka> и слинковаться с mgafka, для подключения к серверу — auto h = MgafkaConnection(“localhost”, 6519) и далее по API.

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

Как хранятся данные?

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

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

Когда и кем создаются очереди

Можно рассмотреть вариант с автоматическим созданием в момент первого подключения консьюмера или продюсера.

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

Прочие подробности реализации проекта

Критерий отсечения должен быть понятный: на диске он может быть примитивным — держит заполненным какой-то объём на диске (при превышении удаляем старое). Держать в памяти всю очередь необязательно — достаточно иметь бегущее окно — самый частый сценарий будет чтение свежих записей. Окна можно подгружать по мере необходимости, если подключится клиент, запросивший совсем старые данные.

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

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

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

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