linux下用c++11寫一個UDP回顯程序

需求：

1）從2個UDP端口接收數據，并在同樣的端口回顯。echo

2）多個處理線程，多個發送線程；

3）使用條件變量喚醒；

#include <stack>
#include <mutex>
#include <atomic>#include <iostream>
#include <thread>
#include <mutex>
#include <queue>
#include <vector>
#include <unistd.h>
#include <string.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <sys/epoll.h>
#include <arpa/inet.h>   // for inet_pton
#include <cstring>       // for memset
#include <fcntl.h>       // for non-block
#include <signal.h>
#include <condition_variable>// #include "concurrentqueue.h" // moodycamel concurrentqueue 頭文件using namespace std;
/*
編譯方式（Linux）
g++ -std=c++11           -pthread cssff.cpp -o udp_server -lstdc++
g++ -std=c++11 -Wall -O2 -pthread cssff.cpp -o udp_server -lstdc++你可以用 netcat 發送測試數據：
echo "hello port1" | nc -u 127.0.0.1 4000
echo "hello port2" | nc -u 127.0.0.1 5000
*//*
你幾乎不會收到大于 1472~1500 的數據
推薦設為 1536 字節：正好是大多數操作系統分配 UDP buffer 的默認對齊粒度，還是 64 的倍數（好對齊
*/#define NUM_PROC_THREADS 1
#define NUM_SEND_THREADS 1#define BUF_SIZE_DEF 1536
constexpr int PORT1 = 4000;
constexpr int PORT2 = 5000;
constexpr int MAX_EVENTS = 10;
constexpr int BUF_SIZE = BUF_SIZE_DEF;// 一次批量處理至多32個，防止抖動和阻塞
const size_t MAX_BATCH = 32;// 退出標志
std::atomic<bool> g_running(true);// 數據包的簡單封裝
struct Packet {char data[BUF_SIZE];sockaddr_in addr_from;socklen_t addr_from_len;int port_from;    // 記錄是哪個端口接收的size_t data_len;sockaddr_in addr_to;socklen_t addr_to_len;int port_to;// 構造函數，默認初始化所有成員Packet() {reset();}// reset 函數，供復用時調用，清空所有數據和結構void reset() {memset(data, 0, sizeof(data));memset(&addr_from, 0, sizeof(addr_from));addr_from_len = 0;port_from = 0;data_len = 0;memset(&addr_to, 0, sizeof(addr_to));addr_to_len = 0;port_to = 0;}
};// 這個無鎖內存池，需要使用外部類支持
// 手動下載源碼（沒有官方包管理器直接安裝）
// git clone https://github.com/cameron314/concurrentqueue.git
// 
// 編譯指定頭文件目錄
// g++ -std=c++11 -pthread -I/path/to/concurrentqueue cssff.cpp -o udp_server
// 
// #define LOCK_IN_POOL 1
// class PacketPool {
//     moodycamel::ConcurrentQueue<Packet*> pool;
//     const size_t MAX_POOL_SIZE;
//     std::atomic<size_t> total_allocated;// public:
//     PacketPool(size_t max_pool = 10000) 
//         : MAX_POOL_SIZE(max_pool), total_allocated(0) {}//     ~PacketPool() {
//         Packet* pkt;
//         while (pool.try_dequeue(pkt)) {
//             delete pkt;
//         }
//     }//     Packet* acquire() {
//         Packet* pkt = nullptr;
//         if (pool.try_dequeue(pkt)) {
//             return pkt;
//         }
//         total_allocated.fetch_add(1, std::memory_order_relaxed);
//         return new Packet();
//     }//     void release(Packet* pkt) {
//         if (!pkt) return;//         size_t cached = pool.size_approx(); // 估算當前緩存大小
//         if (cached < MAX_POOL_SIZE) {
//             pool.enqueue(pkt);
//         } else {
//             delete pkt;
//             total_allocated.fetch_sub(1, std::memory_order_relaxed);
//         }
//     }//     size_t allocatedCount() const {
//         return total_allocated.load(std::memory_order_relaxed);
//     }//     // 估算緩存數（不是嚴格準確）
//     size_t cachedCount() const {
//         return pool.size_approx();
//     }
// };class PacketPool {std::vector<Packet*> pool;mutable std::mutex mtx;const size_t MAX_POOL_SIZE = 10000;public:~PacketPool() {for (auto pkt : pool) delete pkt;}Packet* acquire() {std::lock_guard<std::mutex> lock(mtx);if (!pool.empty()) {Packet* pkt = pool.back();pool.pop_back();pkt->reset();return pkt;}return new Packet();}void release(Packet* pkt) {if (pkt == nullptr) return;std::lock_guard<std::mutex> lock(mtx);if (pool.size() < MAX_POOL_SIZE) {pool.push_back(pkt);} else {delete pkt;}}size_t cachedCount() const {std::lock_guard<std::mutex> lock(mtx);return pool.size();}
};/////////////////////////////////////////////////////////////////////
// TODO: std::condition_variable 優化等待，暫時可能性能問題不大// 線程安全隊列
std::queue<Packet*> recv_queue_1;
std::queue<Packet*> send_queue_1;
// 枷鎖使用，或者配合那個條件變量使用
std::mutex recv_mutex_1;
std::mutex send_mutex_1;
// 使用條件變量來同步
std::condition_variable recv_cv_1;
std::condition_variable send_cv_1;
/*
notify_one 基本規則：
每調用一次 notify_one()，只會喚醒 一個正在等待在該 condition_variable 上的線程。
如果沒有線程在等待，這次調用 不會保存或累積信號，即后來的線程 wait() 依然會阻塞，直到下一次 notify_one() 或 notify_all()。
?? 注意：條件變量不是信號量，不會記住之前的“通知次數”。
?? 所以：謂詞一定要加上隊列不為空，否則，如果干活時候錯過了喚醒會造成一直等待信號的問題。
其中wait原理偽代碼：
function wait(unique_lock& lock, predicate pred):while not pred():                  // 先檢查謂詞// 1. 把 mutex 解鎖，讓其他線程可以修改共享狀態lock.unlock()// 2. 阻塞等待條件變量通知wait_for_notify()// 3. 被喚醒后，重新加鎖lock.lock()end while// 當 predicate() 返回 true 時，函數返回return*/// 全局的內存池子
PacketPool pool;const int MAX_READ_PER_FD = 10;// 1）這個線程負責從這2個端口接收數據，放到對應的隊列中；
void recv_thread(int sock1, int sock2) {int epfd = epoll_create1(0);if (epfd == -1) {perror("epoll_create1");return;}epoll_event ev1 = {0}, ev2 = {0};ev1.events = EPOLLIN;ev1.data.fd = sock1;epoll_ctl(epfd, EPOLL_CTL_ADD, sock1, &ev1);ev2.events = EPOLLIN;ev2.data.fd = sock2;epoll_ctl(epfd, EPOLL_CTL_ADD, sock2, &ev2);epoll_event events[MAX_EVENTS];while (true) {if (false == g_running){std::cout << "recv thread end here by signal " << std::endl;return;}int nfds = epoll_wait(epfd, events, MAX_EVENTS, 1000);if (nfds == -1) {if (errno == EINTR) continue; // 被信號中斷，可忽略std::cout << "recv thread end, epoll_wait err " << std::endl;perror("epoll_wait");break;  }for (int i = 0; i < nfds; ++i) {int sockfd = events[i].data.fd;// 嘗試讀多次，直到沒數據，提高效率; 但是也要防止一個fd上數據太多，造成其他的句柄餓死int read_count = 0;while (read_count < MAX_READ_PER_FD) {read_count ++;Packet* packet = pool.acquire();socklen_t addr_len = sizeof(packet->addr_from);ssize_t len = recvfrom(sockfd, packet->data, BUF_SIZE, 0,(sockaddr*)&packet->addr_from, &addr_len);if (len > 0) {packet->data_len = len;packet->addr_from_len = addr_len;packet->port_from = (sockfd == sock1) ? PORT1 : PORT2;// 轉換IP地址為字符串char ip_str[INET_ADDRSTRLEN];  // 存儲IPv4地址的緩沖區const char* ip_addr;// 判斷地址族（IPv4）if (packet->addr_from.sin_family == AF_INET) {// 轉換IPv4地址struct sockaddr_in* ipv4_addr = (struct sockaddr_in*)&packet->addr_from;ip_addr = inet_ntop(AF_INET, &(ipv4_addr->sin_addr), ip_str, INET_ADDRSTRLEN);}// 如需支持IPv6可添加以下代碼// else if (packet->addr_from.sa_family == AF_INET6) {//     struct sockaddr_in6* ipv6_addr = (struct sockaddr_in6*)&packet->addr_from;//     ip_addr = inet_ntop(AF_INET6, &(ipv6_addr->sin6_addr), ip_str, INET6_ADDRSTRLEN);// }else {ip_addr = "未知地址族";}// 放到同一個隊列中{std::lock_guard<std::mutex> lock(recv_mutex_1);recv_queue_1.push(packet);}recv_cv_1.notify_one();std::cout << "read data from port " << packet->port_from << " IP="<< ip_addr << std::endl;} // end if len > 0else {pool.release(packet);if (errno == EAGAIN || errno == EWOULDBLOCK) {// 緩沖區讀空了，退出循環break;} else {std::cerr << "read data error: " << strerror(errno) << std::endl;break;}}  // end of else  }// end 針對一個socket的循環讀取while } // end foreach(events)}// end of while(true)return;
}// 處理PORT1端口數據，echo 模擬
// 將要發送的數據放到隊列中
void process_port1(Packet * pkt){pkt->addr_to = pkt->addr_from;pkt->addr_to_len = pkt->addr_from_len;{std::lock_guard<std::mutex> send_lock(send_mutex_1);send_queue_1.push(pkt);}send_cv_1.notify_one();
}// 處理PORT1端口數據echo 模擬
void process_port2(Packet * pkt){pkt->addr_to = pkt->addr_from;pkt->addr_to_len = pkt->addr_from_len;{std::lock_guard<std::mutex> send_lock(send_mutex_1);send_queue_1.push(pkt);}send_cv_1.notify_one();
}// 處理收到數據的線程，這里只是簡單的拷貝到發送隊列中，做一個echo的邏輯
void process_thread() {while (true) {if (false == g_running){std::cout << "process thread end here by signal " << std::endl;return;}// 提高鎖的效率，一次性處理多個包，std::vector<Packet*> pkts1;// 使用條件變量等待{std::unique_lock<std::mutex> lock(recv_mutex_1);recv_cv_1.wait(lock, [] { return !recv_queue_1.empty() || !g_running; });// 喚醒后檢查一下是否退出if (!g_running.load() && recv_queue_1.empty()) return;size_t cnt = 0;while (!recv_queue_1.empty() && cnt < MAX_BATCH) { pkts1.push_back(recv_queue_1.front());recv_queue_1.pop();cnt++; }}// 處理這最多32個數據包，for (auto pkt : pkts1) {if (pkt->port_from == PORT1){process_port1(pkt);}else{process_port2(pkt);}}}// end of while 
}// 處理需要發送的數據
void send_thread(int sock1, int sock2) {while (true) {if (!g_running.load()) {std::cout << "send thread end here by signal " << std::endl;return;}std::vector<Packet*> pkts;// 批量取出發送隊列的數據{std::unique_lock<std::mutex> lock(send_mutex_1);// 這里的謂詞很重要，謂詞中一定要包含隊列不為空send_cv_1.wait(lock, []{return !send_queue_1.empty() || !g_running.load();});// 喚醒后檢查退出條件if (!g_running.load() && send_queue_1.empty())return;size_t cnt = 0;while (!send_queue_1.empty() && cnt < MAX_BATCH) {pkts.push_back(send_queue_1.front());send_queue_1.pop();cnt++;}}// 循環發送for (auto pkt : pkts) {if (pkt->port_from == PORT1) {sendto(sock1, pkt->data, pkt->data_len, 0, (sockaddr *)&pkt->addr_to, pkt->addr_to_len);} else {sendto(sock2, pkt->data, pkt->data_len, 0, (sockaddr *)&pkt->addr_to, pkt->addr_to_len);}pool.release(pkt);}}  // end of while
}// 設置為非阻塞模式端口
int set_nonblocking(int sockfd) {int flags = fcntl(sockfd, F_GETFL, 0);if (flags == -1) return -1;return fcntl(sockfd, F_SETFL, flags | O_NONBLOCK);
}// 打開2個端口進行監聽
int create_udp_socket(int port) {int sock = socket(AF_INET, SOCK_DGRAM, 0);if (sock < 0) {perror("socket create error");exit(1);}sockaddr_in addr{};addr.sin_family = AF_INET;addr.sin_addr.s_addr = INADDR_ANY;addr.sin_port = htons(port);if (bind(sock, (sockaddr *)&addr, sizeof(addr)) < 0) {std::cout << "binding on UDP port " << port << " error " << std::endl;perror("bind error");exit(1);}return sock;
}// 讓線程檢測這個標志退出
void signal_handler(int signum) {if (signum == SIGINT) {std::cout << "\nSIGINT received. Exiting here！" << std::endl;g_running.store(false);send_cv_1.notify_all();recv_cv_1.notify_all();}
}sockaddr_in buildSockaddr(const std::string& ip, uint16_t port) {sockaddr_in addr;memset(&addr, 0, sizeof(addr));         // 清零結構體addr.sin_family = AF_INET;              // IPv4 協議addr.sin_port = htons(port);            // 端口轉網絡字節序inet_pton(AF_INET, ip.c_str(), &addr.sin_addr); // 將IP字符串轉為二進制return addr;
}
/// @brief 入口函數
/// @return 0
int main() {// 注冊 Ctrl+C 信號signal(SIGINT, signal_handler);int sock1 = create_udp_socket(PORT1);int sock2 = create_udp_socket(PORT2);// 在 create_udp_socket 后調用：set_nonblocking(sock1);set_nonblocking(sock2);std::cout << "Listening on UDP ports " << PORT1 << " and " << PORT2 << std::endl;std::thread t_recv(recv_thread, sock1, sock2);std::vector<std::thread> proc_threads, send_threads;for (int i = 0; i < NUM_PROC_THREADS; ++i)proc_threads.emplace_back(process_thread);for (int i = 0; i < NUM_SEND_THREADS; ++i)send_threads.emplace_back(send_thread, sock1, sock2);// 主線程等信號while (g_running) {std::this_thread::sleep_for(std::chrono::milliseconds(100));}// 等待線程退出if (t_recv.joinable())t_recv.join();// 等待處理線程退出for (auto& t : proc_threads) {if (t.joinable())t.join();}// 等待發送線程退出for (auto& t : send_threads) {if (t.joinable())t.join();}// 關閉資源close(sock1);close(sock2);std::cout << "ssff app  Clean shutdown complete." << std::endl;return 0;
}

編譯以后，

可以使用nc測試一下;

echo "hello port1" | nc -u 127.0.0.1 4000
echo "hello port2" | nc -u 127.0.0.1 5000

或者寫一個python程序發送代碼，可以跨主機試試：

import socket
import time
import argparse
import sys
import threadingdef udp_handler(local_ip, local_port, target_ip, target_port, count, interval, data):"""單個套接字處理發送和接收：- 綁定本地4000端口- 向目標4000端口發送數據- 在同一個端口接收回顯數據"""# 創建UDP套接字并綁定本地4001端口（發送和接收共用）sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)try:# 綁定本地端口（關鍵：發送和接收都用這個端口）sock.bind((local_ip, local_port))print(f"? 已綁定本地端口 {local_ip}:{local_port}，開始發送和接收數據...")# 啟動接收線程（用同一個套接字）def receive_loop():while True:try:# 接收數據（阻塞等待）data_recv, addr = sock.recvfrom(1024)# 顯示接收內容try:message = data_recv.decode('utf-8')print(f"\n📥 收到來自 {addr} 的數據: {message}")except UnicodeDecodeError:print(f"\n📥 收到來自 {addr} 的數據（無法解碼）: {data_recv.hex()}")except Exception as e:print(f"\n? 接收出錯: {str(e)}")break# 啟動接收線程receive_thread = threading.Thread(target=receive_loop, daemon=True)receive_thread.start()# 發送數據print(f"\n📤 開始向 {target_ip}:{target_port} 發送數據（共 {count} 個）...")for i in range(count):# 用已綁定的套接字發送（源端口固定為4001）sock.sendto(data.encode('utf-8'), (target_ip, target_port))# 顯示發送進度sys.stdout.write(f"\r已發送 {i+1}/{count} 個數據包")sys.stdout.flush()# 最后一個包不等待if i < count - 1:time.sleep(interval)print("\n\n? 發送完成！繼續等待接收數據（按Ctrl+C退出）...")# 保持程序運行，等待接收while True:time.sleep(1)except KeyboardInterrupt:print("\n\n?? 用戶中斷程序")except Exception as e:print(f"\n? 程序出錯: {str(e)}")finally:sock.close()print("\n🔌 套接字已關閉")if __name__ == "__main__":port = 5000parser = argparse.ArgumentParser(description='UDP雙向通信工具（固定本地4000端口）')parser.add_argument('--local-ip', type=str, default='0.0.0.0',help='本地綁定IP，默認0.0.0.0（所有網卡）')parser.add_argument('--local-port', type=int, default=port,help='本地綁定端口，默認4000')parser.add_argument('--target-ip', type=str, default='192.168.228.129',help='目標IP地址，默認虛擬機IP')parser.add_argument('--target-port', type=int, default=port,help='目標端口，默認4001')parser.add_argument('--count', type=int, default=10,help='發送數據包數量，默認10個')parser.add_argument('--interval', type=float, default=1.0,help='發送間隔(秒)，默認1秒')parser.add_argument('--data', type=str, default='Hello UDP',help='發送的數據內容，默認"Hello UDP"')args = parser.parse_args()# 啟動UDP處理邏輯udp_handler(local_ip=args.local_ip,local_port=args.local_port,target_ip=args.target_ip,target_port=args.target_port,count=args.count,interval=args.interval,data=args.data)