项目github https://github.com/dreamyouxi/LiteHttp
文件ThreadPool.cpp
#include "ThreadPool.h"
#include "winsock.h"
#include "stdlib.h"
#include "stdio.h"
#include "string.h"
#include <fstream>
#include <iostream>
#include <string>
#include <atomic>
#include "Defs.h"
#include "FileCache.h"
#include "HttpRequest.h"
#include "HttpRespone.h"
using namespace std;
void SendThread(HttpRespone* rep)
{
send(rep->sock, (const char*)rep->header, rep->header_size, 0);
send(rep->sock, (const char*)rep->buffer,rep->size, 0);
}
class ThreadCounterRAII
{
public:
ThreadCounterRAII()
{
_count.fetch_add(1);
}
static atomic<int > _count;
~ThreadCounterRAII()
{
_count.fetch_sub(1);
}
};
atomic<int> ThreadCounterRAII::_count = 0;
void ProcessRequestThread(HttpRequest*request )
{
ThreadCounterRAII counter;
..........
}
ThreadPool * ThreadPool::getInstance()
{
static ThreadPool * ins = nullptr;
if (!ins)ins = new ThreadPool;
return ins;
}
void ThreadPool::WorkThread()
{
while (true)
{
this->_mutex.lock();
while(this->works.empty())
{
this->_cond.wait(this->_mutex);
}
HttpRequest * req = this->works.front();
this->works.pop();
this->_mutex.unlock();
ProcessRequestThread(req);
}
}
void ThreadPool::addTask(HttpRequest*work)
{
this->_mutex.lock();
this->works.push(work);
this->_mutex.unlock();
this->_cond.notify_one();
}
ThreadPool::ThreadPool()
{
int MAX_THREADS = std::thread::hardware_concurrency();
while (MAX_THREADS--)
{
std::thread t(std::bind(&ThreadPool::WorkThread, this));
t.detach();
this->workers.push_back(std::move(t));
}
}
程序初始化时启动 10个工作线程,主线程接收到的TCP链接请求全部加入请求队列,工作线程处理
主线程:Application.cpp
while (true)
{
int sock_client = accept(sock, (sockaddr *)&client_ipaddr, &length);
if (sock_client == SOCKET_ERROR)
{
return;
}
HttpRequest*req = new HttpRequest;
req->sock = sock_client;
ThreadPool::getInstance()->addTask(req);
}