【C++】從零實現Json-Rpc框架（2）

JsonCpp庫

1.1- Json數據格式

1.2 - JsonCpp介紹

? 序列化接口?

? 反序列化接口

1.3 - Json序列化實踐

JsonCpp使用

Muduo庫

2.1 - Muduo庫是什么

2.2 - Muduo庫常見接口介紹

TcpServer類基礎介紹

EventLoop類基礎介紹

TcpConnection類基礎介紹

TcpClient類基礎介紹

Buffer類基礎介紹

2.3 - Muduo庫快速上手

英譯漢TCP服務器

?編輯

英譯漢客戶端

Makefile

項目匯總：uyeonashi的博客-CSDN博客

項目源碼：https://gitee.com/uyeonashi/project

本篇文章是對第三方庫的介紹和使用（JsonCpp庫，Muduo庫）！

JsonCpp庫

1.1- Json數據格式

Json 是一種數據交換格式，它采用完全獨立于編程語言的文本格式來存儲和表示數據。

例如: 我們想表示一個同學的學生信息

C 代碼表示：

char *name = "xx";
int age = 18;
float score[3] = {88.5, 99, 58};

Json 表示：

{"姓名" : "xx","年齡" : 18,"成績" : [88.5, 99, 58],"愛好" ：{"書籍" : "西游記","運動" : "打籃球"}
}

Json 的數據類型包括對象，數組，字符串，數字等。

對象：使用花括號{ } 括起來的表示一個對象
數組：使用中括號[ ] 括起來的表示一個數組
字符串：使用常規雙引號" " 括起來的表示一個字符串
數字：包括整形和浮點型，直接使用

1.2 - JsonCpp介紹

Jsoncpp 庫主要是用于實現Json 格式數據的序列化和反序列化，它實現了將多個數據對象組織成為json 格式字符串，以及將Json 格式字符串解析得到多個數據對象的功能。

Jsoncpp的介紹：3個類

? ? ? ? Json::Value類：中間數據存儲類

? ? ? ? 如果要將數據進行序列化，就需要先存儲到Json::Value對象當中

? ? ? ? 如果要將數據進行反序列化，就是解析后，將數據對象放入到Json::Value對象當中

先看一下Json 數據對象類的表示

class Json::Value
{Value &operator=(const Value &other); //Value重載了[]和=，因此所有的賦值和獲取數據都可以通過Value& operator[](const std::string& key);//簡單的方式完成 val["name"] ="xx";Value& operator[](const char* key);Value removeMember(const char* key);//移除元素const Value& operator[](ArrayIndex index) const; //val["score"][0]Value& append(const Value& value);//添加數組元素val["score"].append(88);ArrayIndex size() const;//獲取數組元素個數 val["score"].size();std::string asString() const;//轉string string name = val["name"].asString();const char* asCString() const;//轉char* char *name = val["name"].asCString();Int asInt() const;//轉int int age = val["age"].asInt();float asFloat() const;//轉float float weight = val["weight"].asFloat();bool asBool() const;//轉 bool bool ok = val["ok"].asBool();
};

Jsoncpp 庫主要借助三個類以及其對應的少量成員函數完成序列化及反序列化

? 序列化接口?

Json::StreamWriter類：用于進行數控序列化

? ? ? ? Json::StreamWriter::write() 序列化函數

Json::StreamWriterBuilder類：Json::StreamWriter工廠類 —?用于生成Json::StreamWriter對象

class JSON_API StreamWriter 
{virtual int write(Value const& root, std::ostream* sout) = 0;
}
class JSON_API StreamWriterBuilder : public StreamWriter::Factory 
{virtual StreamWriter* newStreamWriter() const;
}

? 反序列化接口

Json::CharReader類：反序列化類

? ? ? ? Json::CharReader::parse() 反序列化函數

Json::CharReaderBuilder類：Json::CharReader工廠類 — 用于生產Json::CharReader對象

class JSON_API CharReader 
{virtual bool parse(char const* beginDoc, char const* endDoc,Value* root, std::string* errs) = 0;
}
class JSON_API CharReaderBuilder : public CharReader::Factory 
{virtual CharReader* newCharReader() const;
}

1.3 - Json序列化實踐

JsonCpp使用

#include<iostream>
#include<memory>
#include<string>
#include<sstream>
#include<jsoncpp/json/json.h>//實現數據的序列化
bool serialize(const Json::Value &val,std::string &body)
{std::stringstream ss;//先實例化一個工廠類對象Json::StreamWriterBuilder swb;//通過工廠生產派生類對象std::unique_ptr<Json::StreamWriter> sw(swb.newStreamWriter());int ret = sw->write(val,&ss);if(ret != 0){std::cout << "json serialize failed\n";return false;}body = ss.str();return true;
}//實現json的反序列化
bool unserialize(const std::string &body,Json::Value &val)
{//實例化工廠對象Json::CharReaderBuilder crb;//生產Charreader對象std::string errs;std::unique_ptr<Json::CharReader> cr(crb.newCharReader());bool ret = cr->parse(body.c_str(),body.c_str()+body.size(),&val,&errs);if(ret == false){std::cout << "Json unserialize failed: " << errs << std::endl;return false;}return true;
}int main()
{const char *name = "小明";int age = 18;const char *sex = "男";float score[3] = {88,77.5,66};Json::Value student;student["姓名"] = name;student["年齡"] = age;student["性別"] = sex;student["成績"].append(score[0]);student["成績"].append(score[1]);student["成績"].append(score[2]);Json::Value fav;fav["書籍"] = "西游記";fav["運動"] = "打籃球";student["愛好"] = fav;std::string body;serialize(student,body);std::cout << body << std::endl;std::string str = R"({"姓名":"小黑","年齡":19,"成績":[32,45.5,56]})";Json::Value stu;bool ret = unserialize(str,stu);if(ret == false)return -1;std::cout << "姓名: " << stu["姓名"].asString() << std::endl;std::cout << "年齡:"  << stu["年齡"].asInt() << std::endl;int sz = stu["成績"].size();for(int i = 0; i < sz; i++){std::cout << "成績:" << stu["成績"][i].asFloat() << std::endl;} return 0;
}

編譯運行程序查看序列化和反序列化結果?

并在在這將其封裝了一下，方便我們后邊的使用

Muduo庫

2.1 - Muduo庫是什么

Muduo由陳碩大佬開發，是一個基于非阻塞IO和事件驅動的C++高并發TCP網絡編程庫。 它是一款基于主從Reactor模型的網絡庫，其使用的線程模型是one loop per thread, 所謂one loop per thread指的是：

一個線程只能有一個事件循環（EventLoop），用于響應計時器和IO事件
一個文件描述符只能由一個線程進行讀寫，換句話說就是一個TCP連接必須歸屬于某EventLoop管理

2.2 - Muduo庫常見接口介紹

TcpServer類基礎介紹

TcpServer{

? ? ? ? void start();? //啟動服務

? ? ? ? setConnectionCallback(); //設置連接建立 / 關閉時的回調函數

? ? ? ? setMessageCallback();//設置消息處理回調函數

};

typedef std::shared_ptr<TcpConnection> TcpConnectionPtr;
typedef std::function<void (const TcpConnectionPtr&)> ConnectionCallback;
typedef std::function<void (const TcpConnectionPtr&,Buffer*,Timestamp)> MessageCallback;
class InetAddress : public muduo::copyable
{
public:InetAddress(StringArg ip, uint16_t port, bool ipv6 = false);
};class TcpServer : noncopyable
{
public:enum Option{kNoReusePort,kReusePort,};TcpServer(EventLoop* loop,const InetAddress& listenAddr,const string& nameArg,Option option = kNoReusePort);void setThreadNum(int numThreads);void start();/// 當一個新連接建立成功的時候被調用void setConnectionCallback(const ConnectionCallback& cb){ connectionCallback_ = cb; }/// 消息的業務處理回調函數---這是收到新連接消息的時候被調用的函數void setMessageCallback(const MessageCallback& cb){ messageCallback_ = cb; }
};

EventLoop類基礎介紹

EvenLoop{

? ? ? ? void loop(); //開始事件監控事件循環

? ? ? ? void? quit(); //停止循環

? ? ? ? Timerld runAfter(delay,cb); //定時任務

};

class EventLoop : noncopyable
{
public:// Loops forever.// Must be called in the same thread as creation of the object.void loop();// Quits loop.// This is not 100% thread safe, if you call through a raw pointer,///better to call through shared_ptr<EventLoop> for 100% safety.void quit();TimerId runAt(Timestamp time, TimerCallback cb);// Runs callback after @c delay seconds.// Safe to call from other threads.TimerId runAfter(double delay, TimerCallback cb);// Runs callback every @c interval seconds.// Safe to call from other threads.TimerId runEvery(double interval, TimerCallback cb);// Cancels the timer.// Safe to call from other threads.void cancel(TimerId timerId);
private:std::atomic<bool> quit_;std::unique_ptr<Poller> poller_;mutable MutexLock mutex_;std::vector<Functor> pendingFunctors_ GUARDED_BY(mutex_);
};

TcpConnection類基礎介紹

TcpConnection{

? ? ? ? void send(std::string &msg); //發送數據

? ? ? ? bool connected(); //判斷當前連接是否連接正常

? ? ? ? shutdown down(); //關閉連接

}

class TcpConnection : noncopyable,public std::enable_shared_from_this<TcpConnection>
{
public:
// Constructs a TcpConnection with a connected sockfd
//
// User should not create this object.TcpConnection(EventLoop* loop,const string& name,int sockfd,const InetAddress& localAddr,const InetAddress& peerAddr);bool connected() const { return state_ == kConnected; }bool disconnected() const { return state_ == kDisconnected; }void send(string&& message); // C++11void send(const void* message, int len);void send(const StringPiece& message);// void send(Buffer&& message); // C++11void send(Buffer* message); // this one will swap datavoid shutdown(); // NOT thread safe, no simultaneous callingvoid setContext(const boost::any& context){ context_ = context; }const boost::any& getContext() const{ return context_; }boost::any* getMutableContext(){ return &context_; }void setConnectionCallback(const ConnectionCallback& cb){ connectionCallback_ = cb; }void setMessageCallback(const MessageCallback& cb){ messageCallback_ = cb; }
private:enum StateE { kDisconnected, kConnecting, kConnected, kDisconnecting };EventLoop* loop_;ConnectionCallback connectionCallback_;MessageCallback messageCallback_;WriteCompleteCallback writeCompleteCallback_;boost::any context_;
};

TcpClient類基礎介紹

TcpClient{

? ? ? ? void connect(); //連接服務器

????????void disconnect(); //關閉連接

? ? ? ? TcpConnectionPtr connection() const; // 獲取客戶端對應的TcpConnection連接

? ? ? ? Muduo庫的客戶端也是通過Eventloop進行IO事件監控IO處理的

? ? ? ? void setConnectionCallback(ConnectionCallback cb); //連接建立成功 / 關閉的回調處理

? ? ? ? void setMessageCallback(MessageCallback cb); //收到消息的回調處理

}?

CountDownLatch{ //做計數同步操作的類

? ? ? ? void wait(); //計數大于0則阻塞

? ? ? ? countDown(); //計數--，為0時喚醒wait

}

因為Client的connect接口是一個非阻塞操作，所以可能出現移走以外情況：connect連接還沒有完全建立的情況下，調用connection接口獲取連接，send發送數據??

class TcpClient : noncopyable
{
public:// TcpClient(EventLoop* loop);// TcpClient(EventLoop* loop, const string& host, uint16_t port);TcpClient(EventLoop* loop,const InetAddress& serverAddr,const string& nameArg);~TcpClient(); // force out-line dtor, for std::unique_ptr members.void connect();//連接服務器void disconnect();//關閉連接void stop();//獲取客?端對應的通信連接Connection對象的接?，發起connect后，有可能還沒有連接建?成功TcpConnectionPtr connection() const{MutexLockGuard lock(mutex_);return connection_;}// 連接服務器成功時的回調函數void setConnectionCallback(ConnectionCallback cb){ connectionCallback_ = std::move(cb); }// 收到服務器發送的消息時的回調函數void setMessageCallback(MessageCallback cb){ messageCallback_ = std::move(cb); }
private:EventLoop* loop_;ConnectionCallback connectionCallback_;MessageCallback messageCallback_;WriteCompleteCallback writeCompleteCallback_;TcpConnectionPtr connection_ GUARDED_BY(mutex_);
};/*需要注意的是，因為muduo庫不管是服務端還是客?端都是異步操作，對于客?端來說如果我們在連接還沒有完全建?成功的時候發送數據，這是不被允許的。因此我們可以使?內置的CountDownLatch類進?同步控制*/class CountDownLatch : noncopyable{public:explicit CountDownLatch(int count);void wait(){MutexLockGuard lock(mutex_);while (count_ > 0){condition_.wait();}}void countDown(){MutexLockGuard lock(mutex_);--count_;if (count_ == 0){condition_.notifyAll();}}int getCount() const;private:mutable MutexLock mutex_;Condition condition_ GUARDED_BY(mutex_);int count_ GUARDED_BY(mutex_);};

Buffer類基礎介紹

Buffer {

? ? ? ? size_t readableBytes();? //獲取緩沖區可讀數據大小

? ? ? ? const char* peek();? ? // 獲取緩沖區的起始地址

? ? ? ? int32_t peeklnt32() const；? //嘗試從緩沖區獲取4字節數據，進行網絡字節序轉換為整形，但數據并不從緩沖區刪除

? ? ? ? void retrieventlnt32();? //數據讀取位置向后偏移4字節,本質上就是刪除起始位置的4字節數據

? ? ? ? int32_t readlnt32();?

? ? ? ? string retrieveAllAsString(); //從緩沖區取出所有數據，當做string返回，并刪除緩沖區中的數據

? ? ? ? string retrieveAsString(size_t len),從緩沖區中取出len長度的數據，當做string返回，并刪除緩沖區中的數據

}

class Buffer : public muduo::copyable
{public:static const size_t kCheapPrepend = 8;static const size_t kInitialSize = 1024;explicit Buffer(size_t initialSize = kInitialSize): buffer_(kCheapPrepend + initialSize),readerIndex_(kCheapPrepend),writerIndex_(kCheapPrepend);void swap(Buffer& rhs)size_t readableBytes() constsize_t writableBytes() constconst char* peek() constconst char* findEOL() constconst char* findEOL(const char* start) constvoid retrieve(size_t len)void retrieveInt64()void retrieveInt32()void retrieveInt16()void retrieveInt8()string retrieveAllAsString()string retrieveAsString(size_t len)void append(const StringPiece& str)void append(const char* /*restrict*/ data, size_t len)void append(const void* /*restrict*/ data, size_t len)char* beginWrite()const char* beginWrite() constvoid hasWritten(size_t len)void appendInt64(int64_t x)void appendInt32(int32_t x)void appendInt16(int16_t x)void appendInt8(int8_t x)int64_t readInt64()int32_t readInt32()int16_t readInt16()int8_t readInt8()int64_t peekInt64() constint16_t peekInt16() constint8_t peekInt8() constvoid prependInt64(int64_t x)void prependInt32(int32_t x)void prependInt16(int16_t x)void prependInt8(int8_t x)void prepend(const void* /*restrict*/ data, size_t len)
private:std::vector<char> buffer_;size_t readerIndex_;size_t writerIndex_;static const char kCRLF[];
};

2.3 - Muduo庫快速上手

我們使用Muduo網絡庫來實現一個簡單英譯漢服務器和客戶端快速上手Muduo庫

英譯漢TCP服務器

包含頭文件時指定路徑

// 實現一個翻譯服務器，客戶端發來一個英語單詞，返回一個漢語詞典#include <muduo/net/TcpServer.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/TcpConnection.h>
#include <muduo/net/Buffer.h>
#include <iostream>
#include <string>
#include <unordered_map>class DictServer
{
public:DictServer(int port): _server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), "DictServer", muduo::net::TcpServer::kReusePort){//設置連接事件（連接建立/管理）的回調_server.setConnectionCallback(std::bind(&DictServer::onConnection, this, std::placeholders::_1));//設置連接消息的回調_server.setMessageCallback(std::bind(&DictServer::onMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));}void start(){_server.start(); //先開始監聽_baseloop.loop();//開始死循環執行事件監控}private:void onConnection(const muduo::net::TcpConnectionPtr &conn){if (conn->connected())std::cout << "連接建立!\n";elsestd::cout << "連接斷開!\n";}void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp){static std::unordered_map<std::string, std::string> dict_map = {{"hello", "你好"},{"world", "世界"},{"bite", "比特"}};std::string msg = buf->retrieveAllAsString();std::string res;auto it = dict_map.find(msg);if (it != dict_map.end()){res = it->second;}else{res = "未知單詞";}conn->send(res);}private:muduo::net::EventLoop _baseloop;muduo::net::TcpServer _server;
};int main()
{DictServer server(9090);server.start();return 0;
}

英譯漢客戶端

#include <muduo/net/TcpClient.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/TcpConnection.h>
#include <muduo/net/EventLoopThread.h>
#include <muduo/net/Buffer.h>
#include <muduo/base/CountDownLatch.h>
#include <memory>
#include <iostream>
#include <string>class DictClient
{
public:DictClient(const std::string &sip, int sport):_baseloop(_loopthread.startLoop()) ,_downlatch(1), _client(_baseloop, muduo::net::InetAddress(sip, sport), "DictClient"){// 設置連接事件（連接建立/管理）的回調_client.setConnectionCallback(std::bind(&DictClient::onConnection, this, std::placeholders::_1));// 設置連接消息的回調_client.setMessageCallback(std::bind(&DictClient::onMessage, this,std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));//連接服務器_client.connect();_downlatch.wait();}bool send(const std::string &msg){if(_conn->connected() == false){std::cout << "鏈接已斷開，發送數據失敗！\n";return false;}_conn->send(msg);//_baseloop.loop();  //開始事件循環監控 -- 內部是個死循環，客戶端不能直接使用return true;}private:void onConnection(const muduo::net::TcpConnectionPtr &conn){if (conn->connected()){std::cout << "連接建立!\n";_downlatch.countDown(); // 計數--，為0時喚醒阻塞_conn = conn;}else{std::cout << "連接斷開!\n";_conn.reset();}}void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp){std::string res = buf->retrieveAllAsString();std::cout << res << std::endl;}private:muduo::net::TcpConnectionPtr _conn;muduo::CountDownLatch _downlatch;muduo::net::EventLoopThread _loopthread;muduo::net::EventLoop *_baseloop;muduo::net::TcpClient _client;
};int main()
{DictClient client("127.0.0.1",9090);while(1){std::string msg;std::cin >> msg;client.send(msg);}return 0;
}

Makefile

CFLAG=-I ../../build/release-install-cpp11/include/
LFLAF=-L ../../build/release-install-cpp11/lib -lmuduo_net -lmuduo_base -pthread
all: server client
server:server.cppg++ $(CFLAG) -o $@ $^ $(LFLAF) -std=c++11
client:client.cppg++ $(CFLAG) -o $@ $^ $(LFLAF) -std=c++11

?C++11 異步操作?

3.1 - std::future

介紹

std::future是C++11標準庫中的一個模板類，它表示一個異步操作的結果。當我們在多線程編程中使用異步任務時，std::future可以幫助我們在需要的時候獲取任務的執行結果。std::future的一個重要特性是能夠阻塞當前線程，直到異步操作完成，從而確保我們在獲取結果時不會遇到未完成的操作。

應用場景

異步任務： 當我們需要在后臺執行一些耗時操作時，如網絡請求或計算密集型任務等，std::future可以用來表示這些異步任務的結果。通過將任務與主線程分離，我們可以實現任務的并行處理，從而提高程序的執行效率
并發控制： 在多線程編程中，我們可能需要等待某些任務完成后才能繼續執行其他操作。通過使用std::future，我們可以實現線程之間的同步，確保任務完成后再獲取結果并繼續執行后續操作
結果獲取：std::future提供了一種安全的方式來獲取異步任務的結果。我們可以使用std::future::get()函數來獲取任務的結果，此函數會阻塞當前線程，直到異步操作完成。這樣，在調用get()函數時，我們可以確保已經獲取到了所需的結果

用法示例

使用 std::async關聯異步任務

std::async是一種將任務與std::future關聯的簡單方法。它創建并運行一個異步任務，并返回一個與該任務結果關聯的std::future對象。默認情況下，std::async是否啟動一個新線程，或者在等待future時，任務是否同步運行都取決于你給的參數。這個參數為std::launch類型：

? ? ? ??std::launch::deferred 表明該函數會被延遲調用，直到在future上調用get()或者wait()才會開始執行任務
????????std::launch::async 表明函數會在自己創建的線程上運行
????????std::launch::deferred | std::launch::async 內部通過系統等條件自動選擇策略

#include <iostream>
#include <thread>
#include <future>int Add(int num1, int num2) 
{std::cout << "into add!\n";return num1 + num2;
}int main()
{//std::launch::async策略：內部創建一個線程執行函數，函數運行結果通過future獲取//std::launch::deferred策略：同步策略，獲取結果的時候再去執行任務//std::future<int> res = std::async(std::launch::deferred, Add, 11, 22);std::future<int> res = std::async(std::launch::async, Add, 11, 22);//進行了一個異步非阻塞調用std::this_thread::sleep_for(std::chrono::seconds(1));std::cout << "--------------------------\n";//std::future<int>::get()  用于獲取異步任務的結果，如果還沒有結果就會阻塞std::cout << res.get() << std::endl;return 0;
}

使用std::packaged_task和std::future配合

std::packaged_task就是將任務和 std::feature 綁定在一起的模板，是一種對任務的封裝。我們可以通過std::packaged_task對象獲取任務相關聯的std::feature對象，通過調用get_future()方法獲得。
std::packaged_task的模板參數是函數簽名。
可以把std::future和std::async看成是分開的，而 std::packaged_task則是一個整體。

#include<iostream>
#include<future>
#include<thread>
#include<memory>int Add(int num1, int num2)
{std::cout << "into add!\n";return num1 + num2;
}int main()
{//1. 封裝任務auto task = std::make_shared<std::packaged_task<int(int, int)>>(Add);//2. 獲取任務包關聯的future對象std::future<int> res = task->get_future();std::thread thr([&task](){(*task)(11,22);});//4. 獲取結果std::cout << res.get() << std::endl;thr.join();return 0;
}

使用std::promise和std::future配合

std::promise提供了一種設置值的方式，它可以在設置之后通過相關聯的std::future對象進行讀取。換種說法就是之前說過std::future可以讀取一個異步函數的返回值了，但是要等待就緒，而std::promise就提供一種方式手動讓 std::future就緒

#include<iostream>
#include<future>
#include<thread>
#include<memory>int Add(int num1,int num2)
{std::cout << "into add!\n";return num1+num2;
}int main()
{//1.在使用的時候，先實例化一個指定結果的promise對象std::promise<int> pro;//2. 通過promise對象獲取關聯future對象std::future<int> res = pro.get_future();//3. 在任意位置給promise設置數據，就可以通過關聯的future獲取到這個設置的數據了std::thread thr([&pro](){int sum = Add(11,22);pro.set_value(sum);});std::cout << res.get() << std::endl;thr.join();return 0;
}