1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

#ifndef BUFFER_H
#define BUFFER_H
#include ...
class Buffer {
public:
    /*构造和析构*/
    Buffer(int initBuffSize = 1024);
    ~Buffer() = default;

    /*buffer中可写，可读，可预留空间*/
    size_t WritableBytes() const;       
    size_t ReadableBytes() const ;
    size_t PrependableBytes() const;

    /*可读的第一个地址，确保可读长度，移动可写长度*/
    const char* Peek() const;
    void EnsureWriteable(size_t len);
    void HasWritten(size_t len);

    /*移动可读长度，读到末尾*/
    void Retrieve(size_t len);
    void RetrieveUntil(const char* end);

    /*读完所有，读写下表全部归零*/
    void RetrieveAll() ;
    std::string RetrieveAllToStr();

    /*写指针位置*/
    const char* BeginWriteConst() const;
    char* BeginWrite();

    /*对stackbuf的操作，添加str到缓冲区*/
    void Append(const std::string& str);
    void Append(const char* str, size_t len);
    void Append(const void* data, size_t len);
    void Append(const Buffer& buff);

    /*读写的交互接口*/
    ssize_t ReadFd(int fd, int* Errno);
    ssize_t WriteFd(int fd, int* Errno);

private:
    char* BeginPtr_();
    const char* BeginPtr_() const;
    void MakeSpace_(size_t len); // 扩展空间

    /*atomic是一种原子类型，保证多线程的情况下，安全高性能的执行程序更新变量*/
    std::vector<char> buffer_; // 存储实体
    std::atomic<std::size_t> readPos_; // 读缓冲区
    std::atomic<std::size_t> writePos_; // 写缓冲区
};

#endif //BUFFER_H

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

#ifndef LOG_H
#define LOG_H

#include ...
class Log {
public:
    /*初始化日志，日志保存路径，文件后缀，最大容量*/
    void init(int level, const char* path = "./log", const char* suffix =".log", int maxQueueCapacity = 1024);

    /*懒汉模式，局部静态变量法*/
    static Log* Instance();
    /*异步日志的写线程函数*/
    static void FlushLogThread();

    /*写日志*/
    void write(int level, const char *format,...);
    void flush();

    int GetLevel();
    void SetLevel(int level);
    bool IsOpen() { return isOpen_; }
    
private:
    Log();
    /*添加日志等级*/
    void AppendLogLevelTitle_(int level);
    virtual ~Log();
    /*异步写日志*/
    void AsyncWrite_();

private:
    static const int LOG_PATH_LEN = 256;
    static const int LOG_NAME_LEN = 256;
    static const int MAX_LINES = 50000;

    const char* path_; // 路径
    const char* suffix_; // 后缀

    int MAX_LINES_; // 最大日志行数

    int lineCount_; // 日志行数
    int toDay_; // 按照当天日期区分文件
    bool isOpen_;
 
    Buffer buff_; // 输出缓冲区
    int level_; // 日志等级
    bool isAsync_; // 是否异步

    FILE* fp_; // 打开log的文件指针
    std::unique_ptr<BlockDeque<std::string>> deque_; // 阻塞队列
    std::unique_ptr<std::thread> writeThread_; // 写线程的指针
    std::mutex mtx_; // 同步日志的锁
};

#define LOG_BASE(level, format, ...) \
    do {\
        Log* log = Log::Instance();\
        if (log->IsOpen() && log->GetLevel() <= level) {\
            log->write(level, format, ##__VA_ARGS__); \
            log->flush();\
        }\
    } while(0);

/*
* 四个宏定义，用于不同类型日志输出，也就是外部使用日志接口
* ...表示可变参数，_VA_ARGS_就是将...的值复制到这里
* ##的作用在于可变参数为0时，去掉前面的逗号
*/
#define LOG_DEBUG(format, ...) do {LOG_BASE(0, format, ##__VA_ARGS__)} while(0);
#define LOG_INFO(format, ...) do {LOG_BASE(1, format, ##__VA_ARGS__)} while(0);
#define LOG_WARN(format, ...) do {LOG_BASE(2, format, ##__VA_ARGS__)} while(0);
#define LOG_ERROR(format, ...) do {LOG_BASE(3, format, ##__VA_ARGS__)} while(0);

#endif //LOG_H

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

#ifndef THREADPOOL_H
#define THREADPOOL_H

#include...
class ThreadPool {
public:
    /*explicit：
    *主要用于防止隐式类型转换.当构造函数被声明为explicit时，
    *编译器将禁止使用该构造函数进行隐式的类型转换。
    *作用是防止编译器在某些情况下进行自动类型转换，导致不可预期的行为。
    */
    explicit ThreadPool(size_t threadCount = 8): pool_(std::make_shared<Pool>()) {
            assert(threadCount > 0); // 确保线程数目大于0

            for(size_t i = 0; i < threadCount; i++) {
                std::thread([pool = pool_] {
                    /*该线程的互斥锁*/
                    std::unique_lock<std::mutex> locker(pool->mtx);
                    while(true) {
                        if(!pool->tasks.empty()) {/*池子不为空时*/
                            /*从队列中获取一个任务然后执行
                            * 执行时可以加锁
                            * 获取任务时采用move可以节省资源
                            */
                            auto task = std::move(pool->tasks.front());
                            pool->tasks.pop();
                            locker.unlock();
                            task();
                            locker.lock();
                        } /*线程池关闭时跳出*/
                        else if(pool->isClosed) break;
                        /*线程池为空时，利用条件变量等待*/
                        else pool->cond.wait(locker);
                    }
                }).detach();/*线程类的一个成员函数，用于将线程和创建的thread对象分离*/
            }
    }
    ThreadPool() = default;
    ThreadPool(ThreadPool&&) = default;   
    ~ThreadPool() {
        if(static_cast<bool>(pool_)) {
            {
                /*c++11引入的RALL资源获取即初始化风格的做机制*/
                /*当前作用域自动加锁，作用域结束自动释放锁*/
                std::lock_guard<std::mutex> locker(pool_->mtx);
                pool_->isClosed = true; // 关闭线程池
            }
            pool_->cond.notify_all(); // 唤醒所有线程退出
        }
    }

    template<class F>
    void AddTask(F&& task) {
        {
            /*完美转发添加任务*/
            std::lock_guard<std::mutex> locker(pool_->mtx);
            pool_->tasks.emplace(std::forward<F>(task));
        } // 唤醒一个线程
        pool_->cond.notify_one();
    }
private:
    struct Pool {
        std::mutex mtx; // 互斥锁，保护共享数据
        std::condition_variable cond; // 条件变量，管理任务的等待和通知机制
        bool isClosed; // 标志线程池是否关闭
        std::queue<std::function<void()>> tasks; // 任务队列，存放待执行的任务
    };
    std::shared_ptr<Pool> pool_;
};
#endif //THREADPOOL_H

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

#ifndef SQLCONNPOOL_H
#define SQLCONNPOOL_H

#include...
class SqlConnPool {
public:
    /*单例模式*/
    static SqlConnPool *Instance();

    /*得到连接和释放连接*/
    MYSQL *GetConn();
    void FreeConn(MYSQL * conn);
    int GetFreeConnCount();

    void Init(const char* host, int port, const char* user,const char* pwd, const char* dbName, int connSize);
    void ClosePool();

private:
    SqlConnPool();
    ~SqlConnPool();

    int MAX_CONN_; // 最大连接数
    int useCount_, freeCount_;

    std::queue<MYSQL *> connQue_; // 连接队列
    std::mutex mtx_; // 互斥锁
    sem_t semId_; // 信号量
};

/* 资源在对象构造初始化 资源在对象析构时释放*/
class SqlConnRAII {
public:
    SqlConnRAII(MYSQL** sql, SqlConnPool *connpool) {
        assert(connpool);
        // 建立连接
        *sql = connpool->GetConn();
        sql_ = *sql; connpool_ = connpool;
    }
    
    ~SqlConnRAII() {
        // 释放连接
        if(sql_) { connpool_->FreeConn(sql_); }
    }
    
private:
    MYSQL *sql_; SqlConnPool* connpool_;
};

#endif // SQLCONNPOOL_H

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

#ifndef HTTP_RESPONSE_H
#define HTTP_RESPONSE_H

#include...

class HttpResponse {
public:
    HttpResponse();
    ~HttpResponse();

    void Init(const std::string& srcDir, std::string& path, bool isKeepAlive = false, int code = -1);
    void MakeResponse(Buffer& buff); // 该函数中会确定状态码，调用private中的三个添加函数
    void UnmapFile();
    char* File();
    size_t FileLen() const;
    void ErrorContent(Buffer& buff, std::string message);
    int Code() const { return code_; }

private:
    void AddStateLine_(Buffer &buff); // 添加状态行
    void AddHeader_(Buffer &buff); // 添加消息报头
    void AddContent_(Buffer &buff); // 添加响应正文

    void ErrorHtml_();
    std::string GetFileType_();

    int code_; // 状态码
    bool isKeepAlive_; // http保活机制

    /*生成状态码*/
    std::string path_;
    std::string srcDir_; // 源目录
    
    char* mmFile_;  // 映射文件
    struct stat mmFileStat_; // 文件状态信息

    static const std::unordered_map<std::string, std::string> SUFFIX_TYPE; // 后缀类型
    static const std::unordered_map<int, std::string> CODE_STATUS; // http状态码
    static const std::unordered_map<int, std::string> CODE_PATH; // 400，403，404对应的错误路径
};


#endif //HTTP_RESPONSE_H

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

void HttpResponse::MakeResponse(Buffer& buff) {
    /* 判断请求的资源文件 */
    if(stat((srcDir_ + path_).data(), &mmFileStat_) < 0 || S_ISDIR(mmFileStat_.st_mode)) {
        code_ = 404; // 文件状态信息小于0或者文件是一个目录
    }
    else if(!(mmFileStat_.st_mode & S_IROTH)) {
        code_ = 403; // 文件没有其它用户的读取权限时
    }
    else if(code_ == -1) { 
        code_ = 200; // 文件成功找到
    }
    ErrorHtml_(); // 根据 HTTP 响应码查找并设置相应的错误页面路径
    AddStateLine_(buff); // buff.Append(HTTP/1.1 状态码 状态)
    AddHeader_(buff); // 添加connection和content-type
    AddContent_(buff); // 映射内存，追加文件长度，错误页面单独处理
}

1
2
3

    // write核心逻辑在于iov中的多缓冲区的写入。
    // 将writev一次性写入文件描述符中，使用ET写数据，动态调整缓冲区指针和长度
    // ET模式下，循环继续，直到全部数据写出或者数据小于设定阈值。

1
2

// 循环读取到没有数据为止，
// ET模式下，文件描述符的事件只触发一次，除非有新的数据到达

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

#ifndef HTTP_CONN_H
#define HTTP_CONN_H

#include...

class HttpConn {
public:
    HttpConn();
    ~HttpConn();

    void init(int sockFd, const sockaddr_in& addr);

    // 将请求报文的内容读取到读缓冲区里面
    ssize_t read(int* saveErrno);
    // 将从缓冲区中得到的响应报文写到浏览器
    ssize_t write(int* saveErrno);

    void Close();

    int GetFd() const;
    int GetPort() const;
    const char* GetIP() const;
    sockaddr_in GetAddr() const;
    
    // 解析请求报文和生成响应报文
    bool process();

    int ToWriteBytes() { 
        return iov_[0].iov_len + iov_[1].iov_len; 
    }

    bool IsKeepAlive() const {
        return request_.IsKeepAlive();
    }

    static bool isET;
    static const char* srcDir;
    static std::atomic<int>;
    
private:
   
    int fd_;
    struct  sockaddr_in addr_;
    bool isClose_;
    
    int iovCnt_;
    struct iovec iov_[2];
    
    Buffer readBuff_; // 读缓冲区
    Buffer writeBuff_; // 写缓冲区

    HttpRequest request_;
    HttpResponse response_;
};


#endif //HTTP_CONN_H

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

#ifndef HEAP_TIMER_H
#define HEAP_TIMER_H

#include...

typedef std::function<void()> TimeoutCallBack;
typedef std::chrono::high_resolution_clock Clock;
typedef std::chrono::milliseconds MS;
typedef Clock::time_point TimeStamp;

struct TimerNode {
    int id;
    TimeStamp expires; // 超时时间点
    TimeoutCallBack cb; // 超时回调函数
    bool operator<(const TimerNode& t) {
        return expires < t.expires;
    }
};
/*基于对最小堆的定时器类*/
class HeapTimer {
public:
    HeapTimer() { heap_.reserve(64); }
    ~HeapTimer() { clear(); }
    
    void adjust(int id, int newExpires)
    void add(int id, int timeOut, const TimeoutCallBack& cb); // 添加定时器结点，ID不存在则直接插入然后调整；id存在则更新超时时间
    void doWork(int id);// 触发回调函数，删除该定时器
    void clear(); // 清空

    void tick(); // 检查堆顶元素是否超时，超时则回调删除
    void pop(); // 删除最早超时的定时器
    int GetNextTick(); // 获取下一个超时定时器的剩余时间

private:
    void del_(size_t i);  // 删除定时器
    void siftup_(size_t i); // 向上调整
    bool siftdown_(size_t index, size_t n); // 向下调整
    void SwapNode_(size_t i, size_t j); // 交换两个结点位置

    std::vector<TimerNode> heap_;
    std::unordered_map<int, size_t> ref_; // 定时器ID到堆中结点的索引，便于快速查找定时器
};

#endif //HEAP_TIMER_H

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

#ifndef EPOLLER_H
#define EPOLLER_H

#include...
/*基于epoll的事件处理类*/
class Epoller {
public:
    explicit Epoller(int maxEvent = 1024);

    ~Epoller();

    /*添加事件，修改事件，删除事件epoll_ctl*/
    bool AddFd(int fd, uint32_t events);
    bool ModFd(int fd, uint32_t events);
    bool DelFd(int fd);

    /*等待事件发生，epoll_wait*/
    int Wait(int timeoutMs = -1);

    /*第i个事件的类型or文件描述符*/
    int GetEventFd(size_t i) const;
    uint32_t GetEvents(size_t i) const;
        
private:
    int epollFd_;

    std::vector<struct epoll_event> events_;    
};

#endif //EPOLLER_H

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

/* 设置套接字
* SO_LINGER：
确保当服务器关闭时，未发送的数据能够尽量发送完毕，
提供一个“优雅关闭”的机制。
* SO_REUSEADDR：
允许多个套接字绑定到同一个端口，但同一时间只能有一个套接字接收数据
解决TIME_WAIT状态导致的端口绑定问题。
允许端口复用，避免端口被占用，
确保服务器能够快速重启并继续监听同一个端口。
* SO_REUSEPORT： 
允许多个套接字同时绑定到同一个端口，
内核会将收到的数据包分发给不同的套接字，实现真正的负载均衡。
*/

/* 确立事件模式
* EPOLLRDHUP:检测到对方关闭连接时触发，监控对端连接关闭的情况
* EPOLLONSHOT:事件触发后自动取消，防止多线程竞争
* EPOLLET:变韵出发，只有状态发生变化才会通知
*/