OpenSocket是一個(gè)跨全平臺(tái)的高性能網(wǎng)絡(luò)并發(fā)庫(kù)谚咬。
它使用了高性能IO萝快，Linux和安卓用epoll，Win32用IOCP贡翘，iOS和Mac用kqueue，其他系統(tǒng)使用select砰逻。
本文用這種高性能socket庫(kù)鸣驱，設(shè)計(jì)開(kāi)發(fā)一個(gè)HttpClient。
為了開(kāi)發(fā)方面蝠咆，我們使用OpenThread作為線程庫(kù)踊东。OpenThread可以實(shí)現(xiàn)多線程三大設(shè)計(jì)模式，開(kāi)發(fā)這個(gè)HttpClient刚操，使用Worker模式闸翅。
設(shè)計(jì)思路如下：
每個(gè)HttpClient是一條線程O(píng)penThreader。一個(gè)HttpClient對(duì)象菊霜，可以處理任意個(gè)Http請(qǐng)求坚冀。它由Factory類管理和創(chuàng)建。
一次Http請(qǐng)求就是一個(gè)task鉴逞，從Factory挑選一個(gè)HttpClient對(duì)象记某，向該對(duì)象發(fā)送task。并進(jìn)行阻塞等待結(jié)果华蜒。
HttpClient對(duì)象收到task消息辙纬。把IP和端口做參數(shù)，調(diào)用OpenSocket的connect叭喜。
connect有兩個(gè)作用贺拣，一個(gè)是執(zhí)行網(wǎng)絡(luò)連接產(chǎn)生一個(gè)fd，同時(shí)把這個(gè)fd加入到poll捂蕴，fd與HttpClient對(duì)象的線程id進(jìn)行綁定譬涡。
fd加入到poll成功以后，該socket的任何消息可以通過(guò)線程Id啥辨，發(fā)到對(duì)應(yīng)的HttpClient涡匀。HttpClient對(duì)象接收到socket的open消息后，向服務(wù)器發(fā)送http報(bào)文溉知。
因?yàn)閏onnect把fd和線程id進(jìn)行綁定陨瘩。所以，HttpClient會(huì)收到服務(wù)器返回的Http報(bào)文级乍。
HttpClient接收完Http報(bào)文舌劳，就喚醒請(qǐng)求線程。請(qǐng)求線程被喚醒玫荣，拿到Http請(qǐng)求數(shù)據(jù)甚淡。
測(cè)試?yán)邮谦@取交易所的最新龍虎數(shù)據(jù)。
具體源碼如下：
#include <assert.h>
#include <time.h>
#include <math.h>
#include <map>
#include "open/openthread.h"
#include "opensocket.h"
using namespace open;
//請(qǐng)求http對(duì)象捅厂，包含返回對(duì)象贯卦。
class HttpRequest
{
std::string url_;
public:
int port_;
std::string host_;
std::string ip_;
std::string path_;
std::string method_;
std::string body_;
//http請(qǐng)求頭
std::map<std::string, std::string> headers_;
HttpRequest() :port_(80) {}
std::string& operator[](const std::string& key) { return headers_[key]; }
//指定url资柔，并進(jìn)行解析和域名解析
void setUrl(const std::string& url)
{
if (url.empty()) return;
url_ = url;
int len = (int)url.length();
char* ptr = (char*)url.c_str();
if (len >= 8)
{
if (memcmp(ptr, "http://", strlen("http://")) == 0)
ptr += strlen("http://");
else if (memcmp(ptr, "https://", strlen("https://")) == 0)
ptr += strlen("https://");
}
const char* tmp = strstr(ptr, "/");
path_.clear();
if (tmp != 0)
{
path_.append(tmp);
host_.clear();
host_.append(ptr, tmp - ptr);
}
else
{
host_ = ptr;
}
port_ = 80;
ip_.clear();
ptr = (char*)host_.c_str();
tmp = strstr(ptr, ":");
if (tmp != 0)
{
ip_.append(ptr, tmp - ptr);
tmp += 1;
port_ = atoi(tmp);
}
else
{
ip_ = ptr;
}
//域名解析，把域名轉(zhuǎn)ip撵割』哐撸可以緩存，提供效率
ip_ = OpenSocket::DomainNameToIp(ip_);
}
inline void operator=(const std::string& url) { setUrl(url); }
//http返回對(duì)象
struct HttpResponse
{
int code_;
int clen_;
std::string head_;
std::string body_;
//std::multimap<std::string, std::string> headers_;
std::map<std::string, std::string> headers_;
std::string& operator[](const std::string& key) { return headers_[key]; }
HttpResponse():code_(0), clen_(0) {}
//解析返回http消息頭
void parseHeader()
{
if (!headers_.empty() || head_.size() < 12) return;
std::string line;
const char* ptr = strstr(head_.c_str(), "\r\n");
if (!ptr) return;
code_ = 0;
clen_ = 0;
line.append(head_.c_str(), ptr - head_.c_str());
for (size_t i = 0; i < line.size(); i++)
{
if (line[i] == ' ')
{
while (i < line.size() && line[i] == ' ') ++i;
code_ = std::atoi(line.data() + i);
break;
}
}
if (code_ <= 0) return;
line.clear();
int k = -1;
int j = -1;
std::string key;
std::string value;
for (size_t i = ptr - head_.c_str() + 2; i < head_.size() - 1; i++)
{
if (head_[i] == '\r' && head_[i + 1] == '\n')
{
if (j > 0)
{
k = 0;
while (k < line.size() && line[k] == ' ') ++k;
while (k >= 0 && line.back() == ' ') line.pop_back();
value = line.data() + j + 1;
while (j >= 0 && line[j] == ' ') j--;
key.clear();
key.append(line.data(), j);
for (size_t x = 0; x < key.size(); x++)
key[x] = std::tolower(key[x]);
headers_[key] = value;
}
++i;
j = -1;
line.clear();
continue;
}
line.push_back(head_[i]);
if (j < 0 && line.back() == ':')
{
j = line.size() - 1;
}
}
clen_ = std::atoi(headers_["content-length"].c_str());
}
};
HttpResponse response_;
//阻塞當(dāng)前線程睁枕，等待http消息返回官边，才繼續(xù)執(zhí)行。
OpenSync openSync_;
};
//OpenThread的線程之間通信數(shù)據(jù)結(jié)構(gòu)外遇，用isSocket_區(qū)別是socket消息還是http請(qǐng)求消息
struct BaseProto
{
bool isSocket_;
};
//攜帶OpenSocket消息的數(shù)據(jù)結(jié)構(gòu)，isSocket_=true
struct SocketProto : public BaseProto
{
std::shared_ptr<OpenSocketMsg> data_;
};
//攜帶http請(qǐng)求消息的數(shù)據(jù)結(jié)構(gòu)契吉，isSocket_=false
struct TaskProto : public BaseProto
{
int fd_;
OpenSync openSync_;
std::shared_ptr<HttpRequest> request_;
};
//應(yīng)用程序單利跳仿，封裝OpenSocket，一個(gè)進(jìn)程只有一個(gè)對(duì)象捐晶。
class App
{
//OpenSocketMsg需要手動(dòng)釋放菲语，放到智能指針，由智能指針釋放
static void SocketFunc(const OpenSocketMsg* msg)
{
if (!msg) return;
//msg需要手動(dòng)delete惑灵，把它托管給智能指針
auto proto = std::shared_ptr<SocketProto>(new SocketProto);
//OpenThread的線程id >= 0山上，所以只處理非負(fù)數(shù)的條件
if (msg->uid_ >= 0)
{
proto->isSocket_ = true;
proto->data_ = std::shared_ptr<OpenSocketMsg>((OpenSocketMsg*)msg);
//msg->uid_是綁定的線程id，向該線程派發(fā)socket消息
if (!OpenThread::Send((int)msg->uid_, proto))
printf("SocketFunc dispatch faild pid = %lld\n", msg->uid_);
}
}
public:
static App Instance_;
//OpenSocket對(duì)象英支，可以設(shè)計(jì)成單利
OpenSocket openSocket_;
//App構(gòu)造的時(shí)候佩憾，啟動(dòng)OpenSocket。
App() { openSocket_.run(App::SocketFunc); }
};
App App::Instance_;
//HttpClient線程類干花，F(xiàn)actory管理一組線程妄帘。
class HttpClient : public OpenThreader
{
//Factory
class Factory
{
const std::vector<HttpClient*> vectWorker_;
public:
Factory()
:vectWorker_({
new HttpClient("HttpClient1"),
new HttpClient("HttpClient2"),
new HttpClient("HttpClient3"),
new HttpClient("HttpClient4"),
}) {}
//采用隨機(jī)方式，提供一個(gè)線程
HttpClient* getWorker()
{
if (vectWorker_.empty()) return 0;
return vectWorker_[std::rand() % vectWorker_.size()];
}
};
static Factory Instance_;
// name是線程名池凄，必須制定抡驼。在Linux上，top -Hp可以看到這個(gè)線程名肿仑。
HttpClient(const std::string& name)
:OpenThreader(name)
{
start();
}
~HttpClient()
{
//銷毀之前致盟，盡可能喚醒請(qǐng)求線程，防止請(qǐng)求線程阻塞
for (auto iter = mapFdToTask_.begin(); iter != mapFdToTask_.end(); iter++)
iter->second.openSync_.wakeup();
}
//處理請(qǐng)求http線程發(fā)過(guò)來(lái)的消息
void onHttp(TaskProto& proto)
{
auto& request = proto.request_;
//連接Http服務(wù)器尤慰，并把fd與當(dāng)前線程綁定馏锡。該socket的全部消息，都發(fā)到此線程
proto.fd_ = App::Instance_.openSocket_.connect(pid(), request->ip_, request->port_);
request->response_.code_ = -1;
request->response_.head_.clear();
request->response_.body_.clear();
//fd與任務(wù)綁定到任務(wù)列表
mapFdToTask_[proto.fd_] = proto;
}
//與http服務(wù)器連接成功以后割择，發(fā)送http請(qǐng)求報(bào)文
void onSend(const std::shared_ptr<OpenSocketMsg>& data)
{
//需要判斷fd綁定的task是否存在眷篇，否則關(guān)閉與Http服務(wù)器的連接
auto iter = mapFdToTask_.find(data->fd_);
if (iter == mapFdToTask_.end())
{
App::Instance_.openSocket_.close(pid(), data->fd_);
return;
}
auto& task = iter->second;
auto& request = task.request_;
std::string buffer = request->method_ + " " + request->path_ + " HTTP/1.1 \r\n";
auto iter1 = request->headers_.begin();
for (; iter1 != request->headers_.end(); iter1++)
{
buffer.append(iter1->first + ": " + iter1->second + "\r\n");
}
if (!request->body_.empty())
{
buffer.append("Content-Length:" + std::to_string(request->body_.size()) + "\r\n\r\n");
buffer.append(request->body_);
buffer.append("\r\n");
}
else
{
buffer.append("\r\n");
}
//制作好Http請(qǐng)求報(bào)文，發(fā)送給服務(wù)器荔泳。
App::Instance_.openSocket_.send(task.fd_, buffer.data(), (int)buffer.size());
}
//處理Http服務(wù)器發(fā)送過(guò)了socket數(shù)據(jù)流蕉饼，拼成完整的Http返回報(bào)文
void onRead(const std::shared_ptr<OpenSocketMsg>& data)
{
//Http任務(wù)列表沒(méi)有綁定fd的任務(wù)虐杯，就對(duì)該fd關(guān)閉。
auto iter = mapFdToTask_.find(data->fd_);
if (iter == mapFdToTask_.end())
{
App::Instance_.openSocket_.close(pid(), data->fd_);
return;
}
auto& task = iter->second;
auto& response = task.request_->response_;
//處理返回http頭
if (response.code_ == -1)
{
response.head_.append(data->data(), data->size());
const char* ptr = strstr(response.head_.data(), "\r\n\r\n");
if (!ptr) return;
response.code_ = 0;
response.body_.append(ptr + 4);
response.head_.resize(ptr - response.head_.data() + 2);
response.parseHeader();
}
//處理返回的http的body
else
{
response.body_.append(data->data(), data->size());
}
if (response.clen_ > 0)
{
if (response.clen_ >= response.body_.size())
response.body_.resize(response.clen_);
App::Instance_.openSocket_.close(pid(), data->fd_);
}
else if (response.body_.size() > 2)
{
if (response.body_[response.body_.size() - 2] == '\r' && response.body_.back() == '\n')
{
response.body_.pop_back();
response.body_.pop_back();
App::Instance_.openSocket_.close(pid(), data->fd_);
}
}
}
//與Http服務(wù)器關(guān)閉的消息昧港，喚醒請(qǐng)求線程擎椰，并對(duì)fd綁定的任務(wù)，移出任務(wù)列表
void onClose(const std::shared_ptr<OpenSocketMsg>& data)
{
auto iter = mapFdToTask_.find(data->fd_);
if (iter != mapFdToTask_.end())
{
iter->second.openSync_.wakeup();
mapFdToTask_.erase(iter);
}
}
//接收綁定此線程的socket消息创肥。
void onSocket(const SocketProto& proto)
{
const auto& msg = proto.data_;
switch (msg->type_)
{
case OpenSocket::ESocketData:
onRead(msg);
break;
case OpenSocket::ESocketClose:
onClose(msg);
break;
case OpenSocket::ESocketError:
printf("[%s]ESocketError:%s\n", ThreadName((int)msg->uid_).c_str(), msg->info());
onClose(msg);
break;
case OpenSocket::ESocketWarning:
printf("[%s]ESocketWarning:%s\n", ThreadName((int)msg->uid_).c_str(), msg->info());
break;
case OpenSocket::ESocketOpen:
onSend(msg);
break;
case OpenSocket::ESocketAccept:
case OpenSocket::ESocketUdp:
assert(false);
break;
default:
break;
}
}
//處理static bool Http(std::shared_ptr<HttpRequest>& request)發(fā)過(guò)來(lái)的消息
virtual void onMsg(OpenThreadMsg& msg)
{
const BaseProto* data = msg.data<BaseProto>();
if (!data) return;
if (!data->isSocket_)
{
TaskProto* proto = msg.edit<TaskProto>();
if (proto) onHttp(*proto);
}
else
{
const SocketProto* proto = msg.data<SocketProto>();
if (proto) onSocket(*proto);
}
}
std::map<int, TaskProto> mapFdToTask_;
public:
static bool Http(std::shared_ptr<HttpRequest>& request)
{
if (request->ip_.empty())
{
assert(false);
return false;
}
//類型線程池中达舒，選擇一個(gè)。
auto worker = Instance_.getWorker();
if (!worker) return false;
auto proto = std::shared_ptr<TaskProto>(new TaskProto);
proto->request_ = request;
proto->isSocket_ = false;
//接收消息地方：virtual void onMsg(OpenThreadMsg& msg)
bool ret = OpenThread::Send(worker->pid(), proto);
assert(ret);
//阻塞叹侄，等待http請(qǐng)求完成喚醒
proto->openSync_.await();
return ret;
}
};
HttpClient::Factory HttpClient::Instance_;
int main()
{
auto request = std::shared_ptr<HttpRequest>(new HttpRequest);
//請(qǐng)求交易所的最新龍虎數(shù)據(jù)
request->setUrl("http://reportdocs.static.szse.cn/files/text/jy/jy230308.txt");
request->method_ = "GET";
//自定義Http請(qǐng)求頭
(*request)["Host"] = "reportdocs.static.szse.cn";
(*request)["Accept"] = "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7";
(*request)["Accept-Encoding"] = "gzip,deflate";
(*request)["Accept-Language"] = "zh-CN,zh;q=0.9";
(*request)["Cache-Control"] = "max-age=0";
(*request)["User-Agent"] = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36(KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36";
(*request)["Upgrade-Insecure-Requests"] = "1";
//發(fā)送http請(qǐng)求
HttpClient::Http(request);
//返回http請(qǐng)求
auto& response = request->response_;
printf("code:%d, header:%s\n", response.code_, response.head_.c_str());
return getchar();
}
編譯和執(zhí)行
請(qǐng)安裝cmake工具巩搏，用cmake可以構(gòu)建出VS或者XCode工程，就可以在vs或者xcode上編譯運(yùn)行趾代。
源代碼：https://github.com/openlinyou/opensocket
https://gitee.com/linyouhappy/opensocket
#克隆項(xiàng)目
git clone https://github.com/openlinyou/opensocket
cd ./opensocket
#創(chuàng)建build工程目錄
mkdir build
cd build
cmake ..
#如果是win32贯底，在該目錄出現(xiàn)opensocket.sln，點(diǎn)擊它就可以啟動(dòng)vs寫(xiě)代碼調(diào)試
make
./httpclient
