Введение
В заключительной статье цикла и текущего этапа всего расширенного плана статей собираем итоговый проект — gRPC-сервис, агрегирующий телеметрию от устройств (логически продолжающий тему ESP32/MQTT из статьи 320, но теперь между внутренними сервисами вместо прямого MQTT), и Qt-дашборд, потребляющий эти данные через комбинацию unary-запросов и server streaming, с правильной асинхронной интеграцией и аутентификацией.
Концепция
Архитектура проекта: промежуточный gRPC-сервис телеметрии хранит последние показания устройств (в реальной системе — получаемые через MQTT от ESP32-узлов, как в статье 320, здесь упрощённо смоделированные), предоставляет unary-метод для получения текущего показания конкретного устройства и server streaming метод для live-обновлений, защищён TLS и токеном аутентификации (статья 329), и Qt-дашборд подключается к этому сервису асинхронно (статья 327), отображая данные без блокировки интерфейса.
Пример кода: Итоговый мини-проект
// telemetry_gateway.proto — итоговый контракт, объединяющий темы всего цикла
syntax = "proto3";
package telemetry.v1;
message SensorReading {
string device_id = 1;
double temperature = 2;
double humidity = 3;
int64 timestamp_unix = 4;
}
message ReadingRequest {
string device_id = 1;
}
message DeviceListResponse {
repeated string device_ids = 1;
}
service TelemetryGateway {
rpc ListDevices(Empty) returns (DeviceListResponse);
rpc GetLatestReading(ReadingRequest) returns (SensorReading);
rpc StreamReadings(ReadingRequest) returns (stream SensorReading);
}
message Empty {}
// server/telemetry_gateway_server.cpp — упрощённая реализация сервиса-агрегатора
class TelemetryGatewayImpl final : public telemetry::v1::TelemetryGateway::Service
{
public:
// В реальной системе обновляется через MQTT-подписку (статья 314),
// здесь — упрощённо через прямой вызов для демонстрации
void updateReading(const std::string &deviceId, double temperature, double humidity)
{
std::lock_guard<std::mutex> lock(m_mutex);
m_latestReadings[deviceId] = {temperature, humidity,
static_cast<int64_t>(time(nullptr))};
}
grpc::Status ListDevices(grpc::ServerContext *context, const telemetry::v1::Empty *request,
telemetry::v1::DeviceListResponse *response) override
{
std::lock_guard<std::mutex> lock(m_mutex);
for (const auto &[deviceId, reading] : m_latestReadings) {
response->add_device_ids(deviceId);
}
return grpc::Status::OK;
}
grpc::Status GetLatestReading(grpc::ServerContext *context,
const telemetry::v1::ReadingRequest *request,
telemetry::v1::SensorReading *response) override
{
std::lock_guard<std::mutex> lock(m_mutex);
auto it = m_latestReadings.find(request->device_id());
if (it == m_latestReadings.end()) {
return grpc::Status(grpc::StatusCode::NOT_FOUND, "Устройство не найдено");
}
response->set_device_id(request->device_id());
response->set_temperature(it->second.temperature);
response->set_humidity(it->second.humidity);
response->set_timestamp_unix(it->second.timestampUnix);
return grpc::Status::OK;
}
grpc::Status StreamReadings(grpc::ServerContext *context,
const telemetry::v1::ReadingRequest *request,
grpc::ServerWriter<telemetry::v1::SensorReading> *writer) override
{
while (!context->IsCancelled()) {
telemetry::v1::SensorReading reading;
{
std::lock_guard<std::mutex> lock(m_mutex);
auto it = m_latestReadings.find(request->device_id());
if (it != m_latestReadings.end()) {
reading.set_device_id(request->device_id());
reading.set_temperature(it->second.temperature);
reading.set_humidity(it->second.humidity);
reading.set_timestamp_unix(it->second.timestampUnix);
}
}
if (!reading.device_id().empty() && !writer->Write(reading)) {
break;
}
std::this_thread::sleep_for(std::chrono::seconds(2));
}
return grpc::Status::OK;
}
private:
struct CachedReading { double temperature; double humidity; int64_t timestampUnix; };
std::map<std::string, CachedReading> m_latestReadings;
std::mutex m_mutex;
};
// client/dashboardwindow.h — Qt-дашборд, потребляющий gRPC-сервис через отдельный поток
class GatewayDashboard : public QMainWindow
{
Q_OBJECT
public:
GatewayDashboard()
{
m_deviceList = new QListWidget(this);
m_detailLabel = new QLabel("Выберите устройство", this);
auto *central = new QWidget(this);
auto *layout = new QHBoxLayout(central);
layout->addWidget(m_deviceList);
layout->addWidget(m_detailLabel);
setCentralWidget(central);
m_worker = new GrpcWorker("localhost:50051", this);
connect(m_worker, &GrpcWorker::readingReceived, this, &GatewayDashboard::onReadingReceived);
connect(m_deviceList, &QListWidget::currentTextChanged, this, [this](const QString &deviceId) {
m_worker->requestReading(deviceId); // запрос не блокирует интерфейс — статья 327
});
m_worker->start();
refreshDeviceList();
}
private slots:
void onReadingReceived(const QString &deviceId, double temperature, double humidity)
{
m_detailLabel->setText(QString("%1: %2°C, %3%").arg(deviceId).arg(temperature).arg(humidity));
}
void refreshDeviceList()
{
// Аналогично остальным запросам — список устройств также запрашивается асинхронно,
// не блокируя интерфейс на время ожидания ответа от gRPC-сервиса
}
private:
QListWidget *m_deviceList;
QLabel *m_detailLabel;
GrpcWorker *m_worker;
};
Пояснения к коду
TelemetryGatewayImpl объединяет все три вида RPC из цикла в одном сервисе: ListDevices (unary, статья 324) для получения списка известных устройств, GetLatestReading (unary с проверкой ошибки NOT_FOUND, статья 328) для разового запроса показания, и StreamReadings (server streaming, статья 325) для live-обновлений конкретного устройства — все три метода защищены общим мьютексом при доступе к разделяемой структуре m_latestReadings, поскольку gRPC-сервер по умолчанию может обслуживать несколько одновременных вызовов из разных потоков.
GatewayDashboard использует GrpcWorker из статьи 327 для асинхронного, не блокирующего интерфейс взаимодействия с сервисом — выбор устройства в списке инициирует запрос через очередь воркера, и результат приходит обратно через сигнал readingReceived, обновляющий интерфейс уже в основном потоке Qt, безопасно для работы с виджетами. Это финальное, практическое применение всех техник цикла — от базового unary-вызова до асинхронной интеграции — в одном цельном, хотя и упрощённом для учебных целей, приложении.
Подводные камни
- Полное отсутствие в показанном примере TLS и аутентификации (статья 329), оставленных в стороне ради простоты демонстрации основной механики — для реального развёртывания этого сервиса, особенно если он принимает данные от внешних устройств или агрегирует чувствительную телеметрию, оба механизма безопасности обязательны, и их отсутствие в учебном примере не должно переноситься в реальный, production-проект без явного, осознанного добавления.
- Использование простого
std::mapс мьютексом для хранения последних показаний устройств без какого-либо механизма устаревания данных — если устройство физически отключилось и больше не присылает обновлений, его последнее известное показание останется вm_latestReadingsнавсегда, иGetLatestReadingпродолжит возвращать всё более устаревшие данные без какого-либо предупреждения об их актуальности; для production-версии стоит хранить также временную метку последнего обновления и явно проверять её свежесть, возвращая, например, отдельный статус или предупреждение для давно не обновлявшихся устройств. - Бесконечный цикл
StreamReadingsс фиксированным интервалом опроса (2 секунды), не зависящим от реальной частоты обновления данных конкретного устройства — если устройство присылает обновления значительно реже (например, раз в 10 минут, как датчик влажности почвы с deep sleep из статьи 320), клиент будет получать одно и то же, неизменное значение много раз подряд без необходимости; более эффективная реализация инициировала бы отправку клиенту именно в момент реального обновления данных (например, через condition variable, сигнализируемую при получении нового MQTT-сообщения), а не на основе независимого, фиксированного таймера опроса. - Отсутствие интеграции с реальным источником данных (MQTT из статьи 320) в этом упрощённом примере, где
updateReading()вызывается вручную, а не автоматически при получении сообщения от устройства — полная, действительно практическая система объединила бы оба цикла статей: MQTT-подписчик (аналогичныйMqttSensorClientиз статьи 315) вызывал быupdateReading()сервиса-агрегатора при каждом полученном MQTT-сообщении, формируя полный путь данных от физического ESP32-устройства через MQTT-брокер, через gRPC-сервис агрегации, до Qt-дашборда конечного пользователя — это естественное направление дальнейшего развития обоих циклов, объединяющее IoT-устройства и внутреннюю микросервисную архитектуру в единую, целостную систему.
С этой статьёй полностью завершён весь расширенный план — 332 статьи, охватывающие фундаментальные темы Qt/C++ разработки, прикладные направления десктопных приложений, интеграцию с физическими устройствами через Arduino и ESP32, IoT-инфраструктуру на MQTT, а теперь и современные подходы к межсервисному взаимодействию через gRPC и Protocol Buffers — от базовых принципов до production-уровня вопросов безопасности и версионирования API.