15.3. OTP 行为
致力于 Riak 的核心开发团队遍布近十个地理位置。如果没有非常紧密的协调和模板可供参考，结果将是不同的客户端/服务器实现，它们无法处理特殊的边界情况和并发相关的错误。可能没有统一的方式来处理客户端和服务器崩溃，或者保证来自请求的响应确实是响应，而不仅仅是符合内部消息协议的任何消息。
OTP 是一组 Erlang 库和设计原则，提供现成的工具来开发健壮的系统。许多这些模式和库以“行为”的形式提供。
OTP 行为通过提供实现最常见并发设计模式的库模块来解决这些问题。在幕后，无需程序员知晓，库模块确保以一致的方式处理错误和特殊情况。因此，OTP 行为提供了一组标准化的构建块，用于设计和构建工业级系统。
15.3.1. 简介
OTP 行为作为库模块提供，位于 stdlib 应用程序中，该应用程序是 Erlang/OTP 分发的一部分。程序员编写的特定代码放置在单独的模块中，并通过一组预定义的回调函数调用，这些回调函数针对每个行为进行标准化。此回调模块将包含提供所需功能所需的所有特定代码。
OTP 行为包括执行实际处理的 worker 进程和负责监控 worker 和其他 supervisor 的 supervisor。worker 行为（通常在图中表示为圆圈）包括服务器、事件处理程序和有限状态机。supervisor（在插图中表示为方框）监控其子进程（worker 和其他 supervisor），从而创建所谓的监督树。
![[OTP Riak Supervision Tree]](../../static/riak/supervision-tree.png)
图 15.1：OTP Riak 监督树
监督树打包成一个称为应用程序的行为。OTP 应用程序不仅是 Erlang 系统的构建块，也是打包可重用组件的一种方式。工业级系统（如 Riak）由一组松散耦合的、可能是分布式的应用程序组成。其中一些应用程序是标准 Erlang 分发的一部分，而另一些则是构成 Riak 特定功能的组成部分。
OTP 应用程序的示例包括 Corba ORB 或简单网络管理协议 (SNMP) 代理。OTP 应用程序是一个可重用组件，它将库模块与 supervisor 和 worker 进程打包在一起。从现在开始，当我们提到应用程序时，我们指的是 OTP 应用程序。
行为模块包含每个给定行为类型的通用代码。虽然可以实现自己的行为模块，但这样做很少见，因为 Erlang/OTP 分发附带的模块将满足您代码中使用的大多数设计模式。行为模块中提供的通用功能包括以下操作：
- 生成并可能注册进程；
- 发送和接收客户端消息作为同步或异步调用，包括定义内部消息协议；
- 存储循环数据并管理进程循环；以及
- 停止进程。
循环数据是一个变量，它将包含行为在调用之间需要存储的数据。调用后，将返回循环数据的更新变体。此更新的循环数据（通常称为新的循环数据）作为参数传递给下一次调用。循环数据也常被称为行为状态。
要包含在回调模块中以供通用服务器应用程序提供所需特定行为的功能包括以下内容：
- 初始化进程循环数据，以及如果进程已注册，则注册进程名称。
- 处理特定的客户端请求，以及如果是同步的，则处理发送回客户端的回复。
- 处理和更新进程循环数据，以在进程请求之间进行。
- 在终止时清理进程循环数据。
15.3.2. 通用服务器
实现客户端/服务器行为的通用服务器定义在 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 callbacks
-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). 指令识别通用服务器。此指令用于编译器，以确保所有回调函数都被正确导出。记录状态用于服务器循环数据中。
15.3.3. 启动您的服务器
使用 gen_server 行为，您将使用 gen_server:start 和 gen_server:start_link 函数，而不是使用 spawn 和 spawn_link BIF。spawn 和 start 之间的主要区别在于调用的同步性质。使用 start 而不是 spawn 使启动 worker 进程更加确定性，并防止不可预见的数据竞争，因为在 worker 初始化之前，调用将不会返回 worker 的 PID。您可以使用以下任一方式调用这些函数：
gen_server:start_link(ServerName, CallbackModule, Arguments, Options) gen_server:start_link(CallbackModule, Arguments, Options)
ServerName 是一个格式为 {local, Name} 或 {global, Name} 的元组，表示进程别名的本地或全局 Name（如果要注册它）。全局名称允许服务器在分布式 Erlang 节点的集群中透明地访问。如果您不想注册进程，而是使用其 PID 引用它，则省略该参数并使用 start_link/3 或 start/3 函数调用。CallbackModule 是放置特定回调函数的模块的名称，Arguments 是一个有效的 Erlang 术语，传递给 init/1 回调函数，而 Options 是一个列表，允许您设置内存管理标志 fullsweep_after 和 heapsize，以及其他跟踪和调试标志。
在我们的示例中，我们调用 start_link/4，使用 ?MODULE 宏调用将进程注册为与回调模块相同的名称。此宏在编译代码时由预处理器扩展为定义它的模块的名称。始终建议将您的行为命名为与其实现的回调模块相同的别名。我们不传递任何参数，因此只发送空列表。选项列表保持为空
start_link() ->
gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
start_link 和 start 函数之间的明显区别在于 start_link 与其父进程（通常是 supervisor）链接，而 start 不链接。这需要特别说明，因为它是 OTP 行为负责与 supervisor 链接。当从 shell 测试行为时，start 函数通常使用，因为导致 shell 进程崩溃的类型错误不会影响行为。start 和 start_link 函数的所有变体都返回 {ok, Pid}。
start 和 start_link 函数将生成一个新进程，该进程将调用 CallbackModule 中的 init(Arguments) 回调函数，并提供 Arguments。init 函数必须初始化服务器的 LoopData，并且必须返回一个格式为 {ok, LoopData} 的元组。LoopData 包含将传递给回调函数的循环数据的第一个实例。如果您想存储传递给 init 函数的一些参数，您将在 LoopData 变量中执行此操作。Riak 节点监视器服务器中的 LoopData 是使用类型为 state 的记录调用 schedule_broadcast/1 的结果，其中字段设置为默认值
init([]) ->
%% Watch for node up/down events
net_kernel:monitor_nodes(true),
%% Setup ETS table to track node status
ets:new(?MODULE, [protected, named_table]),
{ok, schedule_broadcast(#state{})}.
虽然 supervisor 进程可能调用 start_link/4 函数，但另一个进程调用 init/1 回调函数：刚刚生成的进程。由于此服务器的目的是通知、记录和广播 Riak 中子服务的可用性，因此初始化请求 Erlang 运行时通知它此类事件，并设置一个表来存储此信息。这需要在初始化期间完成，因为如果该结构尚不存在，则对服务器的所有调用都将失败。只执行必要的操作，并最大程度地减少 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) ->
%% Update the set of active services locally
Services = ordsets:add_element(Id, State#state.services),
S2 = State#state { services = Services },
%% Remove any existing mrefs for this service
delete_service_mref(Id),
%% Setup a monitor for the Pid representing this service
Mref = erlang:monitor(process, Pid),
erlang:put(Mref, Id),
erlang:put(Id, Mref),
%% Update our local ETS table and broadcast
S3 = local_update(S2),
{reply, ok, update_avsn(S3)};
handle_call({service_down, Id}, _From, State) ->
%% Update the set of active services locally
Services = ordsets:del_element(Id, State#state.services),
S2 = State#state { services = Services },
%% Remove any existing mrefs for this service
delete_service_mref(Id),
%% Update local ETS table and broadcast
S3 = local_update(S2),
{reply, ok, update_avsn(S3)};
请注意回调函数的返回值。元组包含控制原子 reply，告诉 gen_server 通用代码元组的第二个元素（在这两种情况下都是原子 ok）是发送回客户端的回复。元组的第三个元素是新的 State，它在服务器的新迭代中作为第三个参数传递给 handle_call/3 函数；在这两种情况下，它都更新为反映新的可用服务集。参数 _From 是一个元组，包含唯一的邮件引用和客户端进程标识符。该元组作为一个整体在库函数中使用，我们将在本章中不讨论这些函数。在大多数情况下，您不需要它。
gen_server 库模块内置了许多机制和保护措施，它们在幕后运行。如果您的客户端向您的服务器发送同步消息，并且您在五秒钟内没有收到响应，则执行 call/2 函数的进程将被终止。您可以使用 gen_server:call(Name, Message, Timeout) 覆盖此操作，其中 Timeout 是以毫秒为单位的值或原子 infinity。
超时机制最初是为了防止死锁而设置的，确保意外互相调用的服务器在默认超时后被终止。崩溃报告将被记录，并有望导致错误被调试和修复。大多数应用程序在五秒钟的超时下都能正常运行，但在负载非常重的情况下，您可能需要微调该值，甚至可能使用 infinity；此选择取决于应用程序。Erlang/OTP 中的所有关键代码都使用 infinity。Riak 中的各个地方对超时使用不同的值：infinity 在内部耦合的部分之间很常见，而 Timeout 是根据用户传递的参数设置的，在客户端代码与 Riak 交谈的情况下，该代码指定了操作应该被允许超时。
使用gen_server:call/2函数时，还有一些其他安全措施需要考虑，例如向不存在的服务器发送消息以及服务器在发送回复之前崩溃的情况。在这两种情况下，调用进程都将终止。在原始 Erlang 中，发送在接收子句中从未匹配模式的消息是一个错误，会导致内存泄漏。Riak 使用两种不同的策略来缓解这种情况，这两种策略都涉及“万能”匹配子句。在消息可能由用户发起的那些地方，未匹配的消息可能会被静默丢弃。在那些消息只能来自 Riak 内部的地方，它代表一个错误，因此将用于触发错误警报的内部崩溃报告，重新启动接收它的工作进程。
发送异步消息的工作方式类似。消息被异步发送到通用服务器，并在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}，分别。一些东西必须触发这个返回值，通常是发送到服务器的停止消息。收到包含Reason和State的停止元组后，通用代码将执行terminate(Reason, State)回调。
terminate函数是插入清理服务器State和系统使用的任何其他持久数据所需的代码的自然位置。在我们的示例中，我们会向我们的对等节点发送最后一条消息，以便他们知道这个节点监视器不再处于运行状态并进行监视。在这个示例中，变量State包含一个带有字段status和peers的记录。
terminate(_Reason, State) ->
%% Let our peers know that we are shutting down
broadcast(State#state.peers, State#state { status = down }).
将行为回调用作库函数并从程序的其他部分调用它们是一种非常糟糕的做法。例如，您永远不应该从另一个模块调用riak_core_node_watcher:init(Args)来检索初始循环数据。此类检索应通过对服务器的同步调用来完成。对行为回调函数的调用应仅来自行为库模块，作为系统中发生的事件的结果，而不是由用户直接调用。