【網絡編程】簡易的 p2p 模型，實現兩臺虛擬機之間的簡單點對點通信，并以小見大觀察 TCP 協議的具體運行

文章目錄

基本概念
業務拆解
代碼實現
- 準備工作
- 實現被動的功能——多線程指針函數
- 實現主動的功能——用戶選擇界面
- 主函數
代碼執行效果
意外收獲
總結

推薦一個零聲教育學習教程，個人覺得老師講得不錯，分享給大家：[Linux，Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒體，CDN，P2P，K8S，Docker，TCP/IP，協程，DPDK等技術內容，點擊立即學習: https://github.com/0voice 鏈接。

基本概念

p2p，一個讓所有投資者脊背發涼的金融概念，一個去中心化的金融概念，其實它的本質只是一個技術。P2P（Peer-to-Peer）是一種去中心化的網絡架構模式，中文通常翻譯為"點對點"或"對等網絡"。它代表了與傳統的客戶端-服務器（Client-Server）模型完全不同的網絡通信理念。

P2P 的核心概念

去中心化：
- 沒有中央服務器控制整個網絡
- 所有參與者（節點）地位平等
- 每個節點既是客戶端又是服務器（稱為"對等體"）
直接通信：
- 節點之間直接連接和交換數據
- 不需要通過中間服務器中轉
- 通信路徑更短，延遲更低
資源共享：
- 每個節點貢獻自己的資源（帶寬、存儲、計算能力）
- 資源分布在整個網絡中
- 節點越多，網絡整體能力越強

與傳統客戶端-服務器模型的對比

特性	P2P 網絡	客戶端-服務器模型
架構	去中心化	中心化
節點角色	既是客戶端又是服務器	嚴格區分客戶端和服務器
擴展性	節點越多性能越好	服務器可能成為瓶頸
可靠性	單點故障不影響整個網絡	服務器故障導致服務中斷
資源分布	資源分散在各個節點	資源集中在服務器

P2P 的典型應用場景

文件共享：
- BitTorrent：用戶直接從其他用戶下載文件片段
- 早期Napster：音樂文件共享（混合式P2P）
加密貨幣：
- 比特幣/以太坊：交易驗證通過P2P網絡完成
- 區塊鏈技術的基礎架構
即時通訊：
- 早期Skype：語音通話直接在對等體間建立
- 某些隱私通訊應用
內容分發：
- P2P CDN：利用用戶設備分發內容
- 直播平臺的P2P加速

我所能設想到的一個應用就是 “智能家具” 的設計，我們用手機與智能家居進行點對點的 P2P 連接，直接下命令，而非繞一大圈地經過中央服務器。這樣的設計才是系統開銷小，用戶體驗好。

業務拆解

我們在前面的基本概念介紹里面已經說到過 P2P 網絡的各個節點既是客戶端又是服務器，本篇文章之中，我們要抓住這一個點設計一個點對點通信的簡易代碼。至于像加密驗證等 “高級玩意”，本篇文章是絕對不會涉及的。

問題來了，我們該怎么設計呢？我們可以嘗試一下問自己，到底想要什么功能效果。我問過自己，可以分成兩大類——主動類和被動類。

主動類的功能：

用戶之間隨時發起信息。
用戶選擇想要連接的對象 IP（可以重置 IP）。
自己是一個客戶端，可主動發起并實現與對應 IP 的遠程連接。
結束程序。

被動類的功能：

自己本身是服務器，被動監聽到來訪 IP，并隨即分配套接字資源負責對應的 I/O 任務。
自動地接收信息。（這里回想起《角頭》中白毛對 “憨春” 說：“憨春大，我 BOSS 找你那么多次，你為什么都已讀不回呀？啊？”）

為了簡化問題，本篇文章所展示的代碼，只實現 “一個設備僅有一個連接，如果想要新的連接就必須刪掉舊的連接” 的設計。

對于主動類的功能，我將采用 “用戶界面” 式的循環交互設計，類似的代碼可見我之前寫過的一篇關于 “通訊錄小項目” 的文章（原文鏈接在此）。

對于被動類的功能，我將采用多線程編程的設計。有兩個被動類的功能，那就有兩個子線程分別負責。這兩個子線程因 “一個設備僅有一個連接，如果想要新的連接就必須刪掉舊的連接” 的簡化，而使用了 “SELECT” 定時關注連接所對應的套接字是否對接收、讀寫等事件就緒。select 是一種多路復用（multiplexing）I/O 機制，用于同時監視多個文件描述符（file descriptors），以確定哪些文件描述符已經準備好進行 I/O 操作（如讀取、寫入或異常條件）。類似的代碼可見我之前寫過的一篇關于多路 I/O 復用的文章（原文鏈接在此）。

為了確保程序能夠被正常關閉，所建立套接字都是非阻塞的，即定時執行，重復循環計時。

代碼實現

準備工作

準備頭文件

#include <stdio.h>
#include <stdlib.h>     //  EXIT_FAILURE 是一個標準宏，exit 函數
#include <string.h>
#include <unistd.h>     //  close 函數
#include <pthread.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/select.h> // 添加select用于超時處理
#include <fcntl.h>				// 用于更改套接字的模式，比如非阻塞模式
#include <errno.h>				// 這是全局變量 errno，用于健壯的讀取功能

準備宏定義

#define MAX_MSG_LEN 1024
#define PORT 5000

聲明與定義全局變量，該全局變量能綜合、集成所有的運行參數。我們將之命名為 NodeState

// 全局狀態結構體
typedef struct {int server_fd;int connection_fd;int running;pthread_mutex_t lock;char peer_ip[16];  // 存儲點分十進制IP地址int peer_port;
} NodeState;// 聲明并定義全局變量
// 點號（.）在這里是C99標準引入的指定初始化器語法的一部分。它的作用是明確指定結構體成員的初始化值，而不是依賴于成員在結構體中的順序。
NodeState node_state = {.server_fd = -1,.connection_fd = -1,.running = 1,.lock = PTHREAD_MUTEX_INITIALIZER,.peer_ip = "",.peer_port = PORT
};

需要注意到的是，點號（.）在這里是C99標準引入的指定初始化器語法的一部分。它的作用是明確指定結構體成員的初始化值，而不是依賴于成員在結構體中的順序。

緊接著是錯誤處理函數，

void error(const char *msg) {perror(msg);exit(EXIT_FAILURE);     //  EXIT_FAILURE 是一個標準宏，定義在 <stdlib.h> 中，用于表示程序執行失敗。// 當 exit 函數被調用時，程序會執行以下操作：// 關閉所有打開的文件：關閉所有通過標準 I/O 函數（如 fopen）打開的文件流。// 刷新緩沖區：刷新所有標準 I/O 緩沖區，確保所有未寫入的數據都被寫入目標文件或設備。// 調用清理函數：執行所有通過 atexit 注冊的清理函數（如果有）。// 終止程序：終止程序的執行，并將 status 參數作為退出狀態碼返回給操作系統。
}

當我們結束程序的時候，需要定義清理資源函數

// 清理資源
void cleanup() {pthread_mutex_lock(&node_state.lock);if (node_state.connection_fd != -1) {close(node_state.connection_fd);node_state.connection_fd = -1;  //  重置}if (node_state.server_fd != -1) {close(node_state.server_fd);node_state.server_fd = -1;  //  重置}pthread_mutex_unlock(&node_state.lock);printf("[*] Resources cleaned up\n");return;
}

為了能讓程序正常結束，而非讓套接字對應的 accept 和 recv 函數在用戶選擇退出的時候，一直處于阻塞各自的線程之中，故而我們定義了套接字設置函數

// 設置套接字為非阻塞
void set_nonblocking(int sockfd) {int flags = fcntl(sockfd, F_GETFL, 0);if (flags == -1) {perror("fcntl F_GETFL");return;}if (fcntl(sockfd, F_SETFL, flags | O_NONBLOCK) == -1) {perror("fcntl F_SETFL");}
}

實現被動的功能——多線程指針函數

這個函數是回調函數，它即將被線程所調用。他主要是調用了 select 多路復用機制，過一段時間就會檢查對應的套接字是否關于讀寫事件就緒，而且還設置了超時機制，這是因為所有的套接字都被設置成了非阻塞模式。另外，之所以不用 EPOLL 機制，是因為該線程只針對一個連接（一個套接字），因此 select 的效果會更好。我們的智能家居環境不也只是零星幾個連接？并不是一個服務器，因而并不需要那么多的連接。

// 服務器線程函數
void *server_thread(void *arg) {struct sockaddr_in address;int addrlen = sizeof(address);// 創建服務器套接字node_state.server_fd = socket(AF_INET, SOCK_STREAM, 0);// 設置套接字為非阻塞set_nonblocking(node_state.server_fd);// 設置套接字選項 (允許地址重用)int opt = 1;if (setsockopt(node_state.server_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt))) {error("setsockopt failed");}address.sin_family = AF_INET;address.sin_addr.s_addr = INADDR_ANY;address.sin_port = htons(PORT);// 綁定套接字if (bind(node_state.server_fd, (struct sockaddr *)&address, sizeof(address)) < 0) {error("bind failed");}// 監聽連接if (listen(node_state.server_fd, 3) < 0) {error("listen failed");}printf("[*] Server listening on port %d\n", PORT);while (node_state.running) {fd_set read_fds;FD_ZERO(&read_fds);FD_SET(node_state.server_fd, &read_fds);// 設置超時時間為1秒struct timeval timeout;timeout.tv_sec = 1;timeout.tv_usec = 0;// 使用select等待連接請求或超時int activity = select(node_state.server_fd + 1, &read_fds, NULL, NULL, &timeout);// 檢查是否有新連接if (activity > 0 && FD_ISSET(node_state.server_fd, &read_fds)) {int new_socket;if ((new_socket = accept(node_state.server_fd, (struct sockaddr *)&address, (socklen_t*)&addrlen)) < 0) {if (errno != EWOULDBLOCK && errno != EAGAIN) {perror("accept failed");}continue;}pthread_mutex_lock(&node_state.lock);// 檢查是否已有連接if (node_state.connection_fd != -1) {printf("[!] Connection already exists. Closing new connection.\n");close(new_socket);} else {node_state.connection_fd = new_socket;printf("[+] Accepted connection from %s\n", inet_ntoa(address.sin_addr));}pthread_mutex_unlock(&node_state.lock);}// 檢查是否需要退出if (!node_state.running) {break;}}return NULL;}

這個函數是回調函數，它即將被線程所調用。設計的道理類似于上面。

// 接收消息線程函數
void *recv_thread(void *arg) {char buffer[MAX_MSG_LEN] = {0};while (node_state.running) {pthread_mutex_lock(&node_state.lock);int current_fd = node_state.connection_fd;pthread_mutex_unlock(&node_state.lock);if (current_fd == -1) {// 沒有連接時短暫休眠usleep(100000); // 100mscontinue;}fd_set read_fds;FD_ZERO(&read_fds);FD_SET(current_fd, &read_fds);// 設置超時時間為1秒struct timeval timeout;timeout.tv_sec = 1;timeout.tv_usec = 0;// 使用select等待數據或超時int activity = select(current_fd + 1, &read_fds, NULL, NULL, &timeout);if (activity < 0 && errno != EINTR) {perror("select error");continue;}// 檢查是否有數據到達if (activity > 0 && FD_ISSET(current_fd, &read_fds)) {// 接收數據ssize_t valread = recv(current_fd, buffer, MAX_MSG_LEN - 1, 0);if (valread <= 0) {if (valread == 0) {printf("[!] Connection closed by peer\n");} else if (node_state.running) {perror("recv failed");}pthread_mutex_lock(&node_state.lock);if (node_state.connection_fd == current_fd) {close(current_fd);node_state.connection_fd = -1;}pthread_mutex_unlock(&node_state.lock);printf("[*] Ready for new connections\n");} else {buffer[valread] = '\0';printf("\n[Peer] %s\nYou: ", buffer);fflush(stdout);}}// 檢查是否需要退出if (!node_state.running) {break;}}return NULL;
}

實現主動的功能——用戶選擇界面

用戶可以主動的發起遠程連接，就像客戶端一樣。

// 連接到對等節點
void connect_to_peer() {if (strlen(node_state.peer_ip) == 0) {printf("[!] Peer IP not set\n");return;}struct sockaddr_in serv_addr;int sock = 0;if ((sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {error("socket creation failed");}// 設置套接字為非阻塞set_nonblocking(sock);serv_addr.sin_family = AF_INET;serv_addr.sin_port = htons(node_state.peer_port);// 轉換IP地址if (inet_pton(AF_INET, node_state.peer_ip, &serv_addr.sin_addr) <= 0) {printf("[!] Invalid address/ Address not supported\n");close(sock);return;}printf("[*] Trying to connect to %s:%d...\n", node_state.peer_ip, node_state.peer_port);// 嘗試連接（非阻塞）int connect_result = connect(sock, (struct sockaddr *)&serv_addr, sizeof(serv_addr));if (connect_result < 0 && errno != EINPROGRESS) {perror("Connection failed");close(sock);return;}// 使用 select 檢查連接狀態fd_set write_fds;FD_ZERO(&write_fds);FD_SET(sock, &write_fds);struct timeval timeout;timeout.tv_sec = 5; // 5秒超時timeout.tv_usec = 0;int sel = select(sock + 1, NULL, &write_fds, NULL, &timeout);if (sel <= 0) {printf("[-] Connection timed out\n");close(sock);return;}// 檢查套接字錯誤int so_error;socklen_t len = sizeof(so_error);getsockopt(sock, SOL_SOCKET, SO_ERROR, &so_error, &len);if (so_error != 0) {printf("[-] Connection failed: %s\n", strerror(so_error));close(sock);return;}pthread_mutex_lock(&node_state.lock);node_state.connection_fd = sock;pthread_mutex_unlock(&node_state.lock);printf("[+] Connected to peer %s:%d\n", node_state.peer_ip, node_state.peer_port);
}

連接建立后，雙方可以主動發送信息，可以像網絡噴子一樣 “對噴”

// 發送消息
void send_message(const char *message) {pthread_mutex_lock(&node_state.lock);int current_fd = node_state.connection_fd;pthread_mutex_unlock(&node_state.lock);if (current_fd == -1) {printf("[!] Not connected to any peer\n");return;}if (send(current_fd, message, strlen(message), 0) < 0) {perror("send failed");pthread_mutex_lock(&node_state.lock);if (node_state.connection_fd == current_fd) {close(current_fd);node_state.connection_fd = -1;}pthread_mutex_unlock(&node_state.lock);}
}

綜合前面兩個函數，我們順帶還給交流的雙方提供 “噴不過就絕交掛線” 的功能，這就是用戶選擇互動界面函數。用戶想要發送信息，必須是 3 -> 2 -> 1 的執行順序，即 “3” 先鎖定目標，“2” 對目標進行遠程連接，“1” 在實現連接后可以互發消息。如果我們想要換一個主機連接。我們可以繼續 “3”，重置遠程 IP，并斷開原來的連接，清理套接字資源。實在不想玩，可以直接按 “4”，退出。

// 用戶交互界面，主線程
//  程序員如想發消息，應先 3 -> 2 -> 1
void user_interface() {char input[100];char message[MAX_MSG_LEN];while (node_state.running) {printf("\nOptions:\n1. Send message\n2. Connect to peer\n3. Set peer IP\n4. Exit\nChoose: ");fflush(stdout);if (fgets(input, sizeof(input), stdin) == NULL) {   //  程序員從鍵盤中輸入功能鍵break;}switch (input[0]) {case '1': // 發送消息printf("Enter message: ");fflush(stdout);if (fgets(message, MAX_MSG_LEN, stdin)) {   //  程序員從鍵盤中輸入聊天信息// 移除換行符message[strcspn(message, "\n")] = 0;    //  把輸入的聊天信息最后的 “\n” 精準轉變為 “\0”，以便打印和發送send_message(message);}break;case '2': // 連接到對等節點connect_to_peer();break;case '3': // 設置對等節點IPprintf("Enter peer IP: ");fflush(stdout);if (fgets(node_state.peer_ip, sizeof(node_state.peer_ip), stdin)) {// strcspn 是 C 語言標準庫中的一個字符串處理函數，用于計算一個字符串中不包含指定字符集合的第一個子串的長度。// 移除換行符node_state.peer_ip[strcspn(node_state.peer_ip, "\n")] = 0;printf("Peer IP set to: %s\n", node_state.peer_ip);}if (node_state.connection_fd != -1) {printf("[!] Already connected to a peer. Now change anothor peer to connect.\n");close(node_state.connection_fd);}break;case '4': // 退出node_state.running = 0;printf("[*] Shutting down...\n");break;default:printf("Invalid option\n");}}return;
}

主函數

實現被動的功能需要多線程編程，pthread_create 調用 server_thread 和 recv_thread 兩個指針函數（回調函數），子線程的執行是需要線程來調度的（也就是全自動的）。在建立了兩個子線程后，我們可以執行用戶界面函數，實現發消息等 “主動的功能”。當選擇結束程序后，要及時回收線程資源。

int main() {printf("=== P2P Node ===\n");// 創建服務器線程pthread_t server_tid, recv_tid;if (pthread_create(&server_tid, NULL, server_thread, NULL)) {       //  accept 會掛起error("could not create server thread");}// 創建接收消息線程if (pthread_create(&recv_tid, NULL, recv_thread, NULL)) {           //  recv 會掛起，如果沒鏈接就會睡眠error("could not create receive thread");}//  以上兩個線程都是在沒有監聽到新的來訪 IP、沒有人發來消息時，循環執行超時等待而后睡眠，同時還負責檢查服務器的連接狀態，如果斷了就繼續連接// 設置對等節點IP (初始為空)strcpy(node_state.peer_ip, "");// 啟動用戶界面user_interface();   //  至于主動發消息，關閉程序的，建立連接的 “主動項目” 就有用戶界面統一管理。// 清理資源cleanup();      //  當程序員選擇了 “4”，就會進入清理資源的函數處，線程就沒有什么函數任務可執行了，緊接著就是關閉 套接字 資源// 等待線程結束// 雖然線程的任務函數已經跳出了死循環，但線程資源本身還未釋放pthread_join(server_tid, NULL);     pthread_join(recv_tid, NULL);printf("[*] Program exited\n");return 0;
}

代碼執行效果

編譯代碼

qiming@qiming:~/share/CTASK/TCP_test$ gcc -o p2p p2p_test.c -lpthread

在兩臺虛擬機上執行代碼

qiming@qiming:~/share/CTASK/TCP_test$ ./p2p 
=== P2P Node ===Options:
1. Send message
2. Connect to peer
3. Set peer IP
4. Exit
Choose: [*] Server listening on port 5000

我是使用 Xshell 去遠程運行多臺虛擬機
在這里插入圖片描述
具體界面如下

先選擇 “3” 選項，鎖定要遠程連接的對象

再選 “2” 選項，對設定好的 IP 進行連接

連接好后，選擇 “1” 選項，雙方可以對線互噴了。

選擇 “4” 則會退出程序。
在這里插入圖片描述

意外收獲

如果我們在上述的程序執行過程中，利用 WireShark 軟件去記錄一路上的網絡傳輸發包情況。我們將會觸摸到 TCP 協議層——傳輸層的具體物體——包（處于第四層，對應 TCP 的報文），這些報文的制作都來自底層庫所定義的函數 accept 和 recv、send、close，我們是可以不用關心報文的制作，都有內核底層為我們代勞了。但我們依舊要學習這些底層，因為這能增強我們的計算機工程感覺。
在這里插入圖片描述
我仔細地在網絡抓包工具目錄條，里面抓到了三個包——它們其實就是 “三次握手”

相互發消息，發的對稱兩個包

而后斷開連接產生的四次揮手，對應四個包

以及