zhangwen/front_linux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修改了socket通讯框架，添加了统计数据时间获取，统计数据数据读取

Browse Source
This commit is contained in:
zw
2025-07-03 11:13:16 +08:00
parent 7fdc689f4e
commit 50b21bcd3e
7 changed files with 1761 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

508
LFtid1056/client2.cpp
View File

@@ -1,6 +1,4 @@
#include "client2.h"
#include "PQSMsg.h"
#include "dealMsg.h"
#include <iostream>
#include <cmath>
#include <cstring>
@@ -11,7 +9,7 @@
#include <unordered_map>
// 配置参数
constexpr int BASE_RECONNECT_DELAY = 5000; // 基础重连延迟(ms)
constexpr int BASE_RECONNECT_DELAY = 20000; // 基础重连延迟(ms)
constexpr int MAX_RECONNECT_DELAY = 60000; // 最大重连延迟(ms)
constexpr const char* SERVER_IP = "101.132.39.45"; // 目标服务器IP"101.132.39.45"
constexpr int SERVER_PORT = 1056; // 目标服务器端口1056
@@ -23,7 +21,8 @@ extern SafeMessageQueue message_queue;
// ClientContext 实现
ClientContext::ClientContext(uv_loop_t* loop, const DeviceInfo& device, int index)
: loop(loop), state(ConnectionState::DISCONNECTED),
reconnect_attempts(0), shutdown(false), device_info(device), index_(index) {
reconnect_attempts(0), shutdown(false), device_info(device), index_(index),cloudstatus(0), current_state_(DeviceState::IDLE),
state_start_time_(0) {
recv_buffer_.reserve(4096); // 预分配4KB缓冲区
@@ -54,7 +53,7 @@ void ClientContext::init_tcp() {
void ClientContext::start_timer() {
if (!uv_is_active((uv_handle_t*)&timer)) {
uv_timer_start(&timer, on_timer, 6000, 6000);
uv_timer_start(&timer, on_timer, 5000,5000);
}
}
@@ -187,6 +186,132 @@ void ClientContext::put_packet_into_queue(
}
}
// 新增方法：改变装置状态
void ClientContext::change_state(DeviceState new_state, const std::vector<unsigned char>& packet) {
std::lock_guard<std::mutex> lock(state_mutex_);
// 直接更新状态，不调用其他锁方法
current_state_ = new_state;
current_packet_ = packet;
state_start_time_ = uv_now(loop);
std::cout << "[Device " << device_info.device_id
<< "] State changed to: " << static_cast<int>(new_state) << std::endl;
}
// 新增方法：添加后续动作
void ClientContext::add_action(DeviceState state, const std::vector<unsigned char>& packet) {
std::lock_guard<std::mutex> lock(state_mutex_);
action_queue_.push({ state, packet });
std::cout << "[Device " << device_info.device_id
<< "] Action added to queue: " << static_cast<int>(state) << std::endl;
}
// 新增方法：处理状态超时
void ClientContext::check_state_timeout() {
constexpr uint64_t STATE_TIMEOUT = 30000;//30秒超时
uint64_t now = uv_now(loop);
bool timed_out = false;
{
std::lock_guard<std::mutex> lock(state_mutex_);
if (current_state_ != DeviceState::IDLE &&
(now - state_start_time_) > STATE_TIMEOUT)
{
timed_out = true;
current_state_ = DeviceState::IDLE;
}
}
if (timed_out) {
process_next_action(); // 在锁外调用
}
}
// 新增方法：处理下一个动作
void ClientContext::process_next_action() {
StateAction next;
{
std::lock_guard<std::mutex> lock(state_mutex_);
if (current_state_ != DeviceState::IDLE || action_queue_.empty())
return;
next = action_queue_.front();
action_queue_.pop();
} // 提前释放锁
// 在锁外调用可能阻塞的函数
change_state(next.state, next.packet);
send_current_packet();
}
// 新增方法：发送当前状态对应的报文
void ClientContext::send_current_packet() {
if (!current_packet_.empty()) {
send_binary_data(this, current_packet_.data(), current_packet_.size());
}
}
bool ClientContext::add_stat_packet(const std::vector<unsigned char>& packet, int current_packet, int total_packets) {
std::lock_guard<std::mutex> lock(stat_cache_mutex_);
// 如果是第一帧，初始化缓存
if (current_packet == 1) {
stat_packets_cache_.clear();
expected_total_packets_ = total_packets;
}
// 添加到缓存
stat_packets_cache_.push_back({ current_packet, packet });
// 检查是否收齐所有帧
return (stat_packets_cache_.size() >= expected_total_packets_);
}
std::vector<ClientContext::StatPacket> ClientContext::get_and_clear_stat_packets() {
std::lock_guard<std::mutex> lock(stat_cache_mutex_);
auto packets = std::move(stat_packets_cache_);
stat_packets_cache_.clear();
expected_total_packets_ = 0;
return packets;
}
void ClientContext::clear_stat_cache() {
std::lock_guard<std::mutex> lock(stat_cache_mutex_);
stat_packets_cache_.clear();
expected_total_packets_ = 0;
}
// 添加浮点数据到缓存
bool ClientContext::add_float_data(ushort point_id, int data_type, const tagPqData_Float& float_data) {
if (data_type < 0 || data_type > 3) return false;
std::lock_guard<std::mutex> lock(float_cache_mutex_);
auto& cache = point_float_cache_[point_id];
cache.data[data_type] = float_data;
cache.received[data_type] = true;
// 检查是否四种数据类型都已接收
return cache.received[0] && cache.received[1] &&
cache.received[2] && cache.received[3];
}
// 获取并清除指定测点的完整浮点数据
std::array<tagPqData_Float, 4> ClientContext::get_and_clear_float_data(ushort point_id) {
std::lock_guard<std::mutex> lock(float_cache_mutex_);
auto it = point_float_cache_.find(point_id);
if (it == point_float_cache_.end()) {
return {};
}
auto data = it->second.data;
point_float_cache_.erase(it);
return data;
}
// 清除所有浮点数据缓存
void ClientContext::clear_float_cache() {
std::lock_guard<std::mutex> lock(float_cache_mutex_);
point_float_cache_.clear();
}
/* 缓冲区分配回调 */
void alloc_buffer(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
buf->base = new char[suggested_size];
@@ -230,24 +355,33 @@ void on_write(uv_write_t* req, int status) {
}
/* 定时发送回调 */
//5秒执行一次定时器
void on_timer(uv_timer_t* handle) {
ClientContext* ctx = static_cast<ClientContext*>(handle->data);
if (ctx->state != ConnectionState::CONNECTED) {
return;
}
std::cout << "on_timer: " << ctx->device_info.mac << " send!"<< std::endl;
// 使用装置自己的MAC地址生成登录报文
auto binary_data = generate_frontlogin_message(ctx->device_info.mac);
// 调用发送函数
send_binary_data(ctx, binary_data.data(), binary_data.size());
static int statequerytime = 0;//询问统计数据时间标志 20秒执行一次
//ClientManager::instance().send_to_device(ctx->device_info.mac, binary_data.data(), binary_data.size());
// 检查状态超时 30秒状态未更新则重置为空闲状态
ctx->check_state_timeout();
// 根据装置状态发送其他数据
if (ctx->device_info.status == 1) { // 在线状态
// 可以发送装置配置信息或测点数据
// 装置登录成功后，只在空闲状态处理后续动作
if (ctx->cloudstatus == 1) {
//20秒一次 执行统计数据时间询问
if (++statequerytime >= 4 && ctx->current_state_ == DeviceState::IDLE) {
statequerytime = 0;//重置计时
auto sendbuff = generate_statequerytime_message();//组装询问统计数据时间报文
ctx->add_action(DeviceState::READING_STATS_TIME, sendbuff);//将该状态以及待发送报文存入队列
}
//处理后续工作队列的工作 取出一个并执行
if (ctx->current_state_ == DeviceState::IDLE) {
ctx->process_next_action();
}
}
}
@@ -283,6 +417,18 @@ void on_close(uv_handle_t* handle) {
std::cerr << "[Device " << ctx->device_info.device_id << "] Connection closed" << std::endl;
ctx->stop_timers();
// 清空缓存
ctx->clear_stat_cache();
// 清除浮点数据缓存
ctx->clear_float_cache();
ctx->cloudstatus = 0;
{
std::lock_guard<std::mutex> state_lock(ctx->state_mutex_);
ctx->current_state_ = DeviceState::IDLE; // 直接修改状态
std::queue<StateAction> empty;
std::swap(ctx->action_queue_, empty);
}
if (!ctx->shutdown) {
int delay = BASE_RECONNECT_DELAY * pow(2, ctx->reconnect_attempts);
@@ -344,6 +490,11 @@ void on_connect(uv_connect_t* req, int status) {
ctx->state = ConnectionState::CONNECTED;
ctx->reconnect_attempts = 0;
//客户端连接完毕后，发送装置登陆消息
std::cout << "connected: " << ctx->device_info.mac << " send login msg!" << std::endl;
auto binary_data = generate_frontlogin_message(ctx->device_info.mac);
send_binary_data(ctx, binary_data.data(), binary_data.size());
uv_read_start((uv_stream_t*)&ctx->client, alloc_buffer, on_read);
ctx->start_timer();
}
@@ -493,4 +644,333 @@ void ClientManager::stop_all() {
pair.second->close_handles();
}
clients_.clear();
}
// 在ClientManager成员函数实现中添加方法实现
void ClientManager::restart_device(const std::string& device_id) {
std::lock_guard<std::mutex> lock(mutex_);
ClientContext* target_ctx = nullptr;
// 查找匹配的设备（支持device_id或mac地址）
for (auto& pair : clients_) {
auto& ctx = pair.second;
if (ctx->device_info.device_id == device_id ||
ctx->device_info.mac == device_id) {
target_ctx = ctx.get();
break;
}
}
if (!target_ctx) {
std::cerr << "[restart_device] Device not found: " << device_id << std::endl;
return;
}
std::cout << "[restart_device] Restarting device: " << device_id << std::endl;
// 确保不处于关闭状态
target_ctx->shutdown = false;
// 停止所有定时器
target_ctx->stop_timers();
// 重置重连计数器
target_ctx->reconnect_attempts = 0;
// 关闭TCP连接（会触发on_close回调）
if (!uv_is_closing((uv_handle_t*)&target_ctx->client)) {
uv_close((uv_handle_t*)&target_ctx->client, on_close);
}
else {
// 如果已经在关闭过程中，直接触发重连
target_ctx->state = ConnectionState::DISCONNECTED;
target_ctx->start_reconnect_timer(0); // 立即重连
}
}
//修改客户端云前置登录状态
bool ClientManager::set_cloud_status(const std::string& identifier, int status) {
std::lock_guard<std::mutex> lock(mutex_);
for (auto& pair : clients_) {
auto& ctx = pair.second;
// 匹配装置ID或MAC地址
if (ctx->device_info.device_id == identifier ||
ctx->device_info.mac == identifier) {
// 修改云前置登录状态
ctx->cloudstatus = status;
std::cout << "[Device " << identifier
<< "] Cloud status updated to: " << status << std::endl;
return true;
}
}
std::cerr << "[set_cloud_status] Device not found: " << identifier << std::endl;
return false;
}
bool ClientManager::add_action_to_device(const std::string& identifier,
DeviceState state,
const std::vector<unsigned char>& packet) {
std::lock_guard<std::mutex> lock(mutex_);
for (auto& pair : clients_) {
auto& ctx = pair.second;
if (ctx->device_info.device_id == identifier ||
ctx->device_info.mac == identifier) {
ctx->add_action(state, packet);
return true;
}
}
std::cerr << "[add_action_to_device] Device not found: " << identifier << std::endl;
return false;
}
bool ClientManager::change_device_state(const std::string& identifier,
DeviceState new_state,
const std::vector<unsigned char>& packet) {
std::lock_guard<std::mutex> lock(mutex_);
for (auto& pair : clients_) {
auto& ctx = pair.second;
if (ctx->device_info.device_id == identifier ||
ctx->device_info.mac == identifier) {
ctx->change_state(new_state, packet);
return true;
}
}
std::cerr << "[change_device_state] Device not found: " << identifier << std::endl;
return false;
}
bool ClientManager::clear_action_queue(const std::string& identifier) {
std::lock_guard<std::mutex> lock(mutex_);
for (auto& pair : clients_) {
auto& ctx = pair.second;
if (ctx->device_info.device_id == identifier ||
ctx->device_info.mac == identifier) {
std::lock_guard<std::mutex> state_lock(ctx->state_mutex_);
std::queue<StateAction> empty;
std::swap(ctx->action_queue_, empty);
return true;
}
}
std::cerr << "[clear_action_queue] Device not found: " << identifier << std::endl;
return false;
}
bool ClientManager::get_device_state(const std::string& identifier, DeviceState& out_state) {
std::lock_guard<std::mutex> lock(mutex_);
for (auto& pair : clients_) {
auto& ctx = pair.second;
if (ctx->device_info.device_id == identifier ||
ctx->device_info.mac == identifier) {
std::lock_guard<std::mutex> state_lock(ctx->state_mutex_);
out_state = ctx->current_state_;
return true;
}
}
std::cerr << "[get_device_state] Device not found: " << identifier << std::endl;
return false;
}
bool ClientManager::post_message_processing(const std::string& identifier) {
ClientContext* target = nullptr;
{
std::lock_guard<std::mutex> lock(mutex_);
for (auto& pair : clients_) {
auto& ctx = pair.second;
if (ctx->device_info.device_id == identifier ||
ctx->device_info.mac == identifier) {
target = pair.second.get();
break;
}
}
} // 提前释放manager锁
if (!target) {
std::cerr << "Device not found: " << identifier << std::endl;
return false;
}
// 直接操作client，避免嵌套锁
if (target->current_state_ == DeviceState::IDLE) {
//空闲状态执行下一项工作
target->process_next_action();
return true;
}
else {
//非空闲状态执行当前工作
target->send_current_packet();
return true;
}
}
//通过id或者mac读取装置下属测点信息
bool ClientManager::get_device_points(const std::string& identifier,
std::vector<PointInfo>& out_points) {
std::lock_guard<std::mutex> lock(mutex_);
for (const auto& pair : clients_) {
const auto& ctx = pair.second;
// 匹配装置ID或MAC地址
if (ctx->device_info.device_id == identifier ||
ctx->device_info.mac == identifier) {
// 复制测点信息到输出参数
out_points = ctx->device_info.points;
return true;
}
}
std::cerr << "[get_device_points] Device not found: " << identifier << std::endl;
return false;
}
//保存多帧报文至缓存区等待收全
bool ClientManager::add_stat_packet_to_device(const std::string& identifier,
const std::vector<unsigned char>& packet,
int current_packet,
int total_packets) {
std::lock_guard<std::mutex> lock(mutex_);
for (auto& pair : clients_) {
auto& ctx = pair.second;
if (ctx->device_info.device_id == identifier ||
ctx->device_info.mac == identifier) {
return ctx->add_stat_packet(packet, current_packet, total_packets);
}
}
std::cerr << "[add_stat_packet_to_device] Device not found: " << identifier << std::endl;
return false;
}
//获取缓存区内所有多帧报文并清空缓存
std::vector<ClientContext::StatPacket> ClientManager::get_and_clear_stat_packets(const std::string& identifier) {
std::lock_guard<std::mutex> lock(mutex_);
for (auto& pair : clients_) {
auto& ctx = pair.second;
if (ctx->device_info.device_id == identifier ||
ctx->device_info.mac == identifier) {
return ctx->get_and_clear_stat_packets();
}
}
std::cerr << "[get_and_clear_stat_packets] Device not found: " << identifier << std::endl;
return {};
}
//清空所有缓存区
bool ClientManager::clear_stat_cache(const std::string& identifier) {
std::lock_guard<std::mutex> lock(mutex_);
for (auto& pair : clients_) {
auto& ctx = pair.second;
if (ctx->device_info.device_id == identifier ||
ctx->device_info.mac == identifier) {
ctx->clear_stat_cache();
return true;
}
}
std::cerr << "[clear_stat_cache] Device not found: " << identifier << std::endl;
return false;
}
// 获取pt和CT变比
bool ClientManager::get_pt_ct_ratio(const std::string& identifier,
int16_t nCpuNo,
float& pt_ratio,
float& ct_ratio) {
std::lock_guard<std::mutex> lock(mutex_);
for (auto& pair : clients_) {
auto& ctx = pair.second;
// 匹配装置ID或MAC地址
if (ctx->device_info.device_id == identifier ||
ctx->device_info.mac == identifier) {
// 遍历装置的所有测点
for (const auto& point : ctx->device_info.points) {
// 匹配测点序号
if (point.nCpuNo == nCpuNo) {
// 计算PT变比 (PT1/PT2)
pt_ratio = (point.PT2 != 0.0) ?
static_cast<float>(point.PT1 / point.PT2) : 1.0f;
// 计算CT变比 (CT1/CT2)
ct_ratio = (point.CT2 != 0.0) ?
static_cast<float>(point.CT1 / point.CT2) : 1.0f;
return true;
}
}
std::cerr << "[get_pt_ct_ratio] Point not found for CPU: "
<< nCpuNo << " in device: " << identifier << std::endl;
return false;
}
}
std::cerr << "[get_pt_ct_ratio] Device not found: " << identifier << std::endl;
return false;
}
// 添加浮点数据到指定设备的缓存
bool ClientManager::add_float_data_to_device(const std::string& identifier,
ushort point_id,
int data_type,
const tagPqData_Float& float_data) {
std::lock_guard<std::mutex> lock(mutex_);
for (auto& pair : clients_) {
auto& ctx = pair.second;
if (ctx->device_info.device_id == identifier ||
ctx->device_info.mac == identifier) {
return ctx->add_float_data(point_id, data_type, float_data);
}
}
return false;
}
// 获取并清除指定测点的完整浮点数据
std::array<tagPqData_Float, 4> ClientManager::get_and_clear_float_data(
const std::string& identifier, ushort point_id) {
std::lock_guard<std::mutex> lock(mutex_);
for (auto& pair : clients_) {
auto& ctx = pair.second;
if (ctx->device_info.device_id == identifier ||
ctx->device_info.mac == identifier) {
return ctx->get_and_clear_float_data(point_id);
}
}
return {};
}
// 清除设备的所有浮点缓存
bool ClientManager::clear_float_cache(const std::string& identifier) {
std::lock_guard<std::mutex> lock(mutex_);
for (auto& pair : clients_) {
auto& ctx = pair.second;
if (ctx->device_info.device_id == identifier ||
ctx->device_info.mac == identifier) {
ctx->clear_float_cache();
return true;
}
}
return false;
}