Введение
Во всех предыдущих статьях цикла gRPC-вызовы выполнялись синхронно, что многократно отмечалось как проблема для приложения с пользовательским интерфейсом. В статье разберём правильную интеграцию — выполнение gRPC-вызовов без блокировки основного потока Qt, используя как простой подход через отдельный поток, так и более продвинутый асинхронный API gRPC (CompletionQueue).
Концепция
Самый простой способ избежать блокировки UI — выполнять синхронные gRPC-вызовы в отдельном потоке (QThread, аналогично паттерну из основного цикла статей про потоки), передавая результат обратно в основной поток через сигналы, что concептуально идентично паттерну WorkerThread из статьи 224. Более продвинутый, но и более сложный подход — использование собственного асинхронного API gRPC через CompletionQueue, которая может быть интегрирована непосредственно с циклом обработки событий Qt через периодический опрос в отдельном потоке или через интеграцию с QSocketNotifier, что избегает создания отдельного потока на каждый отдельный вызов.
Пример кода: Интеграция асинхронного gRPC-клиента
// grpcworker.h — простой подход: выполнение gRPC-вызовов в отдельном потоке
#pragma once
#include <QThread>
#include <QObject>
#include "telemetryclient.h"
class GrpcWorker : public QThread
{
Q_OBJECT
public:
explicit GrpcWorker(const QString &serverAddress, QObject *parent = nullptr)
: QThread(parent), m_serverAddress(serverAddress) {}
void requestReading(const QString &deviceId)
{
QMutexLocker locker(&m_mutex);
m_pendingRequests.enqueue(deviceId);
m_waitCondition.wakeOne();
}
protected:
void run() override
{
TelemetryClient client(m_serverAddress.toStdString());
while (!isInterruptionRequested()) {
QString deviceId;
{
QMutexLocker locker(&m_mutex);
if (m_pendingRequests.isEmpty()) {
m_waitCondition.wait(&m_mutex, 500); // периодически просыпаемся для проверки interruption
continue;
}
deviceId = m_pendingRequests.dequeue();
}
auto result = client.getLatestReading(deviceId); // блокирующий вызов — но НЕ в основном потоке
if (result) {
emit readingReceived(deviceId, result->temperature, result->humidity);
} else {
emit requestFailed(deviceId);
}
}
}
signals:
void readingReceived(const QString &deviceId, double temperature, double humidity);
void requestFailed(const QString &deviceId);
private:
QString m_serverAddress;
QQueue<QString> m_pendingRequests;
QMutex m_mutex;
QWaitCondition m_waitCondition;
};
// Использование в основном потоке Qt-приложения — без блокировки интерфейса
auto *worker = new GrpcWorker("localhost:50051", this);
connect(worker, &GrpcWorker::readingReceived, this, [this](const QString &deviceId, double temp, double hum) {
qDebug() << "Получено показание для" << deviceId << ":" << temp << "°C," << hum << "%";
});
worker->start();
connect(refreshButton, &QPushButton::clicked, this, [worker]() {
worker->requestReading("esp32_livingroom_01"); // не блокирует интерфейс — запрос ставится в очередь
});
// Более продвинутый подход: асинхронный gRPC через CompletionQueue,
// интегрированный с Qt через периодический опрос таймером
// (концептуальная схема — полная реализация требует более глубокой работы с grpc::CompletionQueue)
class AsyncGrpcClient : public QObject
{
Q_OBJECT
public:
AsyncGrpcClient(const std::string &serverAddress, QObject *parent = nullptr) : QObject(parent)
{
auto channel = grpc::CreateChannel(serverAddress, grpc::InsecureChannelCredentials());
m_stub = telemetry::TelemetryService::NewStub(channel);
// Таймер периодически проверяет очередь завершённых асинхронных вызовов,
// не блокируя основной поток Qt длительным ожиданием
connect(&m_pollTimer, &QTimer::timeout, this, &AsyncGrpcClient::pollCompletionQueue);
m_pollTimer.start(10); // частый, но не блокирующий опрос — каждые 10 мс
}
void requestReadingAsync(const QString &deviceId)
{
auto *call = new AsyncCall();
call->request.set_device_id(deviceId.toStdString());
call->responseReader = m_stub->AsyncGetLatestReading(&call->context, call->request, &m_queue);
call->responseReader->Finish(&call->response, &call->status, (void *)call);
}
private slots:
void pollCompletionQueue()
{
void *tag;
bool ok;
// AsyncNext с нулевым тайм-аутом — проверяем готовые результаты НЕМЕДЛЕННО,
// не блокируясь, если ничего не готово (в отличие от блокирующего Next())
auto deadline = gpr_time_0(GPR_CLOCK_REALTIME);
while (m_queue.AsyncNext(&tag, &ok, deadline) == grpc::CompletionQueue::GOT_EVENT) {
auto *call = static_cast<AsyncCall *>(tag);
if (ok && call->status.ok()) {
emit readingReceived(QString::fromStdString(call->response.device_id()),
call->response.temperature());
}
delete call;
}
}
signals:
void readingReceived(const QString &deviceId, double temperature);
private:
struct AsyncCall {
telemetry::ReadingRequest request;
telemetry::SensorReading response;
grpc::ClientContext context;
grpc::Status status;
std::unique_ptr<grpc::ClientAsyncResponseReader<telemetry::SensorReading>> responseReader;
};
std::unique_ptr<telemetry::TelemetryService::Stub> m_stub;
grpc::CompletionQueue m_queue;
QTimer m_pollTimer;
};
Пояснения к коду
GrpcWorker реализует паттерн, концептуально идентичный WorkerThread из статьи 224 основного Arduino-цикла — отдельный поток с собственным циклом обработки запросов через очередь (QQueue), защищённую мьютексом, и QWaitCondition для эффективного ожидания новых запросов без активного опроса (busy-waiting) — каждый блокирующий gRPC-вызов выполняется именно в этом отдельном потоке, а результат передаётся обратно в основной поток через обычные Qt-сигналы, что Qt автоматически делает потокобезопасно через очередь событий.
AsyncGrpcClient показывает более продвинутый подход — собственный асинхронный API gRPC (AsyncGetLatestReading, CompletionQueue) интегрируется с циклом событий Qt не через отдельный, постоянно блокирующийся поток, а через QTimer, периодически (каждые 10 мс) проверяющий, появились ли готовые результаты в очереди завершения, используя неблокирующий AsyncNext() с нулевым дедлайном — это позволяет обслуживать множество одновременных асинхронных вызовов через единственный поток (сам основной поток Qt-приложения), без накладных расходов на создание отдельного потока для каждого вызова, хотя за счёт заметно более сложной реализации по сравнению с простым подходом через GrpcWorker.
Подводные камни
- Выбор между простым (отдельный поток) и сложным (CompletionQueue) подходом без реальной оценки потребностей конкретного приложения — для приложений с небольшим количеством редких gRPC-вызовов простой подход через
QThreadзначительно проще в реализации и отладке, и переход наCompletionQueueоправдан прежде всего при необходимости обслуживать большое количество одновременных, параллельных вызовов, где накладные расходы на множество отдельных потоков стали бы заметной проблемой. - Утечка памяти при работе с
CompletionQueue, если объектAsyncCall(выделенный черезnew) не освобождается во всех возможных путях выполнения — в примереdelete callпроисходит после обработки события, но любой дополнительный, незапланированный путь завершения (например, отмена вызова) требует столь же аккуратного управления временем жизни этих объектов, что существенно сложнее, чем работа с обычными, управляемыми через RAII объектами C++. - Слишком частый опрос
CompletionQueueчерезQTimerс очень коротким интервалом создаёт постоянную, не бесплатную нагрузку на CPU даже когда реальных событий нет, а слишком редкий опрос увеличивает задержку между фактическим получением ответа от сервера и его обработкой в Qt-приложении — выбор интервала опроса (10 мс в примере) — компромисс, который стоит подбирать с учётом реальных требований к задержке конкретного приложения. - Забытая остановка потока
GrpcWorker(или таймера опросаAsyncGrpcClient) при закрытии приложения — без явного, корректного завершения (requestInterruption()плюсwait()дляQThread, или остановкаQTimer) поток/таймер может продолжать попытки выполнять gRPC-вызовы во время процесса завершения приложения, что может привести к зависанию процесса завершения или ошибкам при попытке обратиться к уже частично разрушенным объектам приложения.