#include "CVrZeroMQClient.h" #include "VrError.h" #include "VrLog.h" #include "IVrUtils.h" #include "VrStringUtils.h" // 低延迟配置常量 namespace { const int LOW_LATENCY_LINGER = 0; const int LOW_LATENCY_HWM = 1; const int LOW_LATENCY_BUFFER_SIZE = 4096; // 减小缓冲区以减少延迟 const int LOW_LATENCY_CONNECT_TIMEOUT = 500; // 减少连接超时 const int LOW_LATENCY_RECONNECT_INTERVAL = 50; // 更快的重连 const int LOW_LATENCY_RECV_TIMEOUT = 1000; // 大幅减少接收超时 } bool IVrZeroMQClient::CreateObject(IVrZeroMQClient** ppMQClient) { *ppMQClient = new CVrZeroMQClient(); return true; } CVrZeroMQClient::CVrZeroMQClient() { m_pContext = new zmq::context_t; m_pSocket = new zmq::socket_t(*m_pContext, ZMQ_REQ); // 应用低延迟优化配置 m_pSocket->set(zmq::sockopt::linger, LOW_LATENCY_LINGER); m_pSocket->set(zmq::sockopt::rcvhwm, LOW_LATENCY_HWM); m_pSocket->set(zmq::sockopt::sndhwm, LOW_LATENCY_HWM); // TCP_NODELAY在此ZeroMQ版本中不可用,跳过此选项 m_pSocket->set(zmq::sockopt::rcvbuf, LOW_LATENCY_BUFFER_SIZE); m_pSocket->set(zmq::sockopt::sndbuf, LOW_LATENCY_BUFFER_SIZE); // 跨机器时连接可能超过 500ms,这里放宽到 3000ms m_pSocket->set(zmq::sockopt::connect_timeout, 3000); // ★ 接收超时 3 秒:防止服务端无响应时永久卡死 UI 线程 m_pSocket->set(zmq::sockopt::rcvtimeo, 3000); m_pSocket->set(zmq::sockopt::reconnect_ivl, LOW_LATENCY_RECONNECT_INTERVAL); // 设置更短的重连间隔以快速恢复 int reconnect_ivl_max = 200; // 最大重连间隔200ms m_pSocket->set(zmq::sockopt::reconnect_ivl_max, reconnect_ivl_max); } CVrZeroMQClient::~CVrZeroMQClient() { try { if (m_pSocket) m_pSocket->close(); } catch (...) {} delete m_pSocket; delete m_pContext; } int CVrZeroMQClient::Init(const char *sIP, int nPort) { try{ std::string transferAddr = "tcp://"; transferAddr.append(sIP); transferAddr.append(":"); transferAddr.append(CVrStringUtils::Int2Str(nPort)); m_pSocket->connect(transferAddr); }catch(const std::exception& e){ LOG_ERROR("Init zmq client error : %s \n", e.what()); return ERR_CODE(NET_ERR_CONNECT); } return SUCCESS; } // 发送数据 int CVrZeroMQClient::SendAndWaitBack(const char* pdata, const int nLen, char** ppRecvData, size_t& nRecvLen) { std::lock_guard lck(m_mutexObj); zmq::message_t msg(pdata, nLen); // 立即发送 LOG_DEBUG("[ZMQ-REQ] send: %.*s\n", nLen, pdata); auto send_result = m_pSocket->send(msg, zmq::send_flags::none); if (!send_result.has_value()) { LOG_DEBUG("[ZMQ-REQ] send FAIL\n"); LOG_ERROR("Failed to send message\n"); return ERR_CODE(NET_ERR_SEND_DATA); } // 直接阻塞接收 LOG_DEBUG("[ZMQ-REQ] waiting for recv...\n"); zmq::recv_result_t ret = m_pSocket->recv(m_recvmsg, zmq::recv_flags::none); if (ret.has_value()) { nRecvLen = m_recvmsg.size(); *ppRecvData = (char *)m_recvmsg.data(); LOG_DEBUG("[ZMQ-REQ] recv OK: %.*s\n", (int)nRecvLen, *ppRecvData); return SUCCESS; } LOG_DEBUG("[ZMQ-REQ] recv FAIL\n"); LOG_ERROR("Failed to receive response\n"); return ERR_CODE(NET_ERR_RECV_DATA); } int CVrZeroMQClient::UnInit() { try { if (m_pSocket) m_pSocket->close(); } catch (...) {} delete m_pSocket; delete m_pContext; m_pContext = nullptr; m_pSocket = nullptr; return SUCCESS; }