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-06-19 13:24:45 +08:00
parent 8070b3392e
commit c51da2e506
4 changed files with 123 additions and 301 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
LFtid1056/PQSMsg.cpp
View File

@@ -9,51 +9,6 @@
#include "client2.h" #include "client2.h"
#include "PQSMsg.h" #include "PQSMsg.h"
// <20><><EFBFBD>ɴ<EFBFBD>Э<EFBFBD><D0AD>ͷ<EFBFBD>Ķ<EFBFBD><C4B6><EFBFBD><EFBFBD>Ʊ<EFBFBD><C6B1><EFBFBD>
std::vector<unsigned char> generate_binary_message(
uint16_t msg_type,
const std::vector<unsigned char>& payload)
{
std::vector<unsigned char> packet;
// Э<><D0AD>ͷ (4<>ֽ<EFBFBD>)
packet.push_back(0xAA); // <20><>ʼ<EFBFBD><CABC>־
packet.push_back((msg_type >> 8) & 0xFF); // <20><>Ϣ<EFBFBD><CFA2><EFBFBD>͸<EFBFBD><CDB8>ֽ<EFBFBD>
packet.push_back(msg_type & 0xFF); // <20><>Ϣ<EFBFBD><CFA2><EFBFBD>͵<EFBFBD><CDB5>ֽ<EFBFBD>
uint8_t length = payload.size();
packet.push_back(length); // <20>غɳ<D8BA><C9B3><EFBFBD>
// <20>غ<EFBFBD><D8BA><EFBFBD><EFBFBD><EFBFBD>
packet.insert(packet.end(), payload.begin(), payload.end());
// У<><D0A3>λ (1<>ֽ<EFBFBD><D6BD><EFBFBD><EFBFBD><EFBFBD>У<EFBFBD><D0A3>)
uint8_t checksum = 0;
for (auto byte : packet) {
checksum ^= byte;
}
packet.push_back(checksum);
return packet;
}
// ʹ<><CAB9>ʾ<EFBFBD><CABE>
void send_message(client_context_t* ctx) {
// <20><><EFBFBD><EFBFBD><EFBFBD>غ<EFBFBD><D8BA><EFBFBD><EFBFBD><EFBFBD>
std::vector<unsigned char> payload = {
0x01, 0x02, 0x03, 0x04 // ʾ<><CABE><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
};
// <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> (<28><>Ϣ<EFBFBD><CFA2><EFBFBD><EFBFBD>0x1001)
auto binary_data = generate_binary_message(0x1001, payload);
// ת<><D7AA>Ϊ<EFBFBD><CEAA><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ʽ
unsigned char* binary_array = binary_data.data();
size_t data_size = binary_data.size();
// <20>˴<EFBFBD><CBB4>ɵ<EFBFBD><C9B5>÷<EFBFBD><C3B7>ͺ<EFBFBD><CDBA><EFBFBD>
send_binary_data(ctx, binary_array, data_size);
}
// <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>MAC<41><43>ַ<EFBFBD><D6B7><EFBFBD><EFBFBD><EFBFBD>䵽<EFBFBD><E4B5BD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> // <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>MAC<41><43>ַ<EFBFBD><D6B7><EFBFBD><EFBFBD><EFBFBD>䵽<EFBFBD><E4B5BD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
void GetMAC(const std::string& strMAC, std::vector<unsigned char>& packet, size_t startIndex) { void GetMAC(const std::string& strMAC, std::vector<unsigned char>& packet, size_t startIndex) {
// <20>Ƴ<EFBFBD><C6B3><EFBFBD><EFBFBD>пո<D0BF><D5B8>Ͷ̺<CDB6><CCBA><EFBFBD> // <20>Ƴ<EFBFBD><C6B3><EFBFBD><EFBFBD>пո<D0BF><D5B8>Ͷ̺<CDB6><CCBA><EFBFBD>

361
LFtid1056/client2.cpp
View File

@@ -2,30 +2,26 @@
#include <stdlib.h> #include <stdlib.h>
#include <string.h> #include <string.h>
#include <uv.h> #include <uv.h>
#include <pthread.h>
#include <unistd.h>
#include <math.h> #include <math.h>
#include "PQSMsg.h" #include "PQSMsg.h"
#include "client2.h" #include "client2.h"
// 配置参数 // 配置参数
#define INITIAL_DATA_SIZE 128 // 初始数据0.1KB #define INITIAL_DATA_SIZE 128 // 初始数据0.1KB
#define CONNECTIONS 2 // 并发连接数 设备连接数 #define CONNECTIONS 1000 // 支持1000个并发连接
#define SERVER_IP "101.132.39.45" // 目标服务器IP 阿里云服务器 "101.132.39.45""172.27.208.1" #define SERVER_IP "101.132.39.45" // 目标服务器IP "101.132.39.45"
#define SERVER_PORT 1056 // 目标服务器端口 #define SERVER_PORT 1056 // 目标服务器端口
#define BASE_RECONNECT_DELAY 5000 // 基础重连延迟(ms) #define BASE_RECONNECT_DELAY 5000 // 基础重连延迟(ms)
#define MAX_RECONNECT_DELAY 60000 // 最大重连延迟(ms) #define MAX_RECONNECT_DELAY 60000 // 最大重连延迟(ms)
#define MAX_RECONNECT_ATTEMPTS 10 // 最大重连次数
// 全局变量用于管理客户端线程 static uv_loop_t* global_loop; // 全局事件循环
static pthread_t client_threads[CONNECTIONS]; static client_context_t client_contexts[CONNECTIONS]; // 客户端上下文数组
static client_context_t* client_contexts[CONNECTIONS] = { NULL }; static uv_timer_t monitor_timer; // 连接监控定时器
/* 缓冲区分配回调 */ /* 缓冲区分配回调 */
void alloc_buffer(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) { void alloc_buffer(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
void* buffer = malloc(suggested_size); void* buffer = malloc(suggested_size);
if (!buffer) { if (!buffer) {
fprintf(stderr, "Memory allocation failed\n");
*buf = uv_buf_init(NULL, 0); *buf = uv_buf_init(NULL, 0);
return; return;
} }
@@ -41,18 +37,16 @@ void on_read(uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf) {
fprintf(stdout, "[Client %d] RECV ERROR: %s\n", fprintf(stdout, "[Client %d] RECV ERROR: %s\n",
ctx->index, uv_strerror(nread)); ctx->index, uv_strerror(nread));
} }
else {
fprintf(stdout, "[Client %d] Connection closed by server\n", ctx->index);
}
uv_close((uv_handle_t*)stream, on_close); uv_close((uv_handle_t*)stream, on_close);
free(buf->base); free(buf->base);
return;
} }
else if (nread > 0) {
fprintf(stdout, "[Client %d] RECV %zd bytes data\n", ctx->index, nread);
//数据传入客户端缓冲区并尝试取出合法报文格式的数据
free(buf->base); if (nread > 0) {
//fprintf(stdout, "[Client %d] RECV %zd bytes data\n", ctx->index, nread);
// 数据处理逻辑...
} }
free(buf->base);
} }
/* 数据写入回调 */ /* 数据写入回调 */
@@ -63,25 +57,19 @@ void on_write(uv_write_t* req, int status) {
fprintf(stdout, "[Client %d] SEND ERROR: %s\n", fprintf(stdout, "[Client %d] SEND ERROR: %s\n",
ctx->index, uv_strerror(status)); ctx->index, uv_strerror(status));
} }
else {
fprintf(stdout, "[Client %d] SEND bytes success\n", free(req->data); // 释放发送数据缓冲区
ctx->index); free(req); // 释放写入请求
}
free(req->data);
free(req);
} }
/* 定时组装二进制报文并发送 */ /* 定时发送回调 */
void on_timer(uv_timer_t* handle) { void on_timer(uv_timer_t* handle) {
client_context_t* ctx = (client_context_t*)handle->data; client_context_t* ctx = (client_context_t*)handle->data;
if (ctx->state != STATE_CONNECTED) { if (ctx->state != STATE_CONNECTED) {
fprintf(stdout, "[Client %d] Skip sending: Not connected\n", ctx->index);
return; return;
} }
fprintf(stdout, "[Client %d] Preparing periodic data...\n", ctx->index);
// 生成完整报文 装置云服务登录报文 // 生成完整报文 装置云服务登录报文
auto binary_data = generate_frontlogin_message("00-B7-8D-A8-00-D6"); auto binary_data = generate_frontlogin_message("00-B7-8D-A8-00-D6");
@@ -92,46 +80,6 @@ void on_timer(uv_timer_t* handle) {
// 此处可调用发送函数 // 此处可调用发送函数
send_binary_data(ctx, binary_array, data_size); send_binary_data(ctx, binary_array, data_size);
} }
/* 发送初始数据块 - 修复版 */
void send_initial_data(client_context_t* ctx) {
// 添加状态检查
if (ctx->state != STATE_CONNECTED) {
fprintf(stderr, "[Client %d] Cannot send initial data: not connected\n", ctx->index);
return;
}
// 修正为实际数据大小 (1KB)
const size_t data_size = INITIAL_DATA_SIZE; // 使用实际值替换 INITIAL_DATA_SIZE
char* buffer = (char*)malloc(data_size);
if (!buffer) {
fprintf(stderr, "[Client %d] Failed to allocate initial data buffer\n", ctx->index);
return;
}
memset(buffer, 'A', data_size);
buffer[0] = 0xeb; buffer[1] = 0x90; buffer[2] = 0xff; buffer[15] = 0x16;
fprintf(stdout, "[Client %d] Sending initial %zu bytes data\n", ctx->index, data_size);
uv_buf_t buf = uv_buf_init(buffer, data_size);
uv_write_t* write_req = (uv_write_t*)malloc(sizeof(uv_write_t));
if (!write_req) {
free(buffer);
fprintf(stderr, "[Client %d] Failed to allocate write request\n", ctx->index);
return;
}
write_req->data = buffer;
// 检查uv_write返回值
int ret = uv_write(write_req, (uv_stream_t*)&ctx->client, &buf, 1, on_write);
if (ret < 0) {
fprintf(stderr, "[Client %d] uv_write failed: %s\n",
ctx->index, uv_strerror(ret));
free(buffer);
free(write_req);
}
}
/* 发送二进制报文函数 */ /* 发送二进制报文函数 */
void send_binary_data(client_context_t* ctx, const unsigned char* data, size_t data_size) { void send_binary_data(client_context_t* ctx, const unsigned char* data, size_t data_size) {
if (ctx->state != STATE_CONNECTED) { if (ctx->state != STATE_CONNECTED) {
@@ -146,7 +94,7 @@ void send_binary_data(client_context_t* ctx, const unsigned char* data, size_t d
return; return;
} }
write_req->data = NULL; // 不需要额外数据，因为data已经传入 write_req->data = NULL; // 不需要额外数据，因为data已经传入
fprintf(stdout, "[Client %d] Sending initial %zu bytes data\n", ctx->index, data_size); //fprintf(stdout, "[Client %d] Sending initial %zu bytes data\n", ctx->index, data_size);
int ret = uv_write(write_req, (uv_stream_t*)&ctx->client, &buf, 1, on_write); int ret = uv_write(write_req, (uv_stream_t*)&ctx->client, &buf, 1, on_write);
if (ret < 0) { if (ret < 0) {
@@ -155,62 +103,50 @@ void send_binary_data(client_context_t* ctx, const unsigned char* data, size_t d
} }
// 注意：这里不需要释放data，因为data是由调用者管理的 // 注意：这里不需要释放data，因为data是由调用者管理的
} }
/* 连接关闭回调 */ /* 连接关闭回调 */
void on_close(uv_handle_t* handle) { void on_close(uv_handle_t* handle) {
client_context_t* ctx = (client_context_t*)handle->data; client_context_t* ctx = (client_context_t*)handle->data;
ctx->state = STATE_DISCONNECTED; ctx->state = STATE_DISCONNECTED;
fprintf(stdout, "[Client %d] Connection closed\n", ctx->index);
// 停止数据读写定时器和重连定时器 // 停止定时器
uv_timer_stop(&ctx->timer); uv_timer_stop(&ctx->timer);
uv_timer_stop(&ctx->reconnect_timer); uv_timer_stop(&ctx->reconnect_timer);
// 取消读操作 // 自动重连逻辑
uv_read_stop((uv_stream_t*)&ctx->client);
// 如果不是主动关闭，尝试重连
if (!ctx->shutdown) { if (!ctx->shutdown) {
int delay = BASE_RECONNECT_DELAY * pow(2, ctx->reconnect_attempts); int delay = BASE_RECONNECT_DELAY * pow(2, ctx->reconnect_attempts);
delay = delay > MAX_RECONNECT_DELAY ? MAX_RECONNECT_DELAY : delay; delay = delay > MAX_RECONNECT_DELAY ? MAX_RECONNECT_DELAY : delay;
fprintf(stdout, "[Client %d] Will reconnect after %dms (attempt %d/%d)\n", fprintf(stdout, "[Client %d] Reconnecting in %dms (attempt %d)\n",
ctx->index, delay, ctx->reconnect_attempts + 1, MAX_RECONNECT_ATTEMPTS); ctx->index, delay, ctx->reconnect_attempts + 1);
uv_timer_start(&ctx->reconnect_timer, (uv_timer_cb)try_reconnect, delay, 0);
ctx->reconnect_attempts++; ctx->reconnect_attempts++;
uv_timer_start(&ctx->reconnect_timer, try_reconnect, delay, 0);
} }
} }
/* 尝试重连函数 */ /* 尝试重连 */
void try_reconnect(uv_timer_t* timer) { void try_reconnect(uv_timer_t* timer) {
client_context_t* ctx = (client_context_t*)timer->data; client_context_t* ctx = (client_context_t*)timer->data;
if (ctx->state != STATE_DISCONNECTED || ctx->shutdown) { if (ctx->state != STATE_DISCONNECTED || ctx->shutdown) {
fprintf(stdout, "[Client %d] Reconnect skipped: Invalid state or shutdown\n", ctx->index);
return;
}
fprintf(stdout, "[Client %d] Attempting to reconnect...\n", ctx->index);
if (ctx->reconnect_attempts >= MAX_RECONNECT_ATTEMPTS) {
fprintf(stderr, "[Client %d] Max reconnect attempts reached\n", ctx->index);
return; return;
} }
// 重新初始化TCP句柄 // 重新初始化TCP句柄
uv_tcp_init(ctx->loop, &ctx->client); uv_tcp_init(ctx->loop, &ctx->client);
ctx->client.data = ctx; ctx->client.data = ctx;
ctx->state = STATE_CONNECTING; ctx->state = STATE_CONNECTING;
struct sockaddr_in addr; struct sockaddr_in addr;
uv_ip4_addr(SERVER_IP, SERVER_PORT, &addr); uv_ip4_addr(SERVER_IP, SERVER_PORT, &addr);
uv_connect_t* req = (uv_connect_t*)malloc(sizeof(uv_connect_t)); uv_connect_t* req = (uv_connect_t*)malloc(sizeof(uv_connect_t));
req->data = ctx; req->data = ctx;
int ret = uv_tcp_connect(req, &ctx->client, (const struct sockaddr*)&addr, on_connect); int ret = uv_tcp_connect(req, &ctx->client, (const struct sockaddr*)&addr, on_connect);
if (ret < 0) { if (ret < 0) {
fprintf(stderr, "[Client %d] Connect error: %s\n", ctx->index, uv_strerror(ret)); fprintf(stderr, "[Client %d] Connect error: %s\n", ctx->index, uv_strerror(ret));
ctx->state = STATE_DISCONNECTED;
free(req); free(req);
uv_close((uv_handle_t*)&ctx->client, on_close); uv_close((uv_handle_t*)&ctx->client, on_close);
} }
@@ -220,183 +156,126 @@ void try_reconnect(uv_timer_t* timer) {
void on_connect(uv_connect_t* req, int status) { void on_connect(uv_connect_t* req, int status) {
client_context_t* ctx = (client_context_t*)req->data; client_context_t* ctx = (client_context_t*)req->data;
if (ctx->state != STATE_CONNECTING) {
fprintf(stdout, "[Client %d] Connection callback with invalid state\n", ctx->index);
free(req);
return;
}
if (status < 0) { if (status < 0) {
fprintf(stderr, "[Client %d] Connect failed: %s\n", ctx->index, uv_strerror(status)); fprintf(stderr, "[Client %d] Connect failed: %s\n", ctx->index, uv_strerror(status));
ctx->state = STATE_DISCONNECTED; // 直接关闭句柄，避免后续重复关闭
if (!uv_is_closing((uv_handle_t*)&ctx->client)) {
uv_close((uv_handle_t*)&ctx->client, NULL);
}
free(req); free(req);
uv_close((uv_handle_t*)&ctx->client, on_close);
return; return;
} }
else {
ctx->state = STATE_CONNECTED;
ctx->reconnect_attempts = 0;
fprintf(stdout, "[Client %d] Connection established\n", ctx->index);
// 启动数据收发 ctx->state = STATE_CONNECTED;
uv_read_start((uv_stream_t*)&ctx->client, alloc_buffer, on_read); ctx->reconnect_attempts = 0;
//send_initial_data(ctx);
// 启动定时器 // 启动数据接收
//uv_timer_start(&ctx->timer, on_timer, 6000, 6000); uv_read_start((uv_stream_t*)&ctx->client, alloc_buffer, on_read);
}
// 启动定时发送
uv_timer_start(&ctx->timer, on_timer, 6000, 6000);
free(req); free(req);
} }
/* 客户端线程函数 */ /* 初始化所有客户端连接 */
void* run_client(void* arg) { void init_clients(uv_loop_t* loop) {
int index = *(int*)arg; for (int i = 0; i < CONNECTIONS; i++) {
free(arg); client_context_t* ctx = &client_contexts[i];
memset(ctx, 0, sizeof(client_context_t));
// 初始化事件循环 ctx->loop = loop;
uv_loop_t* loop = uv_loop_new(); ctx->index = i;
client_context_t* ctx = (client_context_t*)malloc(sizeof(client_context_t)); ctx->state = STATE_DISCONNECTED;
memset(ctx, 0, sizeof(client_context_t));
// 初始化上下文 // 初始化TCP句柄
ctx->loop = loop; uv_tcp_init(loop, &ctx->client);
ctx->index = index; ctx->client.data = ctx;
ctx->state = STATE_DISCONNECTED;
ctx->reconnect_attempts = 0;
ctx->shutdown = 0; // 初始化关闭标志
// 初始化libuv句柄 // 初始化定时器
uv_tcp_init(loop, &ctx->client); uv_timer_init(loop, &ctx->timer);
uv_timer_init(loop, &ctx->timer); ctx->timer.data = ctx;
uv_timer_init(loop, &ctx->reconnect_timer);
// 设置关联数据 // 初始化重连定时器
ctx->client.data = ctx; uv_timer_init(loop, &ctx->reconnect_timer);
ctx->timer.data = ctx; ctx->reconnect_timer.data = ctx;
ctx->reconnect_timer.data = ctx;
// 首次连接尝试 // 首次连接
try_reconnect(&ctx->reconnect_timer); try_reconnect(&ctx->reconnect_timer);
// 运行事件循环，定期检查关闭标志
while (uv_loop_alive(loop) && !ctx->shutdown) {
uv_run(loop, UV_RUN_NOWAIT);
usleep(10000); // 10ms
} }
fprintf(stdout, "[Client %d] thread closed\n", ctx->index);
// 主动关闭所有资源
if (ctx->shutdown) {
uv_close((uv_handle_t*)&ctx->client, on_close);
uv_close((uv_handle_t*)&ctx->timer, NULL);
uv_close((uv_handle_t*)&ctx->reconnect_timer, NULL);
// 确保所有关闭回调执行
while (uv_loop_alive(loop)) {
uv_run(loop, UV_RUN_ONCE);
}
}
// 资源清理
uv_loop_close(loop);
uv_loop_delete(loop);
free(ctx);
return NULL;
} }
/* 停止所有客户端线程 (新增函数) */ /* 停止所有客户端 */
void stop_all_clients() { void stop_all_clients() {
for (int i = 0; i < CONNECTIONS; i++) { for (int i = 0; i < CONNECTIONS; i++) {
if (client_contexts[i]) { client_context_t* ctx = &client_contexts[i];
client_contexts[i]->shutdown = 1; // 设置关闭标志 ctx->shutdown = 1;
}
}
// 等待所有客户端线程退出 // 关闭所有句柄
for (int i = 0; i < CONNECTIONS; i++) { if (!uv_is_closing((uv_handle_t*)&ctx->client)) {
if (client_threads[i]) { uv_close((uv_handle_t*)&ctx->client, NULL);
pthread_join(client_threads[i], NULL); }
client_threads[i] = 0; if (!uv_is_closing((uv_handle_t*)&ctx->timer)) {
uv_close((uv_handle_t*)&ctx->timer, NULL);
}
if (!uv_is_closing((uv_handle_t*)&ctx->reconnect_timer)) {
uv_close((uv_handle_t*)&ctx->reconnect_timer, NULL);
} }
} }
} }
/* 连接监控回调 */
void monitor_connections(uv_timer_t* handle) {
static int shutdown_flag = 0; // 新增关闭标志
if (shutdown_flag) return; // 已关闭则不再处理
/* 启动所有客户端线程 (新增函数) */ // 自动恢复断开的连接
void start_all_clients() { static int recovery_counter = 0;
for (int i = 0; i < CONNECTIONS; i++) { if (++recovery_counter >= 5) { // 每5次监控执行一次恢复
int* index = (int*)malloc(sizeof(int)); int active_count = 0;
*index = i; for (int i = 0; i < CONNECTIONS; i++) {
if (client_contexts[i].state == STATE_CONNECTED) {
if (pthread_create(&client_threads[i], NULL, run_client, index) != 0) { active_count++;
fprintf(stderr, "Failed to create client thread %d\n", i); }
free(index);
}
else {
// 上下文会在run_client中创建，这里初始化为NULL
client_contexts[i] = NULL;
}
}
}
/* 获取活动客户端数量 (新增函数) */
int get_active_client_count() {
int active_count = 0;
for (int i = 0; i < CONNECTIONS; i++) {
if (client_threads[i] && pthread_tryjoin_np(client_threads[i], NULL) == EBUSY) {
active_count++;
}
}
return active_count;
}
/* 客户端连接监控函数 (新增函数) */
void monitor_client_connections() {
static int monitor_counter = 0;
monitor_counter++;
// 每5次调用（约5秒）检查一次状态
if (monitor_counter >= 5) {
monitor_counter = 0;
int active_clients = get_active_client_count();
printf("Active client connections: %d/%d\n", active_clients, CONNECTIONS);
// 如果所有客户端线程都退出了，自动重启
if (active_clients == 0) {
printf("All clients stopped, restarting...\n");
start_all_clients();
} }
printf("Active connections: %d/%d\n", active_count, CONNECTIONS);
recovery_counter = 0;
} }
/*static int monitor_temp = 0; //static int monitor_temp = 0;
monitor_temp++; //monitor_temp++;
if (monitor_temp >= 30) { //if (monitor_temp >= 30) {
monitor_temp = 0; // monitor_temp = 0;
printf("30 second to stop all client\n"); // printf("30 second to stop all client\n");
stop_all_clients(); // // 设置关闭标志
}*/ // shutdown_flag = 1;
//
// // 停止并关闭监控定时器
// uv_timer_stop(handle);
// uv_close((uv_handle_t*)handle, NULL);
//
// // 停止所有客户端
// stop_all_clients();
//
// // 停止事件循环
// uv_stop(global_loop);
//}
}
void start_client_connect() {
// 创建全局事件循环
global_loop = uv_default_loop();
// 初始化所有客户端
init_clients(global_loop);
// 启动连接监控
uv_timer_init(global_loop, &monitor_timer);
uv_timer_start(&monitor_timer, monitor_connections, 1000, 1000);
// 运行事件循环
uv_run(global_loop, UV_RUN_DEFAULT);
// 清理资源（关键：确保循环完全停止）
uv_loop_close(global_loop);
global_loop = NULL;
} }
///* 主函数 */
//int main() {
// pthread_t threads[CONNECTIONS];
//
// // 创建客户端线程
// for (int i = 0; i < CONNECTIONS; i++) {
// int* index = (int*)malloc(sizeof(int));
// *index = i;
//
// if (pthread_create(&threads[i], NULL, run_client, index) != 0) {
// fprintf(stderr, "Failed to create thread %d\n", i);
// free(index);
// }
// }
//
// printf("Started %d client connections\n", CONNECTIONS);
//
// // 等待所有线程结束
// for (int i = 0; i < CONNECTIONS; i++) {
// pthread_join(threads[i], NULL);
// }
//
// return 0;
//}

6
LFtid1056/client2.h
View File

@@ -24,12 +24,10 @@ typedef struct {
int reconnect_attempts; // <20><>ǰ<EFBFBD><C7B0><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> int reconnect_attempts; // <20><>ǰ<EFBFBD><C7B0><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
volatile int shutdown; // <20>رձ<D8B1>־ (<28><><EFBFBD><EFBFBD>) volatile int shutdown; // <20>رձ<D8B1>־ (<28><><EFBFBD><EFBFBD>)
} client_context_t; } client_context_t;
//<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> //<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
void try_reconnect(uv_timer_t* timer);//<2F>ͻ<EFBFBD><CDBB><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ص<EFBFBD> void try_reconnect(uv_timer_t* timer);//<2F>ͻ<EFBFBD><CDBB><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ص<EFBFBD>
void on_connect(uv_connect_t* req, int status);//<2F>ͻ<EFBFBD><CDBB><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ӻص<D3BB> void on_connect(uv_connect_t* req, int status);//<2F>ͻ<EFBFBD><CDBB><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ӻص<D3BB>
void on_close(uv_handle_t* handle);//<2F>ͻ<EFBFBD><CDBB>˶Ͽ<CBB6><CFBF>ص<EFBFBD> void on_close(uv_handle_t* handle);//<2F>ͻ<EFBFBD><CDBB>˶Ͽ<CBB6><CFBF>ص<EFBFBD>
void send_binary_data(client_context_t* ctx, const unsigned char* data, size_t data_size);//<2F><>װ<EFBFBD><D7B0><EFBFBD><EFBFBD><EFBFBD>Ʊ<EFBFBD><C6B1>IJ<EFBFBD><C4B2><EFBFBD><EFBFBD><EFBFBD> <20>ͻ<EFBFBD><CDBB><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ӣ<EFBFBD><D3A3><EFBFBD><EFBFBD>ݣ<EFBFBD><DDA3><EFBFBD><EFBFBD>ݴ<EFBFBD>С
void stop_all_clients(); // <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> ֹͣ<CDA3><D6B9><EFBFBD>пͻ<D0BF><CDBB><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> void stop_all_clients(); // <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> ֹͣ<CDA3><D6B9><EFBFBD>пͻ<D0BF><CDBB><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
void start_all_clients(); //<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>пͻ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> void start_client_connect();//<2F><><EFBFBD><EFBFBD><EFBFBD>ͻ<EFBFBD><CDBB><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
void monitor_client_connections();//<2F><><EFBFBD>ؿͻ<D8BF><CDBB><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> void send_binary_data(client_context_t* ctx, const unsigned char* data, size_t data_size);

12
LFtid1056/main_thread.cpp
View File

@@ -74,20 +74,10 @@ void* client_manager_thread(void* arg) {
thread_info[index].state = THREAD_RUNNING; thread_info[index].state = THREAD_RUNNING;
pthread_mutex_unlock(&thread_info[index].lock); pthread_mutex_unlock(&thread_info[index].lock);
// <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>пͻ<D0BF><CDBB><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>߳<EFBFBD>
start_all_clients();
printf("Started client connections\n"); printf("Started client connections\n");
// <20><><EFBFBD><EFBFBD>ѭ<EFBFBD><D1AD> start_client_connect();
while (thread_info[index].state == THREAD_RUNNING) {
sleep(1);
// <20><><EFBFBD>ؿͻ<D8BF><CDBB><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>״̬<D7B4><CCAC><EFBFBD><EFBFBD><EFBFBD>ÿͻ<C3BF><CDBB><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ӳ<EFBFBD><D3B2>ֵĺ<D6B5><C4BA><EFBFBD><EFBFBD><EFBFBD>
monitor_client_connections();
}
// ֹͣ<CDA3><D6B9><EFBFBD>пͻ<D0BF><CDBB><EFBFBD><EFBFBD>߳<EFBFBD>
stop_all_clients();
printf("Stopped all client connections\n"); printf("Stopped all client connections\n");
// <20>߳<EFBFBD><DFB3><EFBFBD>ֹ<EFBFBD><D6B9><EFBFBD><EFBFBD> // <20>߳<EFBFBD><DFB3><EFBFBD>ֹ<EFBFBD><D6B9><EFBFBD><EFBFBD>