Riak是一個分布式、容錯和開放源代碼的數據庫，它展示了如何使用Erlang/OTP來構建大型可伸縮系統。Riak提供了一些其他數據庫中并不常見的特性，比如高可用性、容量和吞吐量的線性伸縮能力等，很大程度上，這是借由Erlang對大規模可伸縮分布式系統的支持實現的。 要開發像Riak這樣的系統，Erlang/OTP是一個理想的平臺，因為它提供了可以直接利用的節點間通信、消息隊列、故障探測和客戶-服務器抽象等功能。而且，Erlang中大多數常見的模式都已經以庫模塊的形式實現了，我們一般稱之為OTP behaviors。其中包括了用于并發和錯誤處理的通用代碼框架，可以簡化并發編程，也能避免開發者陷入一些常見的陷阱。Behaviors由管理者負責監管，而管理者本身也是behavior，這樣就組成了一個監管樹。通過將監管樹打包到應用程序中，這就創建了一個Erlang程序的構建塊。 一個完整的Erlang系統，如Riak，是由一組松散耦合且相互作用的應用組成的。其中有些應用是開發者編寫的，有些是標準Erlang/OTP發布包中的，還有一些可能是其他的開源組件。這些應用由一個boot腳本按順序加載并啟動，而該腳本是從應用清單和版本信息中生成的。 系統之間的區別在于，啟動的發布版本中的應用有所不同。在標準的Erlang發行版中，boot文件會啟動Kernel和StdLib（Standard Library，標準庫）等應用。而在有些安裝版本中，還會啟動SASL（Systems Architecture Support Library，系統架構支持庫）應用。SASL中包含了帶有日志功能的發布和軟件更新工具。對Riak而言，除了啟動其特定的應用以及運行時依賴（其中包括Kernel、StdLib和SASL）之外，并沒有什么不同。一個完整的、準備好運行的Riak構建版本，實際上將Erlang/OTP發行包中的這些標準元素都嵌入其中了，當在命令行調用`riak start`?時，它們會一同啟動。Riak由很多復雜的應用組成，所以本章不應看做一個完整的指南。倒是可以把本章看做以Riak源代碼為例，針對OTP的入門指南。圖片和數字主要是為了闡明設計意圖，故有所簡化。 ## 15.1 Erlang簡介 Erlang是一個并發的函數式編程語言，用它編寫的程序會編譯為字節代碼并運行在虛擬機上。程序中互相調用的函數經常會產生副作用，如進程間消息傳遞，I/O和數據庫操作等。而Erlang變量是單賦值的，也就是說，一旦變量被給定了一個值，就再也不能修改了。從下面的計算階乘的例子可以看出，Erlang中大量使用了模式匹配： ~~~ -module(factorial). -export([fac/1]). fac(0) -> 1; fac(N) when N>0 -> Prev = fac(N-1), N*Prev. ~~~ 在這段代碼中，第一個子句（clause）給出了0的階乘，第二個字句計算正數的階乘。每一個子句的主體部分都是一個表達式序列，主體部分中最后一個表達式就是這個子句的計算結果。調用這個函數的時候如果傳入一個負數會導致運行時錯誤，因為沒有一個子句能匹配負數的模式。不處理這種情況的做法是非防御式（non-defensive）編程的一個例子，這種做法也是Erlang中鼓勵的做法。 在模塊之中，函數以正常的方式調用；而在模塊之外，函數名之前應該加上模塊名，如`factorial:fac(3)`。允許定義同名但是參數數目不同的函數——函數的參數數目稱為函數的元數（arity）。在`factorial`模塊的export指令中，元數為1的`fac`函數通過`fac/1`表示。 Erlang支持元組（tuple，也稱為乘積類型（product type））和列表（list）。元組由花括號包圍起來，例如`{ok,37}`。在元組中，通過元素的位置訪問元素。記錄（record）是另一種數據類型；在記錄中可以保存固定數目的元素，這些元素可以通過名字訪問和操作。例如這樣的語法可以定義一個記錄：`-record(state, {id, msg_list=[]})`。通過表達式`Var = #state{id=1}`可以創建一個實例，然后通過這樣的表達式可以查看實例中的內容：`Var#state.id`。如果要使用可變數目的元素，那么我們可以使用列表，列表通過方括號定義，例如`[23,34]`。`[X|Xs]`的表達方式匹配一個非空的列表，其中X匹配頭，Xs匹配尾。用小寫字母開頭的標識符表示一個原子（atom），原子就是一個表示自己的字符串；例如，元組`{ok,37}`中的`ok`就是一個原子。通常通過這種方式使用原子來表示函數的結果，例如除了`ok`結果之外，還可以有`{error, "Error String"}`這種形式的結果。 Erlang系統中的進程在獨立的內存中并發運行，以消息傳遞的方式進行相互通信。進程可以應用于大量的應用，其中包括數據庫的網關，協議棧的處理程序，以及管理從其他進程發送來的跟蹤消息的日志。雖然這些進程處理不同的請求，但是進程處理請求的方式卻是有相似之處的。 因為進程只存在于虛擬機中，一個VM可以同時運行成千上萬個進程，Riak就大量使用了這一特性。例如，對數據的每一個請求——讀、寫和刪除——都采用獨立進程處理的模型，這種方式對于大多數采用操作系統級線程的實現而言都是不可能的。 進程是通過進程標識符識別的，進程標識符稱為PID；此外，進程還可以通過別名注冊，不過注冊別名的方式應該只用于長時間運行的“靜態”進程。如果一個進程注冊了一個別名，那么其他進程就可以在不知道這個進程PID的情況下給這個進程發送消息。進程的創建通過內建函數（built-in function，BIF）?`spawn(Module, Function, Arguments)`完成。BIF是集成在虛擬機中的函數，用于完成純Erlang不可能實現或實現很慢的功能。`spawn/3`這個BIF接受一個`Module`、一個`Function`和一個`Arguments`作為參數。這個BIF的調用返回新創建的進程的PID，并且產生一個副作用，就是創建了一個新的進程以之前傳入的參數執行模塊中的函數。 我們通過`Pid ! Msg`這種寫法將消息`Msg`發送給進程`Pid`。一個進程可以通過調用BIF?`self`來得到其PID，之后該進程可以將PID發送給其他進程，這樣別的進程就能夠利用它與原來的進程通信了。假設一個進程期望接收`{ok, N}`和`{error, Reason}`這種形式的消息。這個進程可以通過receive語句處理這些消息： ~~~ receive {ok, N} -> N+1; {error, _} -> 0 end ~~~ 這條語句的結果是由模式匹配語句確定的數值。如果在模式匹配中并不需要某個變量的值，可以像上面例子中那樣用下劃線來代替。 進程之間的消息傳遞是異步的，進程接收到的消息會按照其到達順序放在其信箱中。假設現在正在執行的就是上面的receive表達式：如果信箱中的第一個元素是`{ok, N}`或`{error, Reason}`，那就可以返回相應結果。如果第一個元素并非這兩種形式之一，那它會繼續保留在信箱之中，然后以類似的方式處理第二個消息。如果沒有消息能匹配成功，receive會繼續等待，直到接收到一個匹配的消息。 進程終止有兩種原因。如果沒有更多的代碼要執行了，它們會以原因*normal*退出。如果進程遇到了運行時錯誤，它會以非*normal*的原因退出。進程的終止只會對和其“鏈接”在一起的進程產生影響。進程可以通過BIF?`link(Pid)`鏈接在一起，也可以在調用`spawn_link(Module, Function, Arguments)`的時候鏈接在一起。如果一個進程終止了，那么這個進程會對其鏈接集合中的所有進程發送一個EXIT信號。如果終止原因不是normal，那么收到這個信號的進程會終止自己，并且進一步傳播EXIT信號。如果調用BIF?`process_flag(trap_exit, true)`，那么進程收到EXIT信號之后不會終止，而是以Erlang消息的方式將EXIT信號放在進程的信箱中。 Riak通過EXIT信號監視輔助進程的健康狀況，這些輔助進程負責執行由請求驅動的有限狀態機發起的非關鍵性的工作。當這些輔助進程異常終止的時候，父進程可以通過EXIT信號決定忽略錯誤或重新啟動進程。 ## 15.2\. 進程框架 我們前面引入了這一概念，即不管進程是出于什么目的創建的，它們總要遵從一個共同的模式。作為開始，我們必須創建一個進程，然后可以為它注冊一個別名，當然后者是可選的。對于新創建的進程而言，它的第一個動作是初始化進程循環數據。循環數據一般通過在進程初始化時傳給內置函數`spawn`的參數得到。循環數據保存在叫做進程狀態的變量中。狀態（一般保存在一個記錄中）會被傳遞給接收-求值函數，該函數是一個循環，負責接收消息，處理消息，更新狀態，之后將狀態作為參數傳給一個尾遞歸調用。如果處理到了‘stop’消息，接收進程會清理自身數據，然后退出。 不管進程要執行什么任務，這都是進程之間反復出現的一種機制。記住這一點之后，我們再來看一下，遵守這一模式的進程之間又有何不同： * 創建不同的進程時傳入BIF?`spawn`的參數會有不同 * 在創建一個進程的時候，要考慮是否為這個進程注冊一個別名，如果需要的話，還要考慮別名是什么。 * 初始化進程狀態的函數要執行的動作依進程執行任務的不同而不同。 * 無論哪種情況，系統的狀態都用循環數據表示，但不同進程的循環數據會有不同。 * 在接收-求值循環體中，不同的進程接收的消息是不一樣的，而且處理的方式也五花八門。 * 最后，在進程結束的時候，清理動作也隨進程而異。 所以，即使存在一個通用的動作框架，它們仍然需要與具體任務相關的各種動作來補充。以該框架為模板，程序員能夠創建不同的進程，用以承擔服務器、有限狀態機、事件處理程序和監督者等不同職責。但是我們不必每次都重新實現這些模式，它們已經作為行為模式放在類庫中了。它們是OTP中間件的一部分。 ## 15.3\. OTP行為 開發Riak的核心開發者團隊分布在十幾個不同的地點。如果沒有非常緊密的合作和可操作的模板，那么最終可能會得到各種不同的客戶端/服務器實現，這些實現可能還不能處理特殊的邊界條件和并發相關的錯誤。此外，可能還無法形成一種處理客戶端和服務器崩潰的統一方法，而且也無法保證來自于一個請求的應答是一個合法應答而不只是某條服從內部消息協議的任意消息。 OTP指的是一組Erlang庫和設計模式，宗旨是為開發健壯系統提供一組現成的工具。其中很多模式和庫都以“行為”（behavior）的形式提供。 OTP行為提供了一些實現了最常見并發設計模式的庫模塊，從而解決了上述問題。在幕后，這些庫模塊可以確保以一致的方式處理錯誤和特殊情況，而程序員并不需要意識到這些。因此，OTP行為提供了一組標準化的構建單元，利用這些構建單元可以設計和構建工業強度的系統。 ### 15.3.1\. OTP行為簡介 OTP行為是通過`stdlib`應用程序中的一些庫模塊提供的，而后者是Erlang/OTP發行版中的一部分。由程序員編寫的具體代碼放在獨立的模塊中，這些代碼通過每一個行為中預定義的一組標準回調函數調用。這個回調模塊要包含實現某個功能所需要的所有具體代碼。 OTP行為中包含工作進程，負責實際的處理工作，還包含監督者進程，負責監視工作進程和其他監督進程。工作進程（worker）行為包括服務器、事件處理程序和有限狀態機，在圖中通常使用圓圈表示。監督者（supervisor）負責監視其子進程，既包含工作進程也包含其他監督者，在圖中通常用方框表示，工作者和監督者共同組成了監督樹（supervision tree）。 圖15.1: OTP Riak監督樹 監督樹包裝在一個名為應用程序（application）的行為中。OTP應用程序不僅是Erlang系統中構建單元，還是一種包裝可重用組件的方法。像Riak這樣的工業級別的系統由一組低耦合且可能分布式的應用程序組成。在這些應用程序中，有一些屬于標準Erlang發行版的一部分，另一些則是為了實現Riak中特定功能所編寫的。 OTP應用程序的例子還包括Corba ORB和簡單網絡管理協議（Simple Network Management Protocol，SNMP）代理。OTP應用程序是一個可重用的組件，通過監督進程和工作進程的方式將庫模塊包裝在一起。從現在開始，我們提到應用程序的時候指的就是OTP應用程序。 行為模塊包含每一種行為類型所需的所有通用代碼。盡管你也可以實現自己的行為模塊，但是一般情況下不需要這樣做，因為Erlang/OTP發行版中自帶的行為可以滿足你的代碼中會使用到的大部分設計模式。行為模塊提供的通用功能包含了以下操作： * 創建進程，還支持注冊進程； * 通過同步或異步的方式發送和接收客戶消息，包括內部消息協議的定義； * 保存循環數據和管理進程循環； * 終止進程。 循環數據是一個變量，行為需要在多個函數調用之間保存的數據都存放在這個變量中。函數調用之后返回修改后的循環數據。這個修改后的循環數據通常稱為新循環數據，這個數據作為參數被傳入下一個調用中。循環數據常被稱為行為狀態。 通用服務器應用程序使用的回調模塊中包含的功能負責提供所需求的具體行為，這些功能包括： * 初始化進程循環數據；如果需要注冊進程，還要初始化進程名稱。 * 處理具體的客戶請求；如果請求是同步的，還要將應答發送回客戶。 * 在請求處理之間維護和更新進程循環數據。 * 在進程終止的時候清理進程循環數據。 ### 15.3.2\. 通用服務器 `gen_server`行為定義了實現了客戶端/服務器行為的通用服務器（generic server），`gen_server`行為是標準庫應用程序自帶的行為。我們下面以`riak_core`應用程序中的`riak_core_node_watcher.erl`為例講解通用服務器。這個服務器負責跟蹤并報告Riak集群中有哪些子服務和節點是可用的。這個模塊頭部的指令如下所示： ~~~ -module(riak_core_node_watcher). -behavior(gen_server). %% 這個模塊提供的API -export([start_link/0,service_up/2,service_down/1,node_up/0,node_down/0,services/0, services/1,nodes/1,avsn/0]). %% gen_server的回調函數 -export([init/1,handle_call/3,handle_cast/2,handle_info/2,terminate/2, code_change/3]). -record(state, {status=up, services=[], peers=[], avsn=0, bcast_tref, bcast_mod={gen_server, abcast}}). ~~~ 通過`-behavior(gen_server)`指令可以很快判斷出這個模塊是一個通用服務器。這條指令的作用是讓編譯器確保所有的回調函數都正確導出了。服務器循環數據會使用到記錄`state`。 ### 15.3.3\. 啟動服務器 使用`gen_server`行為的時候，不要使用BIF?`spawn`或`spawn_link`，而是要使用`gen_server:start`或`gen_server:start_link`函數。`spawn`和`start`的主要區別在于后者本質上是同步的。由于`start`調用直到工作進程完成了初始化之后才會返回，所以使用`start`代替`spawn`可以使得工作進程啟動過程的確定性更好，并能避免意外的競爭條件。這個函數有兩種調用方式： ~~~ gen_server:start_link(ServerName, CallbackModule, Arguments, Options) gen_server:start_link(CallbackModule, Arguments, Options) ~~~ `ServerName`是一個元組，格式為`{local, Name}`或`{global, Name}`，表示如果進程需要注冊，則代表進程別名的本地`Name`或全局`Name`。如果使用了全局名稱，那么則可以在分布式Erlang節點組成的集群中透明地訪問服務器。如果不想注冊進程，而是通過進程PID引用進程，那么請忽略這個參數，使用`start_link/3`或`start/3`函數調用。`CallbackModule`是保存了具體回調函數的模塊名稱；`Arguments`是一個合法的Erlang term，表示傳遞給`init/1`回調函數的參數；`Options`是一個列表，可以用來設置內存管理標志`fullsweep_after`和`heapsize`以及其他跟蹤和調試標志。 在本文的例子中，調用了`start_link/4`，通過回調模塊的名字注冊進程，回調模塊的名字通過`?MODULE`宏調用得到。編譯代碼的時候，這個宏被預處理器展開為這個宏所在的模塊的名稱。將行為的別名設置為實現行為的回調模塊名稱總是一個好的做法。由于不需要傳入任何參數，所以保留空參數列表。選項列表也留空： ~~~ start_link() -> gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). ~~~ `start_link`和`start`函數的明顯區別在于前者將進程鏈接至其父進程，而后者不會鏈接，這個父進程通常是一個監督者進程。這里需要特別說明的是，將自己鏈接至監督者進程是OTP行為的責任所在。`start`函數通常用于在shell中測試行為，因為導致shell進程崩潰的輸入錯誤不會對行為產生影響。`start`和`start_link`函數的所有變體都返回`{ok, Pid}`。 `start`和`start_link`函數會創建一個新的進程調用`CallbackModule`中的回調函數`init(Arguments)`，并且傳入`Arguments`。`init`函數必須初始化服務器的`LoopData`并且返回一個格式為`{ok, LoopData}`的元組。`LoopData`包含了將要傳遞給回調函數的第一個循環數據實例。如果需要保存傳入`init`函數的一些參數，那么應該保存在`LoopData`變量中。在Riak節點監視器服務器中的`LoopData`保存的是調用`schedule_broadcast/1`并傳入一個類型為`state`的記錄得到的結果，這個記錄實例中所有字段的值都設置為默認值： ~~~ init([]) -> %% 監視節點的啟動和停止事件 net_kernel:monitor_nodes(true), %% 設置跟蹤節點狀態的ETS表 ets:new(?MODULE, [protected, named_table]), {ok, schedule_broadcast(#state{})}. ~~~ 盡管監督者進程可能會調用`start_link/4`函數，但是調用`init/1`回調函數的是另外一個進程：即剛剛創建的那個進程。由于這個服務器的作用是發現、記錄和廣播Riak中所有子服務的可用性，因此這個初始化過程要求Erlang運行時通知這個進程這種事件，并且設置一個保存這些信息的表。這些設置必須在初始化過程中完成，因為如果這個數據結構還不存在的話，任何對服務器的調用都會失敗。在`init`函數中只完成必要的設置工作，并且盡量減少`init`中完成的工作，因為`init`的調用是一個同步調用，只有這個調用返回之后才能繼續進行其他串行的過程。 ### 15.3.4\. 傳遞消息 如果需要向服務器發送一條同步的消息，可以使用`gen_server:call/2`函數。如果需要進行異步調用，則使用`gen_server:cast/2`函數。下面首先看一下Riak服務API中的兩個函數，之后再給出剩下的代碼。這兩個函數由客戶端進程調用，調用的結果是向服務器進程發送一個同步消息，其中服務器的注冊名稱和回調模塊的名稱一致。注意，發送給服務器數據的驗證應該在客戶端這邊進行。如果客戶端發送了不正確的信息，服務器應該終止。 ~~~ service_up(Id, Pid) -> gen_server:call(?MODULE, {service_up, Id, Pid}). service_down(Id) -> gen_server:call(?MODULE, {service_down, Id}). ~~~ 收到消息之后，`gen_server`進程調用`handle_call/3`回調函數處理收到的消息，處理的順序和消息發送的順序一致。 ~~~ handle_call({service_up, Id, Pid}, _From, State) -> %% 在本地更新活動服務的集合 Services = ordsets:add_element(Id, State#state.services), S2 = State#state { services = Services }, %% 移除所有和這個服務相關的mref delete_service_mref(Id), %% 為表示這個服務的PID設置一個監視器 Mref = erlang:monitor(process, Pid), erlang:put(Mref, Id), erlang:put(Id, Mref), %% 更新本地ETS表并廣播 S3 = local_update(S2), {reply, ok, update_avsn(S3)}; handle_call({service_down, Id}, _From, State) -> %% 在本地更新活動服務的集合 Services = ordsets:del_element(Id, State#state.services), S2 = State#state { services = Services }, %% 移除所有和這個服務相關的mref delete_service_mref(Id), %% 更新本地ETS表并廣播 S3 = local_update(S2), {reply, ok, update_avsn(S3)}; ~~~ 注意這個回調函數的返回值。返回值是一個元組，這個元組包含了控制用的原子`reply`，告訴`gen_server`的通用代碼這個元組的第二個元素（在上面的兩個函數中都是ok）就是要發送回客戶端的應答。這個元組中第三個元素是新的`State`，也就是說在服務器的下一次循環迭代中要將這第三個元素傳遞給`handle_call/3`函數。在上面的兩個函數中，第三個元素都更新為表示新的可用服務集合。參數`_From`是一個元組，其中包含一個唯一的消息引用和客戶端進程標識符。這個元組會在其他庫函數中使用到，但是本章不會討論這些庫函數。在大多數情況下，都不需要使用這個元組。 `gen_server`庫在這個操作的幕后內建了很多機制和保護措施。如果客戶端向服務器發送一條同步消息，而且5秒鐘內沒有收到應答，那么執行`call/2`函數的進程就會被終止。通過調用`gen_server:call(Name, Message, Timeout)`函數可以修改這個行為，其中`Timeout`既可以是毫秒值，也可以是原子`infinity`（表示不限時）。 最早設計超時機制的目的是為了防止發生死鎖，可以保證不小心互相調用的服務器在超過默認超時時間后會被終止。為了最終能調試并修復錯誤，崩潰的報告會被記錄下來。大部分應用程序在5秒鐘的超時時間內都能正常地工作，但是在非常重的負載下，可能需要調整這個值，甚至有可能使用`infinity`；這些決策都和具體的應用程序相關。在Erlang/OTP中所有的關鍵代碼都是用了`infinity`。在Riak中不同的地方使用了不同的超時時間：內部的耦合組件之間通常使用`infinity`，而在某些情況下當和Riak通信的客戶端指定某個操作應該允許超時的時候，`Timeout`由用戶傳入的參數設置。 使用`gen_server:call/2`函數時其他的防護機制還包括處理給一個不存在的服務器發送消息的情況以及服務器在發送應答之前就崩潰的情況。在這兩種情況下，調用的進程都會終止。在使用純Erlang的時候，發送一個永遠不會在receive子句中匹配模式的消息會導致內存泄露的bug。在Riak中采用了兩種不同的策略緩和這種內存泄露，這兩種策略都使用了“捕捉所有消息”的匹配子句。在消息有可能是用戶產生的情況下，不匹配的消息可能會被忽略掉。在消息只可能來自于Riak內部組件的情況下，這種情況表示出現了一個bug，因此會觸發一個錯誤報警的內部崩潰報告，并且重新啟動接受這個消息的工作進程。 發送異步消息的工作方式類似。消息被異步地發送給通用服務器，并且在`handle_cast/2`回調函數中處理。這個函數必須返回一個格式為`{reply, NewState}`的元組。使用異步調用的場合包括：不關心服務器請求的時候，以及不用擔心產生超出服務器可以承擔的消息量時。當我們對應答本身不感興趣，但是需要等待這條消息被處理完才能發送下一個請求的時候，應該使用`gen_server:call/2`，并且在應答中返回原子`ok`。考慮一個場景，一個進程生成數據庫條目的速度超出了Riak可以消耗的速度。使用異步調用的時候，可能的風險是填滿進程的信箱，使得節點耗盡內存。Riak通過`gen_server`同步調用的消息串行化特性控制負載，只有當前一個請求被處理完時才處理下一個請求。這種方法可以避免使用更為復雜的節流代碼的必要性：除了能提供并發，`gen_server`進程還可以用來產生串行化點。 ### 15.3.5\. 停止服務器 如何停止服務器？在`handle_call/3`和`handle_cast/2`回調函數中，不要返回`{reply, Reply, NewState}`和`{noreply, NewState}`，而是分別返回`{stop, Reason, Reply, NewState}`和`{stop, Reason, NewState}`。需要有某種機制觸發這個返回值，這種機制通常是發送給服務器的`stop`消息。接收到包含`Reason`和`State`的`stop`元組之后，通用代碼執行`terminate(Reason, State)`回調函數。 `terminate`函數非常適合執行清理服務器`State`的代碼和系統使用的所有其他持久化數據的代碼。在我們的例子中，向其他對等進程發送最后一條消息，讓它們知道這個節點監視器不再繼續工作和監視了。在這個例子中，變量`State`包含一個帶有`status`字段和`peers`字段的記錄： ~~~ terminate(_Reason, State) -> %% 通知所有對等進程這個進程正在關閉 broadcast(State#state.peers, State#state { status = down }). ~~~ 將行為的回調函數當做庫函數并且在程序中其他部分調用是一個非常糟糕的做法。例如，絕對不要在另一個模塊中調用riak_core_node_watcher:init(Args)獲得初始的循環數據。要獲得這種數據，應該通過同步調用服務器的方式獲得。行為回調函數只能被行為庫模塊調用，而且是因為系統中發生的事件而觸發的調用，絕對不能直接被用戶調用。 ## 15.4\. 其他工作進程行為 利用同樣的思想，還可以實現很多其他類型的工作進程行為，而且在OTP中已經實現了不少行為。 ### 15.4.1\. 有限狀態機 有限狀態機（Finite state machines，FSM）實現在`gen_fsm`行為模塊中，是電信系統（Erlang最初設計時面向的問題領域）中實現協議棧會使用到的一個關鍵組件。狀態以回調函數的形式定義，回調函數根據狀態的名稱命名，回調函數要返回一個包含下一個狀態和更新后的循環數據的元組。可以向這些狀態發送同步和異步事件。有限狀態機的回調模塊還要導出標準的回調函數，例如`init`、`terminate`和`handle_info`。 當然，有限狀態機并不僅限于使用電信系統。在Riak中，請求處理程序中使用了有限狀態機。當客戶端發出一個請求（例如`get`、`put`和`delete`）時，監聽這個請求的進程會創建出一個實現了對應`gen_fsm`行為的進程。例如，`riak_kv_get_fsm`負責處理`get`請求：檢索數據并將數據發送給客戶端進程。當這個FSM進程判斷對哪些節點請求數據時，向這些節點發送消息時以及從這些節點接收到數據、錯誤或超時應答時，都會經歷一些不同的狀態。 ### 15.4.2\. 事件處理程序 事件處理程序和事件管理器是`gen_event`庫模塊中實現的另一個行為。基本思想是創建一個接收某一種特定類型事件的集中點。事件可以同步發送也可以異步發送，事件帶有一組接收到事件需要采取的預定義動作。接收到事件之后可能發生的響應包括將事件記錄到文件中、以短消息的形式發送報警、或者采集統計信息等。每一個這一類動作都定義在一個獨立的回調模塊中，并且在每一次調用中都有自己的循環數據。每一個特定的事件管理器都可以添加、刪除或更新處理程序。因此在實際應用中，每一個事件管理器都有可能有很多回調模塊，而這些回調模塊的不同實例也可以出現在不同的管理器中。事件處理程序包括接收報警的進程、接收動態跟蹤數據的進程、接收設備相關事件的進程或接收簡單日志的進程。 Riak中使用`gen_event`行為的一個例子是管理“環事件”的訂閱，即一個Riak集群的從屬關系或分區分配的變化。Riak節點上的進程可以在實現了`gen_event`行為的`riak_core_ring_events`實例中注冊一個函數。每當管理這個節點環的中央進程修改了整個集群的從屬關系記錄時，這個進程就會產生一個事件，結果就是使得這些回調模塊都會調用注冊的函數。通過采用這種方式，Riak中的各個部分很容易對Riak的某個最為中心的數據結構發生的變化進行響應，而不會增加這個數據結構中央管理的復雜性。 我們剛才討論的3個主要行為——`gen_server`、`gen_fsm`和`gen_event`——能夠處理最為常見的并發和通信模式。然而，在大型系統中，一些和應用具體相關的模式會隨著時間推移而出現，因而有必要能夠創建新的行為。Riak就包含了這樣一種行為：`riak_core_vnode`，這種行為形式化描述了虛擬節點的實現方式。虛擬節點是Riak中最重要的存儲抽象，對請求驅動的有限狀態機暴露了一個統一的鍵值存儲接口。回調模塊的接口通過`behavior_info/1`函數具體說明，如下所示： ~~~ behavior_info(callbacks) -> [{init,1}, {handle_command,3}, {handoff_starting,2}, {handoff_cancelled,1}, {handoff_finished,2}, {handle_handoff_command,3}, {handle_handoff_data,2}, {encode_handoff_item,2}, {is_empty,1}, {terminate,2}, {delete,1}]; ~~~ 以上示例展示了`riak_core_vnode`中`behavior_info/1`函數的代碼。元組`{CallbackFunction, Arity}`的列表定義了回調模塊必須遵守的協議。具體的虛擬節點實現必須導出這些函數，否則編譯器會發出警告。自定義OTP行為的實現比較簡單直接。除了定義回調函數之外，還需要通過`proc_lib`模塊中的特定函數啟動自定義的行為，以及通過`sys`模塊處理系統消息并監視父進程以防父進程終止。 ## 15.5\. 監督者行為 監督者行為的任務就是監視其子進程，并且根據預先配置好的規則在子進程終止的時候執行相應的動作。子進程既可以是監督者進程也可以是工作進程。由于監督者的存在，Riak的代碼庫可以關注于正確的情況，監督者可以在整個系統中以一致的行為處理軟件的bug、損壞的數據和系統錯誤。在Erlang的世界中，這種非防御式編程的方法通常稱為“讓它崩潰”策略。構成監督樹的子進程既可以包含監督者也可以包含工作進程。工作進程指的是包含`gen_fsm`、`gen_server`或`gen_event`的OTP行為。由于Riak團隊不需要處理邊界錯誤條件，所以可以在較小的代碼庫中開發。由于使用了行為，所以這個代碼庫從一開始就比較小，因為代碼庫中只需要包含實現具體功能的代碼。和大部分Erlang應用程序一樣，Riak有一個頂層的監督者，還有監視一組組負責類似功能的進程的子監督者。具體的例子包括Riak的虛擬節點、TCP套接字監聽者和查詢響應管理器。 ### 15.5.1\. 監督者回調函數 為了演示如何實現監督者行為，我們拿`riak_core_sup.erl`作為例子。Riak核心監督者是Riak核心應用程序的頂層監督者。這個監督者啟動一組靜態工作進程和監督者，同時啟動一些動態工作進程處理節點的RESTful API的HTTP和HTTPS綁定，這些API在應用程序相關的配置文件中定義。和`gen_server`采用的方式類似，所有的監督者回調函數模塊都必須包含`-behavior(supervisor).`指令。監督者通過`start`或`start_link`函數啟動，這兩個函數接受一個可選的`ServerName`參數、一個`CallBackModule`參數和一個`Argument`參數，`Argument`參數會被傳入`init/1`回調函數。 下面看一下`riak_core_sup.erl`模塊的前幾行代碼，包括`-behavior`指令和一個之后要描述的宏定義，注意這里調用的`start_link/3`函數： ~~~ -module(riak_core_sup). -behavior(supervisor). %% API -export([start_link/0]). %% Supervisor callbacks -export([init/1]). -define(CHILD(I, Type), {I, {I, start_link, []}, permanent, 5000, Type, [I]}). start_link() -> supervisor:start_link({local, ?MODULE}, ?MODULE, []). ~~~ 啟動一個監督者會創建一個新的進程，并且調用回調模塊`riak_core_sup.erl`中定義的`init/1`回調函數。`ServerName`是一個格式為`{local, Name}`或`{global, Name}`的元組，其中`Name`是監督者的注冊名稱。在我們的例子中，注冊名稱和回調函數名都為原子`riak_core_sup`，這個原子通過`?MODULE`宏得到。將空列表作為參數傳入`init/1`，表示空參數。`init`函數是唯一的監督者回調函數。這個函數返回一個如下格式的元組： ~~~ {ok, {SupervisorSpecification, ChildSpecificationList}} ~~~ 其中`SupervisorSpecification`是一個三元組`{RestartStrategy, AllowedRestarts, MaxSeconds}`，包含了如何處理進程崩潰和重啟的信息。`RestartStrategy`是三個配置參數中的一個，表示一個行為異常終止的時候這個行為的兄弟受什么影響： * `one_for_one`：監督樹中的其他進程不受影響。 * `rest_for_one`：在終止的進程之后啟動的進程被終止并被重啟。 * `one_for_all`：所有的進程都被終止并被重啟。 `AllowedRestarts`表示監督者的任何一個子進程在監督者終止自己（及其子進程）之前的`MaxSeconds`秒鐘的時間內最多可以終止的次數。當一個進程終止的時候，這個進程向其監督者發送一個`EXIT`信號，監督者根據重啟策略相應地處理這個終止的事件。監督者達到最大重啟次數的限制之后終止可以保證循環重啟以及其他在這個層次不能解決的問題能夠提升嚴重等級。比如說，一個進程中的問題可能定位在一個不同的子監督樹中，這種情況下，接收到升級的問題的監督者可以終止受到影響的子樹并重啟子樹。 仔細看`riak_core_sup.erl`模塊中`init/1`回調函數的最后一行，注意到這個特別的監督者采用了one-for-one策略，也就意味著進程之間是相互獨立的。監督者允許重啟自己之前進行最多10次重啟。 `ChildSpecificationList`指定了這個監督者要啟動和監視的子進程列表，其中帶有如何終止和重啟這些進程的信息。這個列表包含以下格式的元組： ~~~ {Id, {Module, Function, Arguments}, Restart, Shutdown, Type, ModuleList} ~~~ `Id`是用于區分監督者的唯一標識符。`Module`、`Function`和`Arguments`是一個導出的函數，這個函數會調用行為的start_link函數，并且返回格式為`{ok, Pid}`的元組。`Restart`策略表示根據進程終止的類型進行什么操作，這個變量可以取值： * `transient`進程，永遠不重啟； * `temporary`進程，只有異常退出的時候才重啟； * `permanent`進程，總是要重啟，不論終止原因是正常還是異常。 `Shutdown`是一個毫秒值，表示因為重啟或關閉的時候，行為執行`terminate`函數的時間限制。還可以使用原子`infinity`，但是對于除了監督者之外的行為，強烈*不*建議使用`infinity`。`Type`可以取值原子`worker`，表示通用服務器、事件處理程序或有限狀態機，還可以取值原子`supervisor`。還有`ModuleList`，這是實現這個行為的模塊列表，用于運行時軟件升級過程中對進程的控制和暫停。子進程描述列表中只能出現已經存在的行為或用戶實現的行為，因此也只有這些行為才能出現在監督樹中。 有了這些知識之后，我們現在應該可以基于一個公共架構編寫一個定義了進程間依賴、容錯閾值和升級擴散過程的重啟策略。現在還應該能理解`riak_core_sup.erl`模塊中`init/1`中進行的操作。首先學習一下`?CHILD`宏，這個宏創建一個子進程的描述，使用回調模塊的名字作為`Id`，將子進程設置為`permanent`，并且將關閉時間設置為5秒鐘。子進程的類型有`worker`也有`supervisor`。下面看一下這個例子，看看你是否能理解： ~~~ -define(CHILD(I, Type), {I, {I, start_link, []}, permanent, 5000, Type, [I]}). init([]) -> RiakWebs = case lists:flatten(riak_core_web:bindings(http), riak_core_web:bindings(https)) of [] -> %% check for old settings, in case app.config %% was not updated riak_core_web:old_binding(); Binding -> Binding end, Children = [?CHILD(riak_core_vnode_sup, supervisor), ?CHILD(riak_core_handoff_manager, worker), ?CHILD(riak_core_handoff_listener, worker), ?CHILD(riak_core_ring_events, worker), ?CHILD(riak_core_ring_manager, worker), ?CHILD(riak_core_node_watcher_events, worker), ?CHILD(riak_core_node_watcher, worker), ?CHILD(riak_core_gossip, worker) | RiakWebs ], {ok, {{one_for_one, 10, 10}, Children}}. ~~~ 這個監督者啟動的`Children`中大部分都靜態定義為`worker`（只有`riak_core_vnode_sup`定義為`superviso`r）。`RiakWebs`部分是一個例外，這些子進程的定義是通過Riak的配置文件中的HTTP部分動態定義的。 除了庫應用程序之外的每一個OTP應用程序（包括Riak中的那些應用程序）都有自己的監督樹。在Riak中，有多個頂層應用程序運行在Erlang節點中，例如`riak_core`負責運行分布式系統的算法，`riak_kv`負責運行鍵值存儲的語義，`webmachine`負責運行HTTP服務。前面的插圖展示了`riak_core`展開的樹形圖，從中可以看出多層次監督的工作方式。這種結構有很多好處，其中之一就是如果某個子系統崩潰了（因為bug、環境問題或故意的操作等），那么只有這個子系統是第一個被終止的實例。 監督者會重啟需要重啟的進程，整個系統不會受到影響。在實踐中，我們發現這種方式在Riak中運行得很好。也許某個用戶會發現怎樣讓一個虛擬節點崩潰，但是這個虛擬節點很快會被`riak_core_vnode_sup`重啟。如果用戶知道了如何讓`riak_core_vnode_sup`崩潰，那么`riak_core`會負責重啟，將這個終止擴散到頂層監督者。這種故障隔離和恢復機制允許Riak開發者（和Erlang開發者）能夠輕松地構建像小強一樣的系統。 曾經有一個業界的大用戶為了找出一些數據庫系統中哪些部分會崩潰，所以他們創建了一個非常嚴酷的測試環境，此時，這個監督模型的價值就展現出來了。這個測試環境會隨機爆發巨大的流量和故障條件。他們很詫異地發現即使是在這種最嚴酷的環境下Riak也不會停止工作。當然，在幕后，他們也有多種辦法讓單個的進程或子系統崩潰——但是每一次監控者都會做好清理工作，并且重啟崩潰的部分，使得整個系統重新恢復工作。 ### 15.5.2\. 應用程序 之前介紹的應用程序行為的作用是將Erlang模塊和資源包裝在可重用的組件中。在OTP中有兩類應用程序。最常見的形式稱為普通應用程序，這種應用程序會啟動一個監督樹和所有相關的靜態工作進程。而另一種應用程序稱為庫應用程序，例如標準庫就是這一類應用程序，這種應用程序屬于Erlang發行版的一部分，包含庫模塊，但是不會啟動監督樹。這并不是說代碼中不包含進程和監督樹。這只是說它們能夠作為屬于另一個應用程序的監督樹的一部分而啟動。 一個Erlang系統會包含一組松耦合的應用程序。有一些是由開發者編寫的，一些是開源的應用程序，剩下的其他應用程序則屬于Erlang/OTP發行版的一部分。Erlang運行時系統及其工具平等對待所有的應用程序，不論這些應用程序屬于或不屬于Erlang發行版的一部分。 ## 15.6\. Riak中的復制和通信 Riak從Amazon的Dynamo存儲系統[[DHJ+07](http://www.aosabook.org/en/bib1.html#bib%3aamazon%3adynamo)]獲得靈感，是為超大規模情況下極高可靠性和可用性而設計的。Dynamo和Riak的架構結合了分布式散列表（Distributed Hash Tables，DHT）和傳統數據庫的特點。副本放置（replica placement）的一致性散列（consistent hashing）和流言傳播協議（gossip protocol）是Riak和Dynamo都使用了的兩大關鍵技術。 一致性散列要求系統中所有的節點都相互知道，而且知道每一個節點都擁有哪些分區。這種分配數據可以放置在一個集中管理的配置文件中，但是在一個大規模的配置中，這種要求可能極難滿足。另一種替換方法是使用一個中央配置服務器，但是這種方式在系統中會引入單點故障的問題。而Riak采用的方法是通過流言傳播協議將集群歸屬信息和分區所有權數據擴散在整個系統中。 流言傳播協議也稱為傳染病協議（epidemic protocol）。顧名思義，當系統中一個節點想要更新一段共享數據的時候，這個節點對數據的本地副本進行更新，然后將更新后的數據隨機地發送給一個節點。當一個節點收到一個更新后，這個節點將接收到的更新和本地狀態合并，然后再次發送給另一個隨機節點。 當一個Riak集群啟動的時候，所有節點都必須配置相同的分區數。然后一致性散列環被分割為分區數個分區，每一段都在本地保存為一個`{HashRange, Owner}`對。集群中的第一個節點占有所有的分區。當一個新節點加入集群的時候，獲得已有節點的`{HashRange, Owner}`對列表，然后要求占有(分區數)/(節點數)個對，用新的所有權更新其本地狀態。然后將更新后的所有權信息通過流言傳播協議發送給一個節點。接下來這個更新后的信息通過上述算法擴散到整個集群。 通過使用流言傳播協議，Riak可以避免因為中央配置服務器而引入的單點故障的問題，系統操作員也不用維護關鍵的集群配置數據了。任何一個節點都可以根據通過流言傳播協議得到的分配數據對收到的請求進行路由決策。通過結合使用流言傳播協議和一致性散列，Riak可以按照一個真正去中心化系統的方式工作，這對于部署和運營大規模系統是非常重要的。 ## 15.7\. 結論和收獲 大多數程序開發人員都相信更小更簡潔的代碼庫不僅更易于維護，而且bug更少。在集群中通過Erlang提供的基本分布式原語進行通信，使得Riak在開始開發的時候就有一個非常健壯的異步消息層作為基礎，并且可以在這個基礎之上構建自己的協議，而不用擔心底層的實現。隨著Riak慢慢發展為一個成熟的系統，有一些和網絡通信相關的部分脫離了使用Erlang內建的功能（為了直接操縱TCP套接字），而其他部分和Erlang包含的原語結合得很好。由于一開始就用Erlang原生的消息傳遞機制實現一切，所以Riak團隊能夠很快地構建出整個系統。這些原語足夠地干凈和清晰，使得之后可以很容易地將一些被證明不是最適合產品的地方替換掉。 同樣，由于Erlang消息傳遞的本質和Erlang虛擬機輕量級的內核，用戶可以輕松地在1臺機器上運行12個節點，也可以在12臺機器上運行12個節點。相比起更重量級的消息傳遞和集群機制來說，這種方式的開發和測試要簡單得多。由于Riak分布式的基礎本質，這一點尤其重要。過去，大部分分布式系統都很難在一個開發者的筆記本電腦上以“開發模式”操作。因此，開發者們往往需要以完整系統的一個子集作為環境進行代碼測試，這種子集的行為往往會有很大的差異。由于多節點的Riak集群可以很簡單地在一臺筆記本上運行，而不需要很大的資源消耗或復雜的配置，所以這種開發過程很容易得到能夠直接在生產環境部署的代碼。 通過使用Erlang/OTP的監督者行為，Riak在面對子模塊崩潰的時候能表現得更加頑強。Riak還能更進一步，從這種行為得到靈感，Riak集群甚至在整個節點崩潰從系統中消失的時候依然能夠保持功能。這種頑強程度令人驚嘆。例如大型企業對多個數據庫進行壓力測試并且故意讓數據庫崩潰以觀察邊緣條件的時候，他們會感到非常驚訝。當他們測試Riak的時候，他們感覺非常不解。每一次他們發現一種方法（例如通過操作系統級別的操作或損壞的進程間通信等方法）讓Riak的某個子系統崩潰的時候，他們都能觀察到一個短暫的性能下降，然后系統又會恢復正常的行為。這就是深思熟慮的“讓它崩潰”策略的直接結果。Riak有規則地根據需求重新啟動每一個需要重啟的子系統，這樣整個系統就能繼續工作了。這種體驗完美地展示了通過Erlang/OTP的方式構建程序能夠達到的健壯程度。 ### 15.7.1\. 致謝 這一章的內容來源于Francesco Cesarini和Simon Thompson 2009年在布達佩斯和科馬爾諾舉辦的中歐函數式編程學校（Central European Functional Programming School）上的演講稿。主要的貢獻來自于英國坎特伯雷的肯特大學的Simon Thompson。要特別感謝所有的審稿人，謝謝你們在本章編寫的各個階段提供的有價值的反饋意見。