【C++】开源：事件驱动库libevent配置使用

DevFrank

发布于 2024-07-24 15:40:42

690

发布于 2024-07-24 15:40:42

文章被收录于专栏：C++开发学习交流

😏1. 项目介绍

项目Github地址：https://github.com/libevent/libevent

官网：https://libevent.org/

libevent是一个开源的事件驱动库，用于开发高性能、可扩展的网络应用程序。它提供了跨平台的抽象接口，使开发人员能够使用事件回调来管理网络连接、定时器和信号等事件。

以下是libevent库的一些主要特点和功能：

1.事件驱动：libevent使用事件驱动的方式处理网络和I/O操作。它基于回调机制，可以处理各种事件，包括网络连接、读写操作、定时器等。
2.跨平台支持：libevent可以在多个平台上运行，包括Linux、Unix、Windows等。它封装了不同操作系统的底层API，使开发人员能够在不同平台上实现相同的功能。
3.高性能：libevent被设计成高效的事件通知引擎，它使用了高效的I/O多路复用技术（如epoll、kqueue等），能够同时处理大量的并发连接和事件。
4.可扩展性：libevent提供了可扩展的接口和机制，开发人员可以自定义事件的处理方式，并添加自定义的事件类型。它还支持多线程和多进程编程模型，方便实现并发处理。
5.支持多种协议：libevent支持多种网络协议，包括TCP、UDP、SSL等。它提供了相应的API和功能，以便开发人员轻松地构建各种网络应用程序。

libevent 是一个功能强大的事件驱动网络库，广泛应用于构建高性能的服务器程序、代理、负载均衡器等网络应用。它提供了简洁的接口和丰富的功能，使开发人员能够轻松地编写高效、可扩展的网络应用程序。

😊2. 环境配置

下面进行环境配置：

# apt安装
sudo apt install libevent-dev
# 查看版本（ubuntu默认2.1.8-stable）
pkg-config --modversion libevent

# 源码安装
tar -zxvf libevent-2.0.22-stable.tar.gz
cd libevent-2.0.22-stable/
./configure
make
sudo make install

😆3. 使用说明

定时器事件示例：

#include <iostream>
#include <event2/event.h>

void timerCallback(evutil_socket_t fd, short events, void* arg) {
    std::cout << "Timer expired" << std::endl;

    // 停止事件循环
    event_base* base = static_cast<event_base*>(arg);
    event_base_loopbreak(base);
}

int main() {
    // 创建事件基础
    event_base* base = event_base_new();
    if (!base) {
        std::cerr << "Failed to create event base" << std::endl;
        return 1;
    }
    std::cout << "start event..." << std::endl;

    // 创建定时器事件
    timeval tv = {2, 0};  // 2 秒
    event* timerEvent = evtimer_new(base, timerCallback, base);
    if (!timerEvent) {
        std::cerr << "Failed to create timer event" << std::endl;
        return 1;
    }

    // 添加定时器事件
    if (evtimer_add(timerEvent, &tv) < 0) {
        std::cerr << "Failed to add timer event" << std::endl;
        return 1;
    }

    // 启动事件循环
    event_base_dispatch(base);

    // 清理资源
    event_free(timerEvent);
    event_base_free(base);

    return 0;
}

编译运行：

g++ -o main main.cpp -levent
./main

基于libevent的线程池示例：

#include <event2/event.h>
#include <event2/thread.h>
#include <iostream>
#include <vector>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <thread>

// 任务结构体
struct Task {
    std::function<void()> function;

    Task(const std::function<void()>& f) : function(f) {}
};

// 线程池类
class ThreadPool {
public:
    ThreadPool(int numThreads) : stop(false) {
        // 初始化libevent线程支持
        evthread_use_pthreads();

        for (int i = 0; i < numThreads; ++i) {
            threads.emplace_back([this] {
                event_base* base = event_base_new();
                if (!base) {
                    std::cerr << "Failed to create event base." << std::endl;
                    return;
                }

                // 创建事件来触发任务执行
                event* ev = event_new(base, -1, EV_PERSIST, [](evutil_socket_t fd, short events, void* arg) {
                    ThreadPool* threadPool = static_cast<ThreadPool*>(arg);
                    threadPool->executeTask();
                }, this);

                timeval delay = { 0, 1000 }; // 每隔1毫秒触发一次事件
                event_add(ev, &delay);

                // 执行事件循环
                event_base_dispatch(base);

                event_free(ev);
                event_base_free(base);
            });
        }
    }

    ~ThreadPool() {
        {
            std::unique_lock<std::mutex> lock(taskMutex);
            stop = true;
        }

        taskCondition.notify_all();

        for (auto& thread : threads) {
            thread.join();
        }
    }

    template<class F>
    void enqueue(F&& f) {
        {
            std::unique_lock<std::mutex> lock(taskMutex);
            tasks.emplace(new Task(std::forward<F>(f)));
        }

        taskCondition.notify_one();
    }

private:
    std::vector<std::thread> threads;
    std::queue<Task*> tasks;

    std::mutex taskMutex;
    std::condition_variable taskCondition;

    bool stop;

    void executeTask() {
        Task* task = nullptr;

        {
            std::unique_lock<std::mutex> lock(taskMutex);

            if (tasks.empty()) {
                return;
            }

            task = tasks.front();
            tasks.pop();
        }

        task->function();

        delete task;
    }
};

// 示例使用
void taskFunction(int id) {
    std::cout << "Task " << id << " is being executed." << std::endl;
}

int main() {
    const int numThreads = 4;
    ThreadPool threadPool(numThreads);
    std::cout << "Create " << numThreads << " pools..." << std::endl;

    // 提交任务到线程池
    for (int i = 0; i < 10; ++i) {
        threadPool.enqueue([i] {
            taskFunction(i);
        });
    }

    // 等待所有任务完成
    std::this_thread::sleep_for(std::chrono::seconds(3));
    std::cout << "Task end!!!" << std::endl;

    return 0;
}

编译运行：

g++ -o main main.cpp -levent -lpthread -levent_pthreads

TCP服务端和客户端示例：

// server.cpp
#include <iostream>
#include <cstring>
#include <event2/event.h>
#include <event2/listener.h>
#include <event2/bufferevent.h>

void readCallback(bufferevent* bev, void* arg) {
    char data[1024];
    int n;
    while ((n = bufferevent_read(bev, data, sizeof(data))) > 0) {
        std::cout << "Received message: " << std::string(data, n);
        // 在这里可以处理接收到的消息
    }
}

void writeCallback(bufferevent* bev, void* arg) {
    // 在这里可以处理发送完成的逻辑
}

void eventCallback(bufferevent* bev, short events, void* arg) {
    if (events & BEV_EVENT_ERROR) {
        std::cerr << "Error occurred" << std::endl;
    }
    if (events & (BEV_EVENT_EOF | BEV_EVENT_ERROR)) {
        std::cout << "Connection closed" << std::endl;
        bufferevent_free(bev);
    }
}

void acceptCallback(evconnlistener* listener, evutil_socket_t fd, sockaddr* address, int socklen, void* arg) {
    std::cout << "Accepted connection" << std::endl;

    // 创建 bufferevent 对象
    event_base* base = evconnlistener_get_base(listener);
    bufferevent* bev = bufferevent_socket_new(base, fd, BEV_OPT_CLOSE_ON_FREE);
    if (!bev) {
        std::cerr << "Failed to create bufferevent" << std::endl;
        evutil_closesocket(fd);
        return;
    }

    // 设置回调函数
    bufferevent_setcb(bev, readCallback, writeCallback, eventCallback, nullptr);

    // 启用读写事件
    bufferevent_enable(bev, EV_READ | EV_WRITE);
}

int main() {
    // 创建事件基础
    event_base* base = event_base_new();
    if (!base) {
        std::cerr << "Failed to create event base" << std::endl;
        return 1;
    }

    // 创建监听器
    sockaddr_in sin;
    std::memset(&sin, 0, sizeof(sin));
    sin.sin_family = AF_INET;
    sin.sin_addr.s_addr = htonl(INADDR_ANY);
    sin.sin_port = htons(12345);

    evconnlistener* listener = evconnlistener_new_bind(
        base,
        acceptCallback,
        nullptr,
        LEV_OPT_REUSEABLE | LEV_OPT_CLOSE_ON_FREE,
        -1,
        (struct sockaddr*)&sin,
        sizeof(sin)
    );
    if (!listener) {
        std::cerr << "Failed to create listener" << std::endl;
        return 1;
    }

    std::cout << "Server is running..." << std::endl;

    // 启动事件循环
    event_base_dispatch(base);

    // 清理资源
    evconnlistener_free(listener);
    event_base_free(base);

    return 0;
}

// client.cpp
#include <iostream>
#include <cstring>
#include <arpa/inet.h>
#include <event2/event.h>
#include <event2/bufferevent.h>

void readCallback(bufferevent* bev, void* arg) {
    char data[1024];
    int n;
    while ((n = bufferevent_read(bev, data, sizeof(data))) > 0) {
        std::cout << "Received message: " << std::string(data, n);
        // 在这里可以处理接收到的消息
    }
}

void writeCallback(bufferevent* bev, void* arg) {
    // 在这里可以处理发送完成的逻辑
}

void eventCallback(bufferevent* bev, short events, void* arg) {
    if (events & BEV_EVENT_CONNECTED) {
        std::cout << "Connected to server" << std::endl;
        // 在这里可以处理连接成功的逻辑
    }
    else if (events & BEV_EVENT_ERROR) {
        std::cerr << "Error occurred" << std::endl;
    }
    else if (events & (BEV_EVENT_EOF | BEV_EVENT_ERROR)) {
        std::cout << "Connection closed" << std::endl;
        bufferevent_free(bev);
    }
}

int main() {
    // 创建事件基础
    event_base* base = event_base_new();
    if (!base) {
        std::cerr << "Failed to create event base" << std::endl;
        return 1;
    }

    // 创建 bufferevent 对象
    bufferevent* bev = bufferevent_socket_new(base, -1, BEV_OPT_CLOSE_ON_FREE);
    if (!bev) {
        std::cerr << "Failed to create bufferevent" << std::endl;
        event_base_free(base);
        return 1;
    }

    // 设置回调函数
    bufferevent_setcb(bev, readCallback, writeCallback, eventCallback, nullptr);

    // 连接到服务器
    sockaddr_in sin;
    std::memset(&sin, 0, sizeof(sin));
    sin.sin_family = AF_INET;
    sin.sin_addr.s_addr = inet_addr("127.0.0.1");
    sin.sin_port = htons(12345);

    if (bufferevent_socket_connect(bev, (struct sockaddr*)&sin, sizeof(sin)) < 0) {
        std::cerr << "Failed to connect to server" << std::endl;
        bufferevent_free(bev);
        event_base_free(base);
        return 1;
    }

    // 启动事件循环
    event_base_dispatch(base);

    // 清理资源
    bufferevent_free(bev);
    event_base_free(base);

    return 0;
}

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2023-12-07，如有侵权请联系 cloudcommunity@tencent.com 删除

配置