Подкинули задачу сделать микросервис, который получает данные от RabbitMQ, обрабатывает, и отправляет данные дальше по этапу в RabbitMQ. После отправки задания, я посмотрел на то что поучилось. Оказалось, что этот набор компонентов можно использовать для быстрого прототипирования pipeline архитектуры
Используемые компоненты:
Для примера буду делать микросервис для выдачи рейтинга игроков. От ядра системы в микросервис приходят следующие сообщения:
- player_registered(id,name);
- player_renamed(id,name);
- player_won(id, points).
Сервис раз в минуту должен отсылать сообщение с содержимым рейтинга.Рейтинг сортируется по набранным очкам за календарную неделю.
REACT-CPP
REACT-CPP — это обертка над libev на C++11. Эта библиотека нужна для организации цикла обработка событий(event loop). Т.к. кроме работы с сокетом потребуются таймеры и обработчики unix сигналов.
class Application
{
public:
Application();
~Application();
using IntervalWatcherPtr = std::shared_ptr<React::IntervalWatcher>;
void run();
void shutdown();
//...
private:
bool onMinute();
//...
private:
React::MainLoop m_loop;
IntervalWatcherPtr m_minuteTimer;
//...
};void Application::run()
{
m_minuteTimer = m_loop.onInterval(5.0, 60.0, std::bind(&Application::onMinute, this));
m_loop.onSignal(SIGTERM, [this]() -> bool
{
shutdown();
return false;
});
m_loop.onSignal(SIGUSR1, [this]()->bool{
cleanRating();
return true;
});
//...
m_loop.run();
}
bool Application::onMinute()
{
calculateRating();
sendRating();
return true;
}Тут создаю таймер который стартует через 5 секунд и который будет вызывать обработчик каждые 60 секунд. Любой приличный демон/сервис должен иметь обработчик SIGTERM, что бы из вне попросить его корректно завершится. Что касается обработчика SIGUSR1 тут можно самостоятельно вычислять начало/конец недели через Boost.Date_Time, но мне тупо лень, когда в GNU/Linux есть cron+pkill.
AMQP-CPP
С тех пор как опубликовал RabbitMQ tutorials на C++ AMQP-CPP обзавелась реализацией обработчика на libev и libuv.
Подключение и обработка сообщения:
void Application::createChannel(AMQP::TcpConnection &connection)
{
m_channel = std::make_unique<AMQP::TcpChannel>(&connection);
m_channel->declareQueue(m_cfg.source().name, AMQP::durable)
.onSuccess([&](const std::string &name, uint32_t messagecount, uint32_t consumercount)
{
LOG(INFO) << "Declared queue "
<< name
<< ", message count: "
<< messagecount;
m_channel->consume(m_cfg.source().name)
.onReceived([&](const AMQP::Message &message,
uint64_t deliveryTag,
bool redelivered)
{
onMessage(message, deliveryTag, redelivered);
})
.onError([](const char *message)
{
LOG(ERROR) << "Error consume:" << message;
APP->shutdown();
});
})
.onError([&](const char *message)
{
LOG(ERROR) << "Error declare queue:" << message;
shutdown();
});
}
void Application::onMessage(const AMQP::Message &message,
uint64_t deliveryTag,
bool redelivered)
{
parseMessage(message);
m_channel->ack(deliveryTag);
}Публикация сообщения:
AMQP::Envelope env(s.GetString());
m_channel->publish("", m_cfg.destination().name, env);LevelDB
Может потребоваться локальное хранилище данных. Взял LelevDB, я о нем писал в Использование LevelDB. Сделал лишь небольшую RAII обертку:
class DataBase
{
public:
DataBase();
bool open(const std::string &path2base, bool compression = true);
bool put(const std::string &key, const ByteArray &value, bool sync = false);
ByteArray get(const std::string &key);
Snapshot snapshot();
Iterator iterator();
private:
std::shared_ptr<leveldb::DB> m_backend;
};
class Snapshot
{
public:
Snapshot();
~Snapshot();
ByteArray get(const std::string &key);
Iterator iterator();
private:
Snapshot(const std::weak_ptr<leveldb::DB> &backend, const leveldb::Snapshot *snapshot);
private:
friend class DataBase;
std::weak_ptr<leveldb::DB> m_backend;
const leveldb::Snapshot *m_shapshot;
};
class Iterator
{
public:
Iterator(std::unique_ptr<leveldb::Iterator> rawIterator);
Iterator(Iterator &&iter);
/*!
* Create empty iterator
*/
Iterator() = default;
~Iterator();
bool isValid() const noexcept;
void next();
void prev();
std::string key();
ByteArray value();
/*!
* Seek to first
*/
void toFirst();
/*!
* Seek to last
*/
void toLast();
Iterator(const Iterator &) = delete;
Iterator &operator=(const Iterator &) = delete;
private:
std::unique_ptr<leveldb::Iterator> m_iterator;
};LevelDB используется для сохранения/востановления состояния.
void Application::loadFromLocalStorage()
{
auto snapshot = m_localStorage->snapshot();
auto iter = snapshot.iterator();
iter.toFirst();
while (iter.isValid()) {
auto player = new Player(iter.value());
m_id2player[player->id] = player;
m_players.push_back(player);
iter.next();
}
}
void Application::updatePlayerInBD(const Player *player)
{
if (!m_localStorage->put(std::to_string(player->id), player->serialize())) {
LOG(ERROR) << "[" << player->id << ", "
<< player->name
<< "] is not updated in the database";
}
}Логика сервиса
Данные приходят в формате JSON. Разбирает json используя RapidJSON, ищу подходящий метод, вызываю нужный обработчик:
void Application::parseMessage(const AMQP::Message &message)
{
/*
* Схемка имеет вид
* {
* "method":"player_registered",
* "params":{
* ...
* }
* }
*/
rapidjson::Document doc;
doc.Parse(message.body(), message.bodySize());
const std::string method = doc["method"].GetString();
auto iter = m_handlers.find(method);
if (iter != m_handlers.end()) {
iter->second(*this, doc["params"]);
}
else {
LOG(WARNING) << "Unknown method:" << method;
}
}Сами методы простые:
void Application::onPlayerRegistered(const JValue ¶ms)
{
auto obj = params.GetObject();
const uint64_t playerId = obj["id"].GetUint64();
if (!isRegistred(playerId)) {
auto player = new Player;
player->id = playerId;
player->name = obj["name"].GetString();
m_players.push_back(player);
m_id2player[playerId] = player;
updatePlayerInBD(player);
}
}
void Application::onPlayerRenamed(const JValue ¶ms)
{
auto obj = params.GetObject();
const uint64_t playerId = obj["id"].GetUint64();
if (isRegistred(playerId)) {
auto player = m_id2player[playerId];
player->name = obj["name"].GetString();
updatePlayerInBD(player);
}
else {
LOG(WARNING) << "Renaming an unknown user[" << playerId << "]";
}
}
void Application::onPlayerWon(const JValue ¶ms)
{
auto obj = params.GetObject();
const uint64_t playerId = obj["id"].GetUint64();
if (isRegistred(playerId)) {
auto player = m_id2player[playerId];
player->points += obj["points"].GetInt64();
updatePlayerInBD(player);
}
else {
LOG(WARNING) << "Unknown player[" << playerId << "]";
}
}Раз в минуту сортируем игроков и отправляем рейтинг:
bool Application::onMinute()
{
calculateRating();
sendRating();
return true;
}
void Application::calculateRating()
{
std::sort(m_players.begin(), m_players.end(), [](const Player *a, const Player *b)
{
return a->points > b->points;
});
}
void Application::sendRating()
{
using namespace rapidjson;
StringBuffer s;
Writer<StringBuffer> writer(s);
writer.StartArray();
const size_t count = std::min(m_players.size(), size_t(10));
for (size_t i = 0;
i < count;
++i) {
writer.StartObject();
writer.Key("id");
writer.Uint64(m_players[i]->id);
writer.Key("name");
writer.String(m_players[i]->name.c_str());
writer.Key("points");
writer.Int64(m_players[i]->points);
writer.EndObject();
}
writer.EndArray();
AMQP::Envelope env(s.GetString());
m_channel->publish("", m_cfg.destination().name, env);
}Весь код доступен на GitHub'e. Исходники библиотек поставляются вместе с сервисом и собираются автоматически на GNU/Linux с gcc.
Подведем итоги, что имеем:
- event loop с таймерами, обработчиками сигналов и всеми остальными плюшками libev;
- работа с RabbitMQ;
- встроенное key-value хранилище;
- поддержка json.
