epoll+线程池 封装

dwsocket.h

#pragma once

#include <string>
#include <sys/epoll.h>
#include <pthread.h>
#include<condition_variable>
#include <queue>
#include<stdlib.h>

namespace dw
{

    using namespace std;

    class Addr
    {
    public:
        uint16_t port;
        string ip;
    public:
        Addr(string ip, uint16_t port);
        Addr();
        ~Addr();
        friend ostream& operator<<(ostream &out, const Addr &addr);
    private:
    };

    class Socket
    {
    public:
        int socketFd;
    public:
        Socket();
        ~Socket();
        int read(void *data);
    private:
    };

    typedef void(*ServerSocketCallBack)(Socket*);

    /*
     * 服务端Socket 完全异步
     */
    class ServerSocket
    {
    public:
        Addr *addr = NULL;
        int maxListenNum = 128;//最大监听数,libevent也是128
        int epollSize = 65535;//epoll 的最大监听数
        int maxEvents = 1024;
        int maxThread = 1; //最大线程数(socket开启时,会会根据cpu核心数,重新初始化该值)
        ServerSocketCallBack callBack = NULL;
    public:
        ServerSocket(uint16_t port);
        ~ServerSocket();
        int start();
    private:
        int socketfd = -1;
        int epollfd = -1;
        epoll_event* events;
        int *threadArry = NULL;
        queue<int> taskQueue; //任务队列
        mutex taskQueueLock; //任务队列锁,防止两个线程读取同一个socket
    private:
        void initAddr(uint16_t port);
        int initSocket();
        int initEpoll();
        void setNonBlock(int fd); //设置非阻塞socket
        void initMaxThread();
        void loopWait();
        void addSocketFd(int fd);
        int initThreadPool(); //初始化线程池
        static void * threadMain(void* arg);//线程主函数
    };

}

Addr.cpp

#include "dwsocket.h"

#include<iostream>

namespace dw
{

    Addr::Addr()
    {
    }

    Addr::Addr(string ip, uint16_t port)
    {
        this->port = port;
        this->ip = ip;
    }

    Addr::~Addr()
    {
    }

    ostream& operator<<(ostream &out, const Addr &addr)
    {
        out << addr.ip << ":" << addr.port;
        return out;
    }

}

ServerSocket.cpp

#include "dwsocket.h"

#include<iostream>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <sys/sysinfo.h>
#include <unistd.h>

namespace dw
{

    using namespace std;

    ServerSocket::ServerSocket(uint16_t port)
    {
        initAddr(port);
        initMaxThread();
    }

    ServerSocket::~ServerSocket()
    {
        delete addr;

        if (threadArry != NULL) {
            for (int i = 0; i < maxThread; i++) {
                pthread_exit(&(threadArry[i]));
            }
        }

        shutdown(socketfd, SHUT_RDWR);
    }

    void ServerSocket::initMaxThread()
    {
        int cpuCoreNum = get_nprocs();
        maxThread = cpuCoreNum * 2;
    }

    void ServerSocket::initAddr(uint16_t port)
    {
        addr = new Addr("127.0.0.1", port);
    }

    int ServerSocket::initSocket()
    {
        if (maxListenNum <= 0)
        {
            cout << "Error: maxListenNum 必须大于0 " << endl;
            return -1;
        }

        socketfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
        if (socketfd < 0)
        {
            cout << "Error: socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)" << endl;
            return -1;
        }

        //设置socket
        setNonBlock(socketfd);
        int opt = 1;
        setsockopt(socketfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));

        struct sockaddr_in server_addr;
        server_addr.sin_family = AF_INET;
        server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
        server_addr.sin_port = htons(addr->port);

        int err = bind(socketfd, (struct sockaddr*) &server_addr, sizeof(server_addr));
        if (err < 0)
        {
            cout << "Error: bind(socketfd, (struct sockaddr*) &server_addr, sizeof(server_addr))" << endl;
            return -1;
        }

        err = listen(socketfd, maxListenNum);
        if (err < 0)
        {
            cout << "Error: listen(socketfd, maxListenNum)" << endl;
            return -1;
        }

        return 0;
    }

    void ServerSocket::setNonBlock(int fd)
    {
        int fl = fcntl(socketfd, F_GETFL);
        fcntl(socketfd, F_SETFL, fl | O_NONBLOCK);
    }

    int ServerSocket::initEpoll()
    {

        epollfd = epoll_create(epollSize);
        if (epollfd < 0)
        {
            cout << "Error: epoll_create(maxEvents)" << endl;
            return -1;
        }

        events = new epoll_event[maxEvents];
        addSocketFd(socketfd);

        return 0;
    }

    void ServerSocket::loopWait()
    {

        struct sockaddr_in clientaddr;
        socklen_t clilen = sizeof(clientaddr);

        while (1)
        {
            int num = epoll_wait(epollfd, events, maxEvents, -1); //返回活跃用户个数
            for (int i = 0; i < num; i++)
            {
                int connfd = -1;
                if (events[i].data.fd == socketfd)
                {
                    cout << "有新用户连接" << endl;
                    connfd = accept(socketfd, (struct sockaddr *) &clientaddr, &clilen);
                    if (connfd < 0)
                    {
                        cout << "连接失败" << endl;
                        continue;
                    }
                    setNonBlock(connfd);
                }
                else if (events[i].events&EPOLLIN)
                {
                    cout << "有用户发送数据" << endl;
                    if ((connfd = events[i].data.fd) < 0)
                    {
                        cout << "连接失败" << endl;
                        continue;
                    }
                    taskQueueLock.lock();
                    taskQueue.push(connfd);
                    taskQueueLock.unlock();

                }
                else if (events[i].events&EPOLLOUT)
                {
                    //以前的连接,有数据写出
                    cout << "以前的连接 写出" << endl;
                }
                else
                {
                    cout << "其他" << endl;
                    continue;
                }
                addSocketFd(connfd);
            }
        }
    }

    void ServerSocket::addSocketFd(int fd)
    {
        struct epoll_event event;
        event.data.fd = fd;
        event.events = EPOLLIN | EPOLLET;
        epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &event);
    }

    int ServerSocket::initThreadPool()
    {
        pthread_t threads[maxThread];
        threadArry = (int*)threads;

        for (int i = 0; i < maxThread; i++)
        {
            int ret = pthread_create(&threads[i], NULL, threadMain, this);
            if (ret < 0)
            {
                cout << "Error: initThreadPool()" << endl;
                return -1;
            }
        }
        return 0;
    }

    void * ServerSocket::threadMain(void* arg)
    {
        ServerSocket *socket = (ServerSocket*)arg;
        while (true)
        {
            int socketFd = 0;
            socket->taskQueueLock.lock();
            if (!socket->taskQueue.empty()) {
                socketFd = socket->taskQueue.front();
                socket->taskQueue.pop();
            }
            socket->taskQueueLock.unlock();

            if (socketFd <= 0)
            {
                usleep(100000);
            }
            else
            {
                Socket client;
                client.socketFd = socketFd;
                socket->callBack(&client);
            }
        }
        return NULL;
    }

    int ServerSocket::start()
    {
        if (callBack == NULL) {
            cout << "Error: callBack is NULL" << endl;
            return -1;
        }

        int err = initThreadPool();
        if (err < 0)
        {
            cout << "Error: initThreadPool()" << endl;
            return -1;
        }

        err = initSocket();
        if (err < 0)
        {
            cout << "Error: initSocket()" << endl;
            return -1;
        }

        err = initEpoll();
        if (err < 0)
        {
            cout << "Error: initEpoll()" << endl;
            return -1;
        }
        loopWait();
        return 0;
    }
}

Socket.cpp

#include "dwsocket.h"

namespace dw {

    Socket::Socket() {
    }

    Socket::~Socket() {
    }

    int Socket::read(void *data) {
        return -1;
    }

}

main.cpp


#include "dwsocket.h"
#include <iostream>

using namespace dw;
using namespace std;

void test(Socket *socket) {
    cout << "test 有用户连接进来" << endl;
    socket->read(NULL);
}

int main() {
    ServerSocket server(10000);

    static ServerSocketCallBack callback = test;
    server.callBack = callback;

    server.start();
    return 0;
}

编译时需要添加 -lpthread

©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 203,456评论 5赞 477
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 85,370评论 2赞 381
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 150,337评论 0赞 337
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 54,583评论 1赞 273
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 63,596评论 5赞 365
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,572评论 1赞 281
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 37,936评论 3赞 395
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,595评论 0赞 258
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 40,850评论 1赞 297
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,601评论 2赞 321
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,685评论 1赞 329
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,371评论 4赞 318
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 38,951评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,934评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,167评论 1赞 259
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 43,636评论 2赞 349
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,411评论 2赞 342

推荐阅读更多精彩内容

  • 2018-06-11
    mean to add the formatted="false" attribute?.[ 46% 47325/...
    ProZoom阅读 2,689评论 0赞 3
  • 面试题
    Swift1> Swift和OC的区别1.1> Swift没有地址/指针的概念1.2> 泛型1.3> 类型严谨 对...
    cosWriter阅读 11,082评论 1赞 32
  • 编译链接
    转自http://blog.csdn.net/navyhu/article/details/47023317理解链...
    扎Zn了老Fe阅读 1,425评论 0赞 0
  • 18、python网络编程
    网络编程 一.楔子 你现在已经学会了写python代码，假如你写了两个python文件a.py和b.py，分别去运...
    go以恒阅读 1,982评论 0赞 6
  • 神交久到相见欢
    ——我市党外干部队伍建设工作从幕后渐至台前 党外干部队伍建设工作的诸多环节，体现在多方面的工作之中。如，...
    停一秒想笑阅读 545评论 0赞 1