學(xué)習(xí)陳碩寫的網(wǎng)絡(luò)庫(kù)muduo，照著實(shí)現(xiàn)了一遍，項(xiàng)目地址為learn_muduo.

文章目錄

• • base庫(kù)
  • • copyable、noncopyable
    • Atomic
    • Timestamp
    • Date
    • Mutex
    • Condition
    • CountDownLatch
    • Thread
    • CurrentThread
    • Exception
    • BlockingQueue
    • StringPiece
    • LogStream
    • Logging
  • net庫(kù)
  • • Endian
    • InetAddress
    • SocketsOps
    • Socket
    • Channel
    • Poller
    • EventLoop
    • TimerQueue
    • Acceptor
    • TcpConnection
    • Connector
    • Buffer
    • TcpServer
    • TcpClient
    • epoll
    • HttpServer

base庫(kù)

copyable、noncopyable

刪除默認(rèn)的拷貝構(gòu)造和賦值來(lái)實(shí)現(xiàn)類的不可拷貝的屬性。

noncopyable(const noncopyable&) = delete; void operator=(const noncopyable&) = delete;

Atomic

Atomic是原子操作類，它是一個(gè)模板類，使用GCC提供的加減和邏輯原子操作來(lái)實(shí)現(xiàn)。

private:volatile T value_;// 原子比較和交換，先判斷*ptr是否和oldval相等, 如果相等將值設(shè)置為newval __sync_val_compare_add_swap(type *ptr, type oldval, newval);// value_ += x; __sync_fetch_and_add(&value_, x);// value_ = newValue; __sync_lock_test_and_set(&value_, newValue);

通過(guò)以上代碼，封裝自增、自減和賦值等原子操作。

Atomic的類模板定義在namespace detail中，在namespace muduo使用模板創(chuàng)建了兩個(gè)類。

typedef detail::AtomicIntegerT<int32_t> AtomicInt32; typedef detail::AtomicIntegerT<int64_t> AtomicInt64;

Timestamp

Timestamp使用微妙來(lái)計(jì)算時(shí)間，提供toString和格式化的接口，可以獲取當(dāng)前時(shí)間，返回一個(gè)對(duì)應(yīng)的Timestamp對(duì)象，也可以返回一個(gè)

static Timestamp now(); // 返回當(dāng)前時(shí)間 static Timestamp invalid(); // 返回一個(gè)空對(duì)象 string toString() const; // 返回（秒.微妙）格式 string toFormattedString(bool showMicroseconds = true) const; // 格式化時(shí)間，年月日 時(shí):分:秒.微妙

Timestamp繼承boost::less_than_comparable，只需提供<的實(shí)現(xiàn)，自動(dòng)實(shí)現(xiàn)>、<=、>=，繼承boost::equality_comparable只需提供==自動(dòng)實(shí)現(xiàn)!=。

class Timestamp : public boost::equality_comparable<Timestamp>,public boost::less_than_comparable<Timestamp>

跨平臺(tái)，int64_t在32位系統(tǒng)是long long int(%lld), 在64位系統(tǒng)是long int(%ld)

printf("%" PRId64 "\n", value);

Date

使用julianDayNumber來(lái)計(jì)算年月日。距離公元前4713年1月1日的天數(shù)。和Timestamp類似，提供一些常用的接口。

YearMonthDay getYearMonthDay(int julianDayNumber); // 獲取對(duì)應(yīng)的年月日 int getJulianDayNumber(int year, int month, int day)// 獲取julian day string Date::toIsoString(); // 格式化（年-月-日）

Mutex

Mutex封裝鎖，提供加鎖解鎖等操作。

// 屬性 pthread_mutex_t mutex_; // 定義一把鎖 pid_t holder_; // 記錄加鎖的線程ID

Mutex中一共有3個(gè)類：MutexLock、UnassignGuard、MutexLockGuard。

MutexLock使用pthread_mutex_函數(shù)封裝初始化鎖、加鎖、解鎖、銷毀鎖的操作。

UnassignGuard是MutexLock的內(nèi)部類，他的特點(diǎn)是在構(gòu)造函數(shù)中清除鎖的持有者ID，析構(gòu)的時(shí)候設(shè)置鎖的持有者ID，這是供``Condition的wait()`使用。

MutexLockGuard采用RAII，構(gòu)造的時(shí)候申請(qǐng)資源lock，析構(gòu)的時(shí)候釋放資源unlock，解放雙手。

Condition

Condition實(shí)現(xiàn)條件變量功能，使用pthread_cond_函數(shù)來(lái)封裝wait()，notify()，notifyAll功能。也是采用RAII的機(jī)制，在構(gòu)造中初始化條件變量pcond_，在析構(gòu)函數(shù)中銷毀條件變量。

Condition是MutexLock的友元，可以使用MutexLock的內(nèi)部類UnassignGuard來(lái)實(shí)現(xiàn)wait()的功能。

pthread_cond_wait內(nèi)部的機(jī)制時(shí)在線程進(jìn)入阻塞前釋放資源，當(dāng)函數(shù)返回，重新持有鎖。

void wait() {MutexLock::UnassignGuard ug(mutex_);int ret = pthread_cond_wait(&pcond_, mutex_.getPthreadMutex()); // 將線程添加到條件變量assert(ret == 0); }

CountDownLatch

倒計(jì)時(shí)類，將MutexLock和Condition封裝在一起。

mutable MutexLock mutex_; Condition condition_; int count_;void wait(); // 調(diào)用Condition等待 void countDown(); // 技術(shù)減1 int getCount(); // 獲取當(dāng)前的計(jì)數(shù)

Thread

線程類，在namespace muduo中定義了ThreadNameInitializer類負(fù)責(zé)主線程的初始化操作，指定如果fork之后再之進(jìn)程中執(zhí)行after函數(shù)。

ThreadNameInitializer() {muduo::CurrentThread::t_threadName = "main";CurrentThread::tid();pthread_atfork(NULL, NULL, &afterFork); }

設(shè)置ThreadData的結(jié)構(gòu)體，保存線程的回調(diào)函數(shù)，名字，id，計(jì)數(shù)等信息。在ThreadData中定義了runInThread函數(shù)，用來(lái)執(zhí)行回調(diào)函數(shù)。

線程執(zhí)行的流程是：

• Thread構(gòu)造，設(shè)置回調(diào)函數(shù)func_，默認(rèn)CountDownLatch為1。

• pthread_create創(chuàng)建線程，綁定回調(diào)函數(shù)startThread，將ThreadData作為參數(shù)，創(chuàng)建成功之后主線程阻塞在latch_上，等待子線程的退出。

• 在startThread中調(diào)用ThreadData的runInThread函數(shù)

• latch_-1喚醒主線程，同時(shí)執(zhí)行回調(diào)函數(shù)func_，同時(shí)對(duì)異常進(jìn)行處理。

CurrentThread

主線程類，提供stackTrace()用于查看堆棧的信息，同時(shí)包括線程的一些基本屬性，id、名字等。

Exception

異常處理類，繼承std::exception，封裝CurrentThread類的stackTrace()和重寫what()方法。

BlockingQueue

無(wú)界阻塞隊(duì)列，底層是deque，利用條件變量實(shí)現(xiàn)一個(gè)生產(chǎn)者消費(fèi)者模型，另外還有一個(gè)有界的阻塞隊(duì)列（BoundedBlockingQueue），底層是circular_buffer。

StringPiece

C++支持兩種字符串：string和char*，當(dāng)char*傳入函數(shù)，會(huì)構(gòu)造一個(gè)臨時(shí)的string變量，這就發(fā)生了內(nèi)存的拷貝。StringPiece就是為了減少這種內(nèi)存的拷貝，統(tǒng)一使用char*記錄字符串。重載了[]、等于、比較等操作。重載<<支持logged的使用。

LogStream

muduo的日志庫(kù)采用C++的stream風(fēng)格，有個(gè)好處是輸出日志級(jí)別高于語(yǔ)句的日志級(jí)別的時(shí)候，打印是個(gè)空操作。muduo沒有使用標(biāo)準(zhǔn)庫(kù)中的iostream，而是自己封裝的LogStream，不同于iostream，LogStream的<<操作是將數(shù)據(jù)放到緩沖區(qū)(FixedBuffer)中，外部程序可以重定向到任何文件中。

Logging

日志類，muduo日志信息一共有5個(gè)級(jí)別，TRACE，DEBUG，INFO，WARN，ERROR，FATAL。通過(guò)宏定義創(chuàng)建Logger的臨時(shí)對(duì)象，調(diào)用stream()函數(shù)返回LogStream對(duì)象。在Logging中定義了Impl類和SourceFile類，Impl類保存日志需要數(shù)據(jù)，SourceFile中LOG_函數(shù)所在的源文件和行號(hào)。

net庫(kù)

Endian

提供字節(jié)序的轉(zhuǎn)化。

本地字節(jié)序 <–> 網(wǎng)絡(luò)字節(jié)序

InetAddress

InetAddress是對(duì)sockaddr_in和sockaddr_in6的封裝。

// 設(shè)置本地端口 InetAddress(uint16_t port = 0, bool loopbackOnly = false, bool ipv6 = false); // 設(shè)置一個(gè)指定的ip和端口 InetAddress(StringArg ip, uint16_t port, bool ipv6 = false);sa_family_t family(); // 返回協(xié)議類型 string toIpPort() const; // 獲取ip和port string toIp() const; // 獲取ip uint16_t toPort() const; // 獲取port

SocketsOps

封裝對(duì)socket的常用操作。

int createNonblockingOrDie(sa_family_t family); // 創(chuàng)建非阻塞的socket int connect(int sockfd, const struct sockaddr* addr); void bindOrDie(int sockfd, const struct sockaddr* addr); void listenOrDie(int sockfd); int accept(int sockfd, struct sockaddr_in6* addr); // 包含錯(cuò)誤處理 void close(int sockfd); void shutdownWrite(int sockfd);void toIpPort(char* buf, size_t size, const struct sockaddr* addr); // 獲取ip+port void toIp(char* buf, size_t size, const struct sockaddr* addr); // 獲取ip // 根據(jù)ip和port得到對(duì)應(yīng)的sockaddr_in void fromIpPort(const char* ip, uint16_t port, struct sockaddr_in* addr); void fromIpPort(const char* ip, uint16_t port, struct sockaddr_in6* addr); // sockaddr和sockaddr_in(ip和端口分開存儲(chǔ))的轉(zhuǎn)換 int getSocketError(int sockfd); const struct sockaddr* sockaddr_cast(const struct sockaddr_in* addr); const struct sockaddr* sockaddr_cast(const struct sockaddr_in6* addr); struct sockaddr* sockaddr_cast(struct sockaddr_in6* addr); const struct sockaddr_in* sockaddr_in_cast(const struct sockaddr* addr); const struct sockaddr_in6* sockaddr_in6_cast(const struct sockaddr* addr);struct sockaddr_in6 getLocalAddr(int sockfd); struct sockaddr_in6 getPeerAddr(int sockfd); bool isSelfConnect(int sockfd); // 判斷子連接ssize_t read(int sockfd, void *buf, size_t count); ssize_t readv(int sockfd, const struct iovec *iov, int iovcnt); ssize_t write(int sockfd, const void *buf, size_t count);

Socket

Socket是對(duì)socket fd的封裝，通過(guò)調(diào)用SocketsOps來(lái)實(shí)現(xiàn)。

// 獲取tcp的信息 bool getTcpInfo(struct tcp_info*) const; bool getTcpInfoString(char* buf, int len) const;void bindAddress(const InetAddress& localaddr); // 綁定地址 void listen(); // 監(jiān)聽湍口 int accept(InetAddress* peeraddr); // 獲取連接 void shutdownWrite(); // 關(guān)閉寫端而不是直接close// TCP_NODELAY void setTcpNoDelay(bool on); // SO_REUSEADDR void setReuseAddr(bool on); // SO_REUSEPORT void setReusePort(bool on); // SO_KEEPALIVE void setKeepAlive(bool on);

muduo在斷開連接時(shí)，不是直接close socket，而是關(guān)閉寫端，意味著還可以讀，這樣可以完整接受對(duì)方的數(shù)據(jù)。

Channel

Channel類負(fù)責(zé)注冊(cè)每個(gè)fd的事件回調(diào)函數(shù)，每個(gè)Channel只負(fù)責(zé)一個(gè)fd的事件分發(fā)，不擁有fd，不會(huì)在析構(gòu)的時(shí)候關(guān)閉fd，Channel不是基類，不需要繼承，一般作為其他類的成員。

Poller

Poller是IO復(fù)用的封裝，在muduo中是一個(gè)抽象基類，作為poll和epoll兩種IO復(fù)用機(jī)制的父類。Poller是EventLoop的間接成員，只供owner EventLoop在IO線程中調(diào)用。poll返回之后，通過(guò)遍歷pollfds_數(shù)組，找到對(duì)應(yīng)的活動(dòng)事件，復(fù)雜度O(n)，在Poller中有一個(gè)map<int, Channel*>的映射channels_,

插入新的Channel的復(fù)雜度是O(logN)，更新已有的Channel的復(fù)雜度是O(1)，因?yàn)镃hannel記錄了它的pollfds_數(shù)組中的下標(biāo)，可以快速定位。刪除Channel的復(fù)雜度也是O(n)。

EventLoop

EventLoop中的loop不斷的調(diào)用poll，來(lái)獲取當(dāng)前的活動(dòng)事件，然后調(diào)用每個(gè)channel的handleEvent()方法，來(lái)處理事件。

EventLoop的runInLoop()函數(shù)，可以在IO線程中執(zhí)行某個(gè)用戶的任務(wù)回調(diào)，如果當(dāng)前IO線程調(diào)用runInLoop()直接執(zhí)行，否則放入到隊(duì)列中等待執(zhí)行，queueInLoop()，這樣可以將TimerQueue的成員函數(shù)移動(dòng)到其他IO線程，這樣可以在不加鎖的情況下保證線程安全。

IO線程一般阻塞在poll調(diào)用，為了讓IO線程可以立即執(zhí)行用戶回調(diào)，muduo的做法是通過(guò)調(diào)用wakeup來(lái)喚醒IO線程，具體是向wakeupfd_中讀寫一個(gè)字節(jié)來(lái)實(shí)現(xiàn)，通過(guò)wakeup()和handleRead()對(duì)wakeupFd_讀寫數(shù)據(jù)。

queuInLoop()的具體實(shí)現(xiàn)是，將cb放到隊(duì)列中，在必要時(shí)喚醒IO線程，喚醒的條件有兩個(gè)：調(diào)用queueInLoop的不是IO線程、正在執(zhí)行隊(duì)列中的回調(diào)函數(shù)doPendingFunctors()，原因是執(zhí)行回調(diào)的函數(shù)有可能也會(huì)執(zhí)行queueInLoop()，這樣就要wakeup喚醒IO線程及時(shí)做處理，否則新添加的回調(diào)函數(shù)cb就不能及時(shí)被調(diào)用。

Reactor模型核心內(nèi)容時(shí)序圖。

TimerQueue

TimerQueue定時(shí)器，一般通過(guò)select、poll的等待時(shí)間來(lái)實(shí)現(xiàn)定時(shí)，在muduo中使用timerfd，將對(duì)時(shí)間的處理和IO事件統(tǒng)一起來(lái)。

muduo的定時(shí)器由三個(gè)類：TimerId、Timer、TimerQueue。

TimerQueue的接口有addTimer()和cancel()，addTimer()是供EventLoop使用，EventLoop封裝為更好用的runAt()、runAfter()、runEvery()。

TimerQueue使用set管理Timer，set中的key是pair<Timestamp,Timer*>，這樣可以方便處理相同到期時(shí)間的Timer。

TimerQueue使用一個(gè)Channel來(lái)官差timerfd_上的可讀事件。

TimerQueue目前有一個(gè)不理性的地方，Timer使用裸指針的方法管理，需要手動(dòng)delete，在C++11中可以改為unique_ptr，避免手動(dòng)釋放資源。

通過(guò)TimerQueue的getExpired()來(lái)獲取超時(shí)事件。

TimerQueue回調(diào)用戶代碼onTimer()的時(shí)序圖。

一次事件循環(huán)是從poll返回到再次調(diào)用poll阻塞。

循環(huán)中的各種回調(diào)發(fā)生的順序。

Acceptor

Acceptor用于accept()新的TCP連接，通過(guò)回調(diào)函數(shù)通知使用者，供TcpServer使用，生命期由TcpServer控制。

成員函數(shù)包括Socket、Channel。Socket封裝了socket文件描述符生命期，Channel用于觀察socket上的可讀事件，回調(diào)handleRead()，accept來(lái)接受新的連接，并回調(diào)用戶的callback。

TcpConnection

TcpConnection是唯一默認(rèn)使用shared_ptr來(lái)管理的class，是muduo最復(fù)雜的class。

TcpConnection使用Channel來(lái)獲取socket上的IO事件，自己處理可寫事件，把可讀事件通過(guò)MessageCallback傳給用戶，TcpConnection擁有Tcp socket，在析構(gòu)中會(huì)close fd。

TcpConnection關(guān)閉連接的方式是被動(dòng)關(guān)閉，對(duì)方先關(guān)閉連接，read返回0，觸發(fā)關(guān)閉邏輯。

Tcp的關(guān)閉流程，X表示TcpConnection通常在這里析構(gòu)。

TcpConnection增加CloseCallback事件回調(diào)，提供給TcpServer和TcpClient使用，通知移除TcpConnectionPtr，普通用戶使用ConnectionCallback。

TcpConnection的狀態(tài)圖。

Connector

socket是一次性的，一旦出錯(cuò)（對(duì)方拒絕連接），就無(wú)法恢復(fù)，只能重來(lái)。但是Connector是可以反復(fù)使用的，每次嘗試連接都要使用新的socket文件描述符和新的Channel對(duì)象。

重試的間隔應(yīng)該逐漸延長(zhǎng)，例如0.5s、1s、2s、4s直到30秒，對(duì)于對(duì)象的生命期管理方面，如果使用EventLoop::runAfter()定時(shí)，而Connector在定時(shí)器到期之前析構(gòu)了怎么辦？可以在Connector的析構(gòu)函數(shù)中注銷定時(shí)器。

對(duì)于自連接的問(wèn)題的處理，在發(fā)起連接時(shí)，首先在本地選擇IP（由路由表確定）和隨機(jī)選擇端口，如果目標(biāo)IP剛好是主機(jī)而且端口也相同，這就發(fā)生了自連接，處理辦法就是斷開連接重試。

Buffer

muduo在讀取數(shù)據(jù)時(shí)，采用cantter/gatherIO(分散聚集IO)，在一次系統(tǒng)調(diào)用可以對(duì)多個(gè)緩沖區(qū)進(jìn)行輸入輸出，而且一部分的緩沖區(qū)來(lái)自stack，這樣緩沖區(qū)足夠大，通常一次readv調(diào)用就可以取完數(shù)據(jù)，

muduo采用的是水平觸發(fā)，這樣做不會(huì)丟失數(shù)據(jù)或消息，每次讀取數(shù)據(jù)只需要一次系統(tǒng)調(diào)用，照顧了多個(gè)連接的公平性，不會(huì)因?yàn)槟硞€(gè)連接上的數(shù)據(jù)量過(guò)大而影響其他連接處理消息。

發(fā)送數(shù)據(jù)的邏輯是，先嘗試發(fā)送數(shù)據(jù)，如果只發(fā)送了部分?jǐn)?shù)據(jù)，把剩余的數(shù)據(jù)放到outputBuffer_，開始關(guān)注writable事件，在handleWrite()中記錄發(fā)送數(shù)據(jù)，如果outputBuffer_中已經(jīng)有待發(fā)送的的數(shù)據(jù)，就不能嘗試發(fā)送，否則造成數(shù)據(jù)錯(cuò)亂。

TcpServer

TcpServer用于處理新建TcpConnection。

TcpServer新建連接的函數(shù)調(diào)用。

TcpServer用來(lái)管理accpet獲得的TcpConnection，供用戶使用，生命期由用戶控制。使用Accpetor獲取新連接的fd，保存用戶提供的ConnectionCallback和MessageCallback，在新建TcpConnection之后，將這兩個(gè)回調(diào)函數(shù)傳遞給后者。

隨機(jī)選擇pool中的EventLoop，不允許TcpConnection在運(yùn)行中更換EventLoop，每個(gè)TcpServer有自己的EventLoopThreadPool。

TcpClient

TcpClient主要使用Connector來(lái)進(jìn)行連接，Connector具備反復(fù)嘗試連接的功能，因此客戶端和服務(wù)端啟動(dòng)的順序就無(wú)關(guān)緊要了。

連接斷開后初次嘗試連接應(yīng)該具有隨機(jī)性，如果服務(wù)端崩潰大量客戶端重連，同時(shí)重連也會(huì)發(fā)生丟包，每個(gè)TcpClient應(yīng)該 等待一段隨機(jī)時(shí)間（0.5-2s）再嘗試連接避免擁塞。

發(fā)起連接的時(shí)候如果發(fā)生TCP SYN丟包，那么系統(tǒng)默認(rèn)的重試間隔是3s，職期間不會(huì)發(fā)生錯(cuò)誤碼。

epoll

epoll是linux獨(dú)有的高效的IO復(fù)用機(jī)制，它與poll的不同之處主要在于poll每次返回整個(gè)文件描述符數(shù)組，用戶代碼需要遍歷數(shù)組以找到哪些文件描述符上有IO事件，epoll_wait()返回活動(dòng)fd的列表，需要遍歷的數(shù)組通常會(huì)小的多，再并發(fā)連接較大而活動(dòng)連接比例不高時(shí)，epoll比poll更高效。

muduo定義Poller基類并提供兩份實(shí)現(xiàn)PollPoller和EPollPoller。

HttpServer

HttpRequest封裝了HTTP請(qǐng)求包的基本格式。

HttpResponse封裝了HTTP的響應(yīng)包的基本格式。

HttpContext主要是對(duì)HTTP請(qǐng)求包的解析，將解析的結(jié)果存在對(duì)象HttpRequest中。

HttpServer封裝了HTTP的對(duì)請(qǐng)求內(nèi)容的響應(yīng)，通過(guò)用戶指定回調(diào)函數(shù)來(lái)處理請(qǐng)求，使用TcpServer來(lái)進(jìn)行對(duì)連接進(jìn)行處理。

總結(jié)

以上是生活随笔為你收集整理的muduo 笔记的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。