feat:完成对RabbitMQ的自动重连逻辑

This commit is contained in:
duanshengchao 2025-11-07 17:33:04 +08:00
parent b2fb1932c3
commit 0962645144
3 changed files with 60 additions and 8 deletions

View File

@ -10,6 +10,7 @@
class QAmqpClient; class QAmqpClient;
class QAmqpQueue; class QAmqpQueue;
class QAmqpExchange; class QAmqpExchange;
class QTimer;
class AlarmEventDataService : public QObject class AlarmEventDataService : public QObject
{ {
@ -29,6 +30,7 @@ private slots:
void onRabbitMQDisconnected(); void onRabbitMQDisconnected();
void onRabbitMQError(QAMQP::Error error); void onRabbitMQError(QAMQP::Error error);
void onMessageReceived(); void onMessageReceived();
void onReconnectTimeout();
private: private:
explicit AlarmEventDataService(); explicit AlarmEventDataService();
@ -37,6 +39,8 @@ private:
//连接管理 //连接管理
void startRealTimeDataService(); void startRealTimeDataService();
void cleanupRabbitMQConnection(); void cleanupRabbitMQConnection();
void scheduleReconnect();
void cancelReconnect();
//实时信息处理 //实时信息处理
MessageHandleResult processMessage(const QAmqpMessage& message); MessageHandleResult processMessage(const QAmqpMessage& message);
EventData parseEventFromMessage(const QByteArray& data, QString& errorString); EventData parseEventFromMessage(const QByteArray& data, QString& errorString);
@ -52,7 +56,10 @@ private:
QAmqpQueue* m_amqpQueue; QAmqpQueue* m_amqpQueue;
QAmqpExchange* m_amqpExchange; QAmqpExchange* m_amqpExchange;
//重连相关
QTimer* m_reconnectTimer;
int m_reconnectAttempts;
int m_maxReconnectAttempts;
}; };
#endif #endif

View File

@ -22,7 +22,7 @@ struct RabbitMQConfig
RabbitMQConfig() RabbitMQConfig()
: port(5672) : port(5672)
, reconnectInterval(5000) , reconnectInterval(3000)
, heartbeat(60) , heartbeat(60)
, autoAck(true) , autoAck(true)
{} {}
@ -166,7 +166,7 @@ struct ServiceConfig
enum class ServiceStatus enum class ServiceStatus
{ {
Uninitialized,//未初始化 Uninitialized,//未初始化
Initialized, //已初始化 //Initialized, //已初始化
Disconnected, //未连接 Disconnected, //未连接
Connecting, //连接中 Connecting, //连接中
Connected, //已连接 Connected, //已连接

View File

@ -1,4 +1,5 @@
#include "alarmEventDataService.h" #include "alarmEventDataService.h"
#include <QTimer>
#include <QJsonObject> #include <QJsonObject>
#include <QJsonDocument> #include <QJsonDocument>
#include <QJsonArray> #include <QJsonArray>
@ -17,7 +18,13 @@ AlarmEventDataService::AlarmEventDataService()
, m_amqpClient(nullptr) , m_amqpClient(nullptr)
, m_amqpQueue(nullptr) , m_amqpQueue(nullptr)
, m_amqpExchange(nullptr) , m_amqpExchange(nullptr)
{} , m_reconnectAttempts(0)
, m_maxReconnectAttempts(10)
{
m_reconnectTimer = new QTimer(this);
m_reconnectTimer->setSingleShot(true);
connect(m_reconnectTimer, &QTimer::timeout, this, &AlarmEventDataService::onReconnectTimeout);
}
AlarmEventDataService::~AlarmEventDataService() AlarmEventDataService::~AlarmEventDataService()
{} {}
@ -31,21 +38,27 @@ bool AlarmEventDataService::initialize(const ServiceConfig& config)
return false; return false;
m_config = config; m_config = config;
m_serviceStatus = ServiceStatus::Disconnected;
return true; return true;
} }
void AlarmEventDataService::start() void AlarmEventDataService::start()
{ {
if(m_serviceStatus == ServiceStatus::Connected || m_serviceStatus == ServiceStatus::Connecting) if(m_serviceStatus != ServiceStatus::Disconnected)
return; return;
m_serviceStatus = ServiceStatus::Connecting; m_serviceStatus = ServiceStatus::Connecting;
//启动实时数据服务
startRealTimeDataService(); startRealTimeDataService();
} }
void AlarmEventDataService::stop() void AlarmEventDataService::stop()
{} {
cancelReconnect();
cleanupRabbitMQConnection();
m_serviceStatus = ServiceStatus::Disconnected;
}
void AlarmEventDataService::startRealTimeDataService() void AlarmEventDataService::startRealTimeDataService()
{ {
@ -85,6 +98,24 @@ void AlarmEventDataService::cleanupRabbitMQConnection()
} }
} }
void AlarmEventDataService::scheduleReconnect()
{
if (m_reconnectAttempts < m_maxReconnectAttempts)
{
int delay = m_config.rabbitMQConfig.reconnectInterval * (1 << m_reconnectAttempts); // 指数退避,<<n相当于乘以2的n次方
m_reconnectTimer->start(delay);
m_reconnectAttempts++;
}
else
qInfo() << "Maximum reconnection attempts reached";
}
void AlarmEventDataService::cancelReconnect()
{
m_reconnectTimer->stop();
m_reconnectAttempts = 0;
}
MessageHandleResult AlarmEventDataService::processMessage(const QAmqpMessage& message) MessageHandleResult AlarmEventDataService::processMessage(const QAmqpMessage& message)
{ {
QString errorString; QString errorString;
@ -154,12 +185,17 @@ void AlarmEventDataService::onRabbitMQConnected()
} }
void AlarmEventDataService::onRabbitMQDisconnected() void AlarmEventDataService::onRabbitMQDisconnected()
{} {
qWarning() << "Disconnected to RabbitMQ";
m_serviceStatus = ServiceStatus::Disconnected;
scheduleReconnect();
}
void AlarmEventDataService::onRabbitMQError(QAMQP::Error error) void AlarmEventDataService::onRabbitMQError(QAMQP::Error error)
{ {
qWarning() << m_amqpClient->errorString(); qWarning() << m_amqpClient->errorString();
m_serviceStatus = ServiceStatus::Error; m_serviceStatus = ServiceStatus::Error;
scheduleReconnect();
} }
void AlarmEventDataService::onMessageReceived() void AlarmEventDataService::onMessageReceived()
@ -171,3 +207,12 @@ void AlarmEventDataService::onMessageReceived()
if(!m_config.rabbitMQConfig.autoAck && result == MessageHandleResult::Success) if(!m_config.rabbitMQConfig.autoAck && result == MessageHandleResult::Success)
m_amqpQueue->ack(message); m_amqpQueue->ack(message);
} }
void AlarmEventDataService::onReconnectTimeout()
{
if(m_serviceStatus == ServiceStatus::Connected || m_serviceStatus == ServiceStatus::Reconnecting)
return;
m_serviceStatus = ServiceStatus::Reconnecting;
startRealTimeDataService();
}