詳解Theron通過Actor模型解決C++并發(fā)編程的一種思維

時(shí)間：2022-05-25 20:33:01 | 來源：網(wǎng)絡(luò)營(yíng)銷

時(shí)間：2022-05-25 20:33:01 來源：網(wǎng)絡(luò)營(yíng)銷

現(xiàn)今，單臺(tái)機(jī)器擁有多個(gè)獨(dú)立的計(jì)算單元已經(jīng)太常見了，這點(diǎn)在服務(wù)器的處理器上表現(xiàn)尤為明顯，據(jù)AMD的一張2012-2013服務(wù)器路線圖顯示，服務(wù)器處理器的核心數(shù)將在2013年達(dá)到20顆之多，合理的利用CPU資源已是一個(gè)不得不考慮的問題。

不少C++程序員依然使用著多線程模型，但是對(duì)多線程的掌控并不是一件容易的事情，開發(fā)中容易出錯(cuò)、難以調(diào)試，有些開發(fā)者為了避免多線程帶來的復(fù)雜度而棄用多線程，有些開發(fā)者則另投其他語言陣營(yíng)，例如：Erlang，其實(shí)我們還有其他的選擇，Theron就是其中之一。

1、什么是Theron？

Theron是一個(gè)用于并發(fā)編程的C++庫，通過Theron我們可以避免多線程開發(fā)中各種痛處，例如：共享內(nèi)存、線程同步，Theron通過Actor模型向我們展示了另一種思維。

2、什么是Actor模型？

Erlang因?yàn)槠鋬?yōu)秀的并發(fā)特性而被大家所關(guān)注，而其并發(fā)特性的關(guān)鍵之一就是在于其采用了Actor模型，與Actor模型相對(duì)應(yīng)的模型則是我們?cè)诿嫦驅(qū)ο缶幊讨惺褂玫腛bject模型，Object模型中宣揚(yáng)，一切皆為Object（對(duì)象），而Actor模型則認(rèn)為一切皆為Actor。

Actor模型中，Actor之間通過消息相互通信，這是其和Object模型的一個(gè)顯著的區(qū)別，換而言之Actor模型使用消息傳遞機(jī)制來代替了Object模型中的成員方法調(diào)用。

在億企邦看來，這樣做意義重大，因?yàn)橄鄬?duì)于成員方法的調(diào)用來說，消息的發(fā)送是非阻塞的，它無需等待被調(diào)用方法執(zhí)行完成就可以返回，下圖顯示了此種區(qū)別：

A::a()調(diào)用了objB.b()，此時(shí)A::a()必須等待B::b()的返回才能繼續(xù)執(zhí)行，在Actor模型中，對(duì)應(yīng)的做法是Actor A向Actor B發(fā)送消息并立即返回，這時(shí)候Actor A可以繼續(xù)執(zhí)行下去，與此同時(shí)Actor B收到消息被喚醒并和Actor A并行執(zhí)行下去。

Theron中的每個(gè)Actor都會(huì)綁定一個(gè)唯一的地址，通過Actor的地址就可以向其發(fā)送消息了，每個(gè)Actor都有一個(gè)消息隊(duì)列。

從編碼者的角度看來，每實(shí)例化一個(gè)Actor都創(chuàng)建了一個(gè)和Actor相關(guān)的“線程”（非系統(tǒng)級(jí)的線程），每個(gè)Actor總是被單線程的執(zhí)行。

總的來說，Theron的并發(fā)特性的關(guān)鍵就在于：每個(gè)Actor在屬于自己的單個(gè)“線程”中執(zhí)行，而多個(gè)Actor并發(fā)執(zhí)行。

3、Hello Theron

在談及更多內(nèi)容之前，我們先來看看Theron的一個(gè)簡(jiǎn)單的范例，借以獲得一個(gè)最直觀的印象，Theron提供了makefile便于gcc用戶編譯，同時(shí)其也為Windows用戶提供了Visual Studio solution文件Theron.sln用于構(gòu)建Theron。

編譯Theron很容易，不會(huì)有太多的障礙，需要注意的是構(gòu)建Theron需要指定依賴的線程庫，Theron支持三種線程庫：std::thread（C++11 標(biāo)準(zhǔn)線程庫）、Boost.Thread和Windows threads。

使用makefile構(gòu)建時(shí)，通過threads參數(shù)指定使用的線程庫，使用Visual Studio構(gòu)建時(shí)，通過選擇適當(dāng)?shù)腟olution configuration來指定使用的線程庫，下面我們來看一個(gè)最簡(jiǎn)單的范例：

#include <stdio.h>
　　#include <Theron/Framework.h>
　　#include <Theron/Actor.h>
　　// 定義一個(gè)消息類型
　　// 在 Theron 中，任何類型都可以作為一個(gè)消息類型
　　// 唯一的一個(gè)約束是消息類型的變量能夠被拷貝的
　　// 消息按值發(fā)送（而非發(fā)送它們的地址）
　　struct StringMessage
　　{
　　 char m_string[64];
　　};
　　// 用戶定義的 Actor 總需要繼承于 Theron::Actor
　　// 每個(gè) Actor 和應(yīng)用程序的其他部分通信的唯一途徑就是通過消息
　　class Actor : public Theron::Actor
　　{
　　public:
　　 inline Actor()
　　 {
　　 // 注冊(cè)消息的處理函數(shù)
　　 RegisterHandler(this, &Actor::Handler);
　　 }
　　private:
　　 // 消息處理函數(shù)的第一個(gè)參數(shù)指定了處理的消息的類型
　　 inline void Handler(const StringMessage& message, const Theron::Address from)
　　 {
　　 printf("%sn", message.m_string);
　　 if (!Send(message, from))
　　 printf("Failed to send message to address %dn", from.AsInteger());
　　 }
　　};
　　int main()
　　{
　　 // Framework 對(duì)象用于管理 Actors
　　 Theron::Framework framework;
　　 // 通過 Framework 構(gòu)建一個(gè) Actor 實(shí)例并持有其引用
　　 // Actor 的引用類似于 Java、C# 等語言中的引用的概念
　　 // Theron::ActorRef 采用引用計(jì)數(shù)的方式實(shí)現(xiàn)，類似于 boost::shared_ptr
　　 Theron::ActorRef simpleActor(framework.CreateActor<Actor>());
　　 // 創(chuàng)建一個(gè)Receiver用于接收Actor發(fā)送的消息
　　 // 用于在非Actor代碼中（例如main函數(shù)中）與Actor通信
　　 Theron::Receiver receiver;
　　 // 構(gòu)建消息
　　 StringMessage message;
　　 strcpy(message.m_string, "Hello Theron!");
　　 // 通過 Actor 的地址，我們就可以向 Actor 發(fā)送消息了
　　 if (!framework.Send(message, receiver.GetAddress(), simpleActor.GetAddress()))
　　 printf("Failed to send message!n");
　　 // 等到 Actor 發(fā)送消息，避免被關(guān)閉主線程
　　 receiver.Wait();
　　 return 0;
　　}

這個(gè)范例比較簡(jiǎn)單，通過Actor輸出了Hello Theron，需要額外說明的一點(diǎn)是消息在Actor之間發(fā)送時(shí)會(huì)被拷貝，接收到消息的Actor只是引用到被發(fā)送消息的一份拷貝，這么做的目的在于避免引入共享內(nèi)存、同步等問題。

Theron的消息處理前面談到過，每個(gè)Actor都工作在一個(gè)屬于自己的“線程”上，我們通過一個(gè)例子來認(rèn)識(shí)這一點(diǎn)，我們修改上面例子中的Actor:: Handler成員方法：

inline void Handler(const StringMessage& message, const Theron::Address from)
　　{
　　 while (true)
　　 {
　　 printf("%s --- %dn", message.m_string, GetAddress().AsInteger());
　　#ifdef _MSC_VER
　　 Sleep(1000);
　　#else
　　 sleep(1);
　　#endif
　　 }
　　}

此Handler會(huì)不斷的打印message并且?guī)袭?dāng)前Actor的地址信息，在main函數(shù)中，我們構(gòu)建兩個(gè)Actor實(shí)例并通過消息喚醒它們，再觀察輸出結(jié)果：

Hello Theron! --- 1
　　Hello Theron! --- 2
　　Hello Theron! --- 2
　　Hello Theron! --- 1
　　Hello Theron! --- 2
　　Hello Theron! --- 1
　　Hello Theron! --- 2
　　Hello Theron! --- 1
　　......

這和我們預(yù)期的一樣，兩個(gè)Actor實(shí)例在不同的線程下工作，實(shí)際上，F(xiàn)ramework創(chuàng)建的時(shí)候會(huì)創(chuàng)建系統(tǒng)級(jí)的線程，默認(rèn)情況下會(huì)創(chuàng)建兩個(gè)（可以通過 Theron::Framework 構(gòu)造函數(shù)的參數(shù)決定創(chuàng)建線程的數(shù)量），它們構(gòu)成一個(gè)線程池，我們可以根據(jù)實(shí)際的CPU核心數(shù)來決定創(chuàng)建線程的數(shù)量，以確保CPU被充分利用。

那么，線程池的線程是以何種方式進(jìn)行調(diào)度的呢？如下圖所示：

接收到消息的Actor會(huì)被放置于一個(gè)線程安全的Work隊(duì)列中，此隊(duì)列中的Actor會(huì)被喚醒的工作線程取出，并進(jìn)行消息的處理，這個(gè)過程中有兩個(gè)需要注意的地方：

（1）、對(duì)于某個(gè)Actor我們可以為某個(gè)消息類型注冊(cè)多個(gè)消息處理函數(shù)，那么，此消息類型對(duì)應(yīng)的多個(gè)消息處理函數(shù)會(huì)按照注冊(cè)的順序被串行執(zhí)行下去。

（2）、線程按順序處理Actor收到的消息，一個(gè)消息未處理完成不會(huì)處理消息隊(duì)列中的下一個(gè)消息我們可以想象，如果存在三個(gè)Actor，其中兩個(gè)Actor的消息處理函數(shù)中存在死循環(huán)（例如上例中的while(true)），那么它們一旦執(zhí)行就會(huì)霸占兩條線程，若線程池中沒有多余線程，那么另一個(gè)Actor將被“餓死”（永遠(yuǎn)得不到執(zhí)行）。

我們可以在設(shè)計(jì)上避免這種 Actor 的出現(xiàn)，當(dāng)然也可以適當(dāng)?shù)恼{(diào)整線程池的大小來解決此問題，Theron中，線程池中線程的數(shù)量是可以動(dòng)態(tài)控制的，線程利用率也可以測(cè)量，但是務(wù)必注意的是，過多的線程必然導(dǎo)致過大的線程上下文切換開銷。

4、一個(gè)詳細(xì)的例子

我們?cè)賮砜匆粋€(gè)詳細(xì)的例子，借此了解Theron帶來的便利，生產(chǎn)者消費(fèi)者的問題是一個(gè)經(jīng)典的線程同步問題，我們來看看Theron如何解決這個(gè)問題：

#include <stdio.h>
　　#include <Theron/Framework.h>
　　#include <Theron/Actor.h>
　　const int PRODUCE_NUM = 5;
　　class Producer : public Theron::Actor
　　{
　　public:
　　 inline Producer(): m_item(0)
　　 {
　　 RegisterHandler(this, &Producer::Produce);
　　 }
　　private:
　　 // 生產(chǎn)者生產(chǎn)物品
　　 inline void Produce(const int& /* message */, const Theron::Address from)
　　 {
　　 int count(PRODUCE_NUM);
　　 while (count--)
　　 {
　　 // 模擬一個(gè)生產(chǎn)的時(shí)間
　　#ifdef _MSC_VER
　　 Sleep(1000);
　　#else
　　 sleep(1);
　　#endif
printf("Produce item %dn", m_item);
if (!Send(m_item, from))
printf("Failed to send message!n");
++m_item;
　 }
　　 }
　　 // 當(dāng)前生產(chǎn)的物品編號(hào)
　　 int m_item;
　　};
　　class Consumer : public Theron::Actor
　　{
　　public:
　　 inline Consumer(): m_consumeNum(PRODUCE_NUM)
　　 {
　　 RegisterHandler(this, &Consumer::Consume);
　　 }
　　private:
　　 inline void Consume(const int& item, const Theron::Address from)
　　 {
　　 // 模擬一個(gè)消費(fèi)的時(shí)間
　　#ifdef _MSC_VER
　　 Sleep(2000);
　　#else
　　 sleep(2);
　　#endif
　　 printf("Consume item %dn", item);
　　 --m_consumeNum;
　　 // 沒有物品可以消費(fèi)請(qǐng)求生產(chǎn)者進(jìn)行生產(chǎn)
　　 if (m_consumeNum == 0)
　　 {
　　 if (!Send(0, from))
　　 printf("Failed to send message!n");
　　 m_consumeNum = PRODUCE_NUM;
　　 }
　　 }
　　 int m_consumeNum;
　　};
　　int main()
　　{
　　 Theron::Framework framework;
　　 Theron::ActorRef producer(framework.CreateActor<Producer>());
　　 Theron::ActorRef consumer(framework.CreateActor<Consumer>());
　　 if (!framework.Send(0, consumer.GetAddress(), producer.GetAddress()))
　　 printf("Failed to send message!n");
　　 // 這里使用 Sleep 來避免主線程結(jié)束
　　 // 這樣做只是為了簡(jiǎn)單而并不特別合理
　　 // 在實(shí)際的編寫中，我們應(yīng)該使用Receiver
　　#ifdef _MSC_VER
　　 Sleep(100000);
　　#else
　　 sleep(100);
　　#endif
　　 return 0;
　　}

生產(chǎn)者生產(chǎn)物品，消費(fèi)者消費(fèi)物品，它們并行進(jìn)行，我們沒有編寫創(chuàng)建線程的代碼，沒有構(gòu)建共享內(nèi)存，也沒有處理線程的同步，這一切都很輕松的完成了。

5、代價(jià)和設(shè)計(jì)

和傳統(tǒng)的多線程程序相比Theron有不少優(yōu)勢(shì)，通過使用Actor，程序能夠自動(dòng)的并行執(zhí)行，而無需開發(fā)者費(fèi)心，Actor總是利用消息進(jìn)行通信，消息必須拷貝，這也意味著我們必須注意到，在利用Actor進(jìn)行并行運(yùn)算的同時(shí)需避免大量消息拷貝帶來的額外開銷。

Actor模型強(qiáng)調(diào)了一切皆為Actor，這自然可以作為我們使用Theron的一個(gè)準(zhǔn)則，但過多的Actor存在必然導(dǎo)致Actor間頻繁的通信。

適當(dāng)?shù)氖褂肁ctor并且結(jié)合Object模型也許會(huì)是一個(gè)不錯(cuò)的選擇，例如，我們可以對(duì)系統(tǒng)進(jìn)行適當(dāng)劃分，得到一些功能相對(duì)獨(dú)立的模塊，每個(gè)模塊為一個(gè)Actor，模塊內(nèi)部依然使用Object模型，模塊間通過Actor的消息機(jī)制進(jìn)行通信。

億企邦點(diǎn)評(píng)：

Theron是個(gè)有趣的東西，也許你沒有使用過它，你也不了解Actor模型，但是Actor的思想?yún)s不新鮮，甚至你可能正在使用，目前來說，我還沒有找到Theron在哪個(gè)實(shí)際的商業(yè)項(xiàng)目中使用，因此對(duì)Theron的使用還存在一些未知的因素。

還有一些特性，諸如跨主機(jī)的分布式的并行執(zhí)行是Theron不支持的，這些都限制了Theron的使用，不過我也正在積極的改變一些東西，無論Theron未來如何，Theron以及Actor模型帶來的思想會(huì)讓我們更加從容面對(duì)多核的挑戰(zhàn)。