将客户端连接功能整合至主线程管理框架下

LFtid1056/client2.cpp

@@ -10,13 +10,17 @@
 
  // 配置参数
 #define INITIAL_DATA_SIZE  128          // 初始数据0.1KB
-#define CONNECTIONS 5                    // 并发连接数 设备连接数
+#define CONNECTIONS 2                    // 并发连接数 设备连接数
 #define SERVER_IP "101.132.39.45"        // 目标服务器IP 阿里云服务器 "101.132.39.45""172.27.208.1"
 #define SERVER_PORT 1056                 // 目标服务器端口
 #define BASE_RECONNECT_DELAY 5000        // 基础重连延迟(ms)
 #define MAX_RECONNECT_DELAY 60000        // 最大重连延迟(ms)
 #define MAX_RECONNECT_ATTEMPTS 10        // 最大重连次数
 
+// 全局变量用于管理客户端线程
+static pthread_t client_threads[CONNECTIONS];
+static client_context_t* client_contexts[CONNECTIONS] = { NULL };
+
 /* 缓冲区分配回调 */
 void alloc_buffer(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
     void* buffer = malloc(suggested_size);
@@ -162,10 +166,11 @@ void on_close(uv_handle_t* handle) {
     uv_timer_stop(&ctx->timer);
     uv_timer_stop(&ctx->reconnect_timer);
 
-    // 取消读操作（如果有的话）
+    // 取消读操作
     uv_read_stop((uv_stream_t*)&ctx->client);
 
-    // 指数退避算法
+    // 如果不是主动关闭，尝试重连
+    if (!ctx->shutdown) {
         int delay = BASE_RECONNECT_DELAY * pow(2, ctx->reconnect_attempts);
         delay = delay > MAX_RECONNECT_DELAY ? MAX_RECONNECT_DELAY : delay;
 
@@ -174,14 +179,15 @@ void on_close(uv_handle_t* handle) {
 
         uv_timer_start(&ctx->reconnect_timer, (uv_timer_cb)try_reconnect, delay, 0);
         ctx->reconnect_attempts++;
+    }
 }
 
 /* 尝试重连函数 */
 void try_reconnect(uv_timer_t* timer) {
     client_context_t* ctx = (client_context_t*)timer->data;
 
-    if (ctx->state != STATE_DISCONNECTED) {
-        fprintf(stdout, "[Client %d] Reconnect skipped: Invalid state\n", ctx->index);
+    if (ctx->state != STATE_DISCONNECTED || ctx->shutdown) {
+        fprintf(stdout, "[Client %d] Reconnect skipped: Invalid state or shutdown\n", ctx->index);
         return;
     }
 
@@ -191,7 +197,7 @@ void try_reconnect(uv_timer_t* timer) {
         return;
     }
 
-    // 重新初始化TCP句柄(重要!)
+    // 重新初始化TCP句柄
     uv_tcp_init(ctx->loop, &ctx->client);
     ctx->client.data = ctx;
 
@@ -237,13 +243,12 @@ void on_connect(uv_connect_t* req, int status) {
         //send_initial_data(ctx);
 
         // 启动定时器
-        uv_timer_start(&ctx->timer, on_timer, 6000, 6000);
+        //uv_timer_start(&ctx->timer, on_timer, 6000, 6000);
     }
 
     free(req);
 }
 
-
 /* 客户端线程函数 */
 void* run_client(void* arg) {
     int index = *(int*)arg;
@@ -259,6 +264,7 @@ void* run_client(void* arg) {
     ctx->index = index;
     ctx->state = STATE_DISCONNECTED;
     ctx->reconnect_attempts = 0;
+    ctx->shutdown = 0;  // 初始化关闭标志
 
     // 初始化libuv句柄
     uv_tcp_init(loop, &ctx->client);
@@ -273,41 +279,124 @@ void* run_client(void* arg) {
     // 首次连接尝试
     try_reconnect(&ctx->reconnect_timer);
 
-    // 运行事件循环
-    uv_run(loop, UV_RUN_DEFAULT);
+    // 运行事件循环，定期检查关闭标志
+    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);
-    free(loop);
+    uv_loop_delete(loop);
     free(ctx);
     return NULL;
 }
 
-/* 主函数 */
-int main() {
-    pthread_t threads[CONNECTIONS];
+/* 停止所有客户端线程 (新增函数) */
+void stop_all_clients() {
+    for (int i = 0; i < CONNECTIONS; i++) {
+        if (client_contexts[i]) {
+            client_contexts[i]->shutdown = 1;  // 设置关闭标志
+        }
+    }
 
-    // 创建客户端线程
+    // 等待所有客户端线程退出
+    for (int i = 0; i < CONNECTIONS; i++) {
+        if (client_threads[i]) {
+            pthread_join(client_threads[i], NULL);
+            client_threads[i] = 0;
+        }
+    }
+}
+
+/* 启动所有客户端线程 (新增函数) */
+void start_all_clients() {
     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);
+        if (pthread_create(&client_threads[i], NULL, run_client, index) != 0) {
+            fprintf(stderr, "Failed to create client thread %d\n", i);
             free(index);
         }
+        else {
+            // 上下文会在run_client中创建，这里初始化为NULL
+            client_contexts[i] = NULL;
         }
-
-    printf("Started %d client connections\n", CONNECTIONS);
-
-    while (1) {
-        printf("sleep 1\n");
-        sleep(10);
     }
-    // 等待所有线程结束
-    for (int i = 0; i < CONNECTIONS; i++) {
-        pthread_join(threads[i], NULL);
-    }
-
-    return 0;
 }
+
+/* 获取活动客户端数量 (新增函数) */
+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();
+        }
+    }
+
+    /*static int monitor_temp = 0;
+    monitor_temp++;
+    if (monitor_temp >= 30) {
+        monitor_temp = 0;
+        printf("30 second to stop all client\n");
+        stop_all_clients();
+    }*/
+}
+///* 主函数 */
+//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;
+//}

LFtid1056/client2.h

@@ -22,6 +22,7 @@ typedef struct {
     uv_timer_t reconnect_timer;// <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ʱ<EFBFBD><CAB1>
     connection_state_t state;  // <20><>ǰ<EFBFBD><C7B0><EFBFBD><EFBFBD>״̬
     int reconnect_attempts;    // <20><>ǰ<EFBFBD><C7B0><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
+    volatile int shutdown;     // <20>رձ<D8B1>־ (<28><><EFBFBD><EFBFBD>)
 } client_context_t;
 
 //<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
@@ -29,3 +30,6 @@ void try_reconnect(uv_timer_t* timer);//
 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 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 start_all_clients(); //<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>пͻ<D0BF><CDBB><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
+void monitor_client_connections();//<2F><><EFBFBD>ؿͻ<D8BF><CDBB><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>

LFtid1056/main_thread.cpp

@@ -4,6 +4,8 @@
 #include <unistd.h>
 #include <errno.h>
 #include <time.h>
+#include "PQSMsg.h"
+#include "client2.h"
 
 /* <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> */
 #define THREAD_CONNECTIONS 10        // <20><><EFBFBD><EFBFBD><EFBFBD>߳<EFBFBD><DFB3><EFBFBD>
@@ -61,33 +63,37 @@ void* work_thread(void* arg) {
     return NULL;
 }
 /* <20>̹߳<DFB3><CCB9><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> 1<><31><EFBFBD><EFBFBD><EFBFBD>߳<EFBFBD>*/
-void* work_thread_1(void* arg) {
-    int index = *(int*)arg;  // <20><>ȡ<EFBFBD>߳<EFBFBD><DFB3><EFBFBD><EFBFBD><EFBFBD>
-    free(arg);  // <20>ͷŶ<CDB7>̬<EFBFBD><CCAC><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ڴ<EFBFBD>
+/* <20>ͻ<EFBFBD><CDBB><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ӹ<EFBFBD><D3B9><EFBFBD><EFBFBD>̺߳<DFB3><CCBA><EFBFBD>*/
+void* client_manager_thread(void* arg) {
+    int index = *(int*)arg;
+    free(arg);
 
     // <20><><EFBFBD><EFBFBD><EFBFBD>߳<EFBFBD>״̬Ϊ<CCAC><CEAA><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
     pthread_mutex_lock(&thread_info[index].lock);
-    printf("work_thread_1 %d started\n", index);
+    printf("Client Manager Thread %d started\n", index);
     thread_info[index].state = THREAD_RUNNING;
     pthread_mutex_unlock(&thread_info[index].lock);
 
-    // 1<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>߳<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ѭ<EFBFBD><EFBFBD>
-    while (1) {
-        sleep(5);
+    // <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>пͻ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>߳<EFBFBD>
+    start_all_clients();
+    printf("Started client connections\n");
 
-        // 10%<EFBFBD><EFBFBD><EFBFBD><EFBFBD>ģ<EFBFBD><EFBFBD><EFBFBD>̹߳<EFBFBD><EFBFBD><EFBFBD>
-        if (rand() % 10 == 0) {
-            pthread_mutex_lock(&thread_info[index].lock);
-            printf("Thread %d simulated failure\n", index);
-            pthread_mutex_unlock(&thread_info[index].lock);
-            break;
-        }
+    // <20><><EFBFBD><EFBFBD>ѭ<EFBFBD><EFBFBD>
+    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");
+
     // <20>߳<EFBFBD><DFB3><EFBFBD>ֹ<EFBFBD><D6B9><EFBFBD><EFBFBD>
     pthread_mutex_lock(&thread_info[index].lock);
     thread_info[index].state = THREAD_STOPPED;
-    printf("work_thread_1 %d stopped\n", index);
+    printf("Client Manager Thread %d stopped\n", index);
     pthread_mutex_unlock(&thread_info[index].lock);
     return NULL;
 }
@@ -108,26 +114,19 @@ void restart_thread(int index) {
     *new_index = index;
 
     if (index == 0) {
-        //<EFBFBD><EFBFBD><EFBFBD>߳<EFBFBD>1<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
-        if (pthread_create(&thread_info[index].tid, NULL, work_thread_1, new_index) != 0) {
+        // <20>ͻ<EFBFBD><CDBB>˹<EFBFBD><EFBFBD><EFBFBD><EFBFBD>߳<EFBFBD>
+        if (pthread_create(&thread_info[index].tid, NULL, client_manager_thread, new_index) != 0) {
             pthread_mutex_lock(&global_lock);
-            printf("Failed to restart work_thread_1 %d\n", index);
+            printf("Failed to restart client manager thread %d\n", index);
             thread_info[index].state = THREAD_CRASHED;
             pthread_mutex_unlock(&global_lock);
             free(new_index);
         }
     }
     else {
-        //<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>̳߳<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
-        if (pthread_create(&thread_info[index].tid, NULL, work_thread, new_index) != 0) {
-            pthread_mutex_lock(&global_lock);
-            printf("Failed to restart thread %d\n", index);
-            thread_info[index].state = THREAD_CRASHED;
-            pthread_mutex_unlock(&global_lock);
-            free(new_index);
+        // <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>
     }
-    }
-    
 }
 
 /* <20>̴߳<DFB3><CCB4><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> */
@@ -136,7 +135,7 @@ int is_thread_alive(pthread_t tid) {
 }
 
 /* <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD> */
-int main1() {
+int main() {
     srand(time(NULL));  // <20><>ʼ<EFBFBD><CABC><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
 
     // <20><>ʼ<EFBFBD><CABC><EFBFBD>߳<EFBFBD><DFB3><EFBFBD><EFBFBD><EFBFBD>
@@ -152,20 +151,18 @@ int main1() {
         *index = i;
 
         if (i == 0) {
-            //<EFBFBD><EFBFBD><EFBFBD>߳<EFBFBD><EFBFBD>߳<EFBFBD>1<EFBFBD><EFBFBD>
-            if (pthread_create(&thread_info[i].tid, NULL, work_thread_1, index) != 0) {
-                printf("Failed to create work_thread_1 %d\n", i);
+            // <20>ͻ<EFBFBD><EFBFBD>˹<EFBFBD><EFBFBD><EFBFBD><EFBFBD>߳<EFBFBD>
+            if (pthread_create(&thread_info[i].tid, NULL, client_manager_thread, index) != 0) {
+                printf("Failed to create client manager thread %d\n", i);
                 free(index);
             }
         }
         else {
-            //<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>̳߳<EFBFBD>
-            if (pthread_create(&thread_info[i].tid, NULL, work_thread, index) != 0) {
-                printf("Failed to create thread %d\n", i);
+            // <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>
             free(index);
         }
     }
-    }
 
     printf("Thread monitoring system started with %d workers\n", THREAD_CONNECTIONS);