当前位置：首页 > news >正文

【libevent多线程服务器】--UDP

news 来源：原创 2024/9/20 23:48:58

大佬多年前写的：link
基于大佬的，改成了windows版的，用了更新的Libevent API
后面把tcp多线程服务器写好也会发出来，若有误/改进点，往指正

#include <iostream>
#include <string>
#include <thread>
#include <event2/event.h>
#include <event.h>
#include <event2/thread.h>
#include <windows.h>
#include <winsock2.h>
#include <mutex>
#include <vector>
#include <atomic>
#include <future> // 用于std::async
#pragma comment(lib, "Iphlpapi.lib")
#pragma comment(lib, "ws2_32.lib")
#pragma comment(lib, "wsock32.lib")
#pragma comment(lib, "event.lib")
#pragma comment(lib, "event_core.lib")
#pragma comment(lib, "event_extra.lib")
#pragma comment(lib, "bcrypt.lib")
// RAII风格的类，用于管理WSA的初始化和清理
class WSAInitializer {
public:WSAInitializer() {if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0) {throw std::runtime_error("WSAStartup error");}}~WSAInitializer() {WSACleanup();}private:WSADATA wsaData{};
};int init_count = 0;
std::mutex init_lock;typedef struct {std::thread thread; // 线程对象event_base *base;      // 线程使用的libevent句柄event *notify_event;    // 通知管道的监听事件
} mythread;void worker_libevent(void *arg) {auto *p = (mythread *)arg;std::unique_lock<std::mutex> guard(init_lock);init_count++;guard.unlock();event_base_loop(p->base, 0);
}void process(evutil_socket_t fd, short which, void *arg) {auto *p = (mythread *)arg;char buffer[100] = {0}; // 使用字符串构造函数清空缓冲区int ilen = recv(fd, buffer, sizeof(buffer), 0);if (ilen <= 0) { // 检查错误和连接关闭std::cerr << "recv error or connection closed" << std::endl;return;}std::cout << "I am in the thread: " << p->thread.get_id() << std::endl;std::cout << "read num is: " << ilen << std::endl;std::cout << "the buffer: " << buffer << std::endl;
}void setup_thread(mythread *p, int fd) {p->base = event_base_new();p->notify_event=event_new(p->base, fd, EV_READ | EV_PERSIST, process, p);event_add(p->notify_event, 0);
}int main() {WSAInitializer wsainit; // 确保WSA的初始化和清理SOCKET sock = socket(AF_INET, SOCK_DGRAM, 0);if (sock == INVALID_SOCKET) {std::cerr << "socket error" << std::endl;return -1;}sockaddr_in server_addr{};server_addr.sin_family = AF_INET;server_addr.sin_addr.s_addr = INADDR_ANY; // 监听所有地址server_addr.sin_port = htons(19870);if (bind(sock, (sockaddr *)&server_addr, sizeof(server_addr)) == SOCKET_ERROR) {std::cerr << "bind error" << std::endl;return -1;}const int thread_num = 10;std::vector<mythread> g_threads(thread_num);for (int i = 0; i < thread_num; i++) {setup_thread(&g_threads[i], sock);std::cout<<"create thread: "<<i<<std::endl;}for (int i = 0; i < thread_num; i++) {g_threads[i].thread = std::thread(worker_libevent, &g_threads[i]);std::cout<<" thread: "<<i<<std::endl;}std::cout<<"init_count: "<<init_count<<std::endl;std::unique_lock<std::mutex> guard(init_lock);printf("IN THE MAIN LOOP\n");// 使用std::async和future来回收线程资源，提高性能std::vector<std::future<void>> futures;for (auto &t : g_threads) {futures.push_back(std::async(std::launch::async, [&](){t.thread.join();}));}// 等待所有线程结束for (auto &f : futures) {f.get();}closesocket(sock);return 0;
}