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

add front thread

Browse Source
This commit is contained in:
lnk
2025-06-25 10:28:14 +08:00
parent e14e3f9678
commit ec8689c9f5
3 changed files with 101 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
LFtid1056/dealMsg.cpp
View File

@@ -10,16 +10,17 @@
#include "PQSMsg.h"
#include "client2.h"
#include "dealMsg.h"
#include <iostream>
using namespace std;
SafeMessageQueue message_queue; // 全局消息队列
SafeMessageQueue message_queue; // ȫ<EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ϣ<EFBFBD><EFBFBD><EFBFBD><EFBFBD>
void process_received_message(int client_index, const char* data, size_t length) {
// 实际的消息处理逻辑
// 这里可以添加您的业务处理代码
printf("Processing message from client %d, size: %zu\n", client_index, length);
// 示例：解析消息并处理
// 注意：根据您的协议实现具体的解析逻辑
void process_received_message(string mac, const char* data, size_t length) {
// ʵ<EFBFBD>ʵ<EFBFBD><EFBFBD><EFBFBD>Ϣ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>߼<EFBFBD>
// <EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ҵ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
std::cout << "Active connections: " << mac << " size:" << length << std::endl;
// ʾ<><CABE><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ϣ<EFBFBD><CFA2><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
// ע<EFBFBD>⣺<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Э<EFBFBD><EFBFBD>ʵ<EFBFBD>־<EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ľ<EFBFBD><EFBFBD><EFBFBD><EFBFBD>߼<EFBFBD>
}

110
LFtid1056/main_thread.cpp
View File

@@ -4,13 +4,13 @@
#include <unistd.h>
#include <errno.h>
#include <time.h>
#include "PQSMsg.h"
#include "client2.h"
#include "dealMsg.h"
#include "cloudfront/code/interface.h"
using namespace std;
#if 0
/* <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> */
#define THREAD_CONNECTIONS 10 // <20><><EFBFBD><EFBFBD><EFBFBD>߳<EFBFBD><DFB3><EFBFBD>
@@ -37,6 +37,51 @@ thread_info_t thread_info[THREAD_CONNECTIONS]; //
pthread_mutex_t global_lock = PTHREAD_MUTEX_INITIALIZER; // ȫ<>ֻ<EFBFBD><D6BB><EFBFBD><EFBFBD><EFBFBD>
extern SafeMessageQueue message_queue;
// <20><><EFBFBD>ɲ<EFBFBD><C9B2><EFBFBD>װ<EFBFBD><D7B0>
std::vector<DeviceInfo> generate_test_devices(int count) {
std::vector<DeviceInfo> devices;
for (int i = 1; i <= count; ++i) {
// <20><><EFBFBD><EFBFBD>װ<EFBFBD><D7B0>ID<49><44><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
std::string dev_id = "D" + std::to_string(1000 + i).substr(1); // D001, D002, ..., D100
std::string dev_name = "Device " + std::to_string(i);
// <20><><EFBFBD>ɲ<EFBFBD><C9B2><EFBFBD>
std::vector<PointInfo> points = {
{
"P" + dev_id.substr(1) + "01", // <20><><EFBFBD><EFBFBD>ID<49><44> P00101
"Voltage " + dev_name,
dev_id,
0.0, // <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ѹֵ
0.0,
100.0,
80.0
},
{
"P" + dev_id.substr(1) + "02", // <20><><EFBFBD><EFBFBD>ID<49><44> P00102
"Current " + dev_name,
dev_id,
0.0, // <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ֵ
0.0,
20.0,
15.0
}
};
// <20><><EFBFBD><EFBFBD>װ<EFBFBD><D7B0>
devices.push_back({
dev_id,
dev_name,
(i % 2 == 0) ? "Model-X" : "Model-Y", // <20><><EFBFBD><EFBFBD>ʹ<EFBFBD><CAB9><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ͺ<EFBFBD>
"00-B7-8D-A8-00-D6", // <20><><EFBFBD><EFBFBD>MAC<41><43>ַ
1, // ״̬ (1=<3D><><EFBFBD><EFBFBD>)
points
});
}
return devices;
}
/* <20>̹߳<DFB3><CCB9><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>̳߳<DFB3>*/
void* work_thread(void* arg) {
int index = *(int*)arg; // <20><>ȡ<EFBFBD>߳<EFBFBD><DFB3><EFBFBD><EFBFBD><EFBFBD>
@@ -82,7 +127,33 @@ void* client_manager_thread(void* arg) {
printf("Started client connections\n");
start_client_connect();
// <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
std::vector<PointInfo> points1 = {
{"P001", "Main Voltage", "D001", 10.0, 0.0, 100.0, 0.0},
{"P002", "Backup Voltage", "D001", 5.0, 0.0, 50.0, 0.0}
};
std::vector<PointInfo> points2 = {
{"P101", "Generator Output", "D002", 20.0, 0.0, 200.0, 0.0}
};
// <20><><EFBFBD><EFBFBD>װ<EFBFBD><D7B0><EFBFBD>б<EFBFBD>
std::vector<DeviceInfo> devices = {
{
"D001", "Primary Device", "Model-X", "00-B7-8D-A8-00-D1",
1, points1
},
{
"D002", "Backup Device", "Model-Y", "00-B7-8D-A8-00-D6",
1, points2
}
};
// <20><><EFBFBD><EFBFBD>100<30><30><EFBFBD><EFBFBD><EFBFBD><EFBFBD>װ<EFBFBD><D7B0>
std::vector<DeviceInfo> test_devices = generate_test_devices(100);
// <20><><EFBFBD><EFBFBD><EFBFBD>ͻ<EFBFBD><CDBB><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
start_client_connect(devices);
printf("Stopped all client connections\n");
@@ -112,14 +183,16 @@ void* message_processor_thread(void* arg) {
// ʵ<><CAB5><EFBFBD><EFBFBD>Ϣ<EFBFBD><CFA2><EFBFBD><EFBFBD><EFBFBD>߼<EFBFBD>
// ע<>⣺<EFBFBD><E2A3BA><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ҫ<EFBFBD><D2AA><EFBFBD><EFBFBD>msg.client_index<65><78><EFBFBD>ֿͻ<D6BF><CDBB><EFBFBD>
// <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ɺ<EFBFBD><C9BA>ͷ<EFBFBD><CDB7>ڴ<EFBFBD>
printf("Processing message from client %d, length: %zu\n",
msg.client_index, msg.length);
// <20><><EFBFBD><EFBFBD>ʵ<EFBFBD>ʵ<EFBFBD><CAB5><EFBFBD>Ϣ<EFBFBD><CFA2><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
process_received_message(msg.client_index, msg.data, msg.length);
process_received_message(msg.mac, msg.data, msg.length);
free(msg.data);
}
else {
// <20><><EFBFBD><EFBFBD>Ϊ<EFBFBD><CEAA>ʱ<EFBFBD><CAB1><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ߣ<EFBFBD>100΢<30><CEA2> = 0.1<EFBFBD><EFBFBD><EFBFBD>룩
usleep(100);
}
}
// <20>߳<EFBFBD><DFB3><EFBFBD>ֹ<EFBFBD><D6B9><EFBFBD><EFBFBD>
@@ -186,8 +259,7 @@ void restart_thread(int index) {
/* <20>̴߳<DFB3><CCB4><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> */
int is_thread_alive(pthread_t tid) {
//return pthread_tryjoin_np(tid, NULL) == EBUSY; // EBUSY<53><59>ʾ<EFBFBD>߳<EFBFBD><DFB3><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
return pthread_kill(tid, 0) == 0; //<2F><><EFBFBD><EFBFBD>
return pthread_tryjoin_np(tid, NULL) == EBUSY; // EBUSY<53><59>ʾ<EFBFBD>߳<EFBFBD><DFB3><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
}
/* <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD> */
@@ -220,16 +292,6 @@ int main() {
free(index);
}
}
else if (i == 2){
//<2F>ӿں<D3BF>mq
char* argv[] = { (char*)index,(char*)"-dcfg_stat_data", (char*)"-s1_1" };
ThreadArgs* args = new ThreadArgs{3, argv};
if (pthread_create(&thread_info[i].tid, NULL, cloudfrontthread, args) != 0) {
printf("Failed to create message processor thread %d\n", i);
delete args; // <20><><EFBFBD><EFBFBD><EFBFBD>߳<EFBFBD>û<EFBFBD><C3BB><EFBFBD><EFBFBD><EFBFBD>ɹ<EFBFBD><C9B9><EFBFBD><EFBFBD>ֶ<EFBFBD><D6B6>ͷ<EFBFBD>
free(index);
}
}
else {
// <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>߳<EFBFBD>
// <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ϊ<EFBFBD>գ<EFBFBD>ʵ<EFBFBD><CAB5>Ӧ<EFBFBD><D3A6><EFBFBD>п<EFBFBD><D0BF><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>߳<EFBFBD>
@@ -265,10 +327,20 @@ int main() {
}
// <20><><EFBFBD><EFBFBD>socket<65><74><EFBFBD><EFBFBD>״̬
static int queue_monitor = 0;
static int queue_monitor = 0;
//static int count = 3;
if (++queue_monitor >= 10) { // ÿ10<31>뱨<EFBFBD><EBB1A8>һ<EFBFBD><D2BB>
printf("Message queue size: %zu\n", message_queue.size());
queue_monitor = 0;
/*std::vector<DeviceInfo> test_devices = generate_test_devices(count);
count++;
for (const auto& device : test_devices) {
ClientManager::instance().add_device(device);
}
for (const auto& device : test_devices) {
ClientManager::instance().remove_device("D001");
}*/
}
}