第十六章 OTP概述
Table of Contents
第十六章 OTP概述
16.1 通用服务器程序的进化路线
16.1.1 server1: 原始服务器程序
服务端实现
-module(server1). -export([start/2, rpc/2]). %% 启动服务 start(Name, Mod) -> %% 注册进程名为Name, 并在启动进程时完成模块Mod的初始化并在调用loop进行监测 register(Name, spawn(fun() ->loop(Name, Mod, Mod:init()) end)). %% 远程调用 rpc(Name, Request) -> %% 向指定的进程发送请求, 使用self()获得自身的Pid, 用于识别请求者 Name ! {self(), Request}, receive %% 接收到处理结果后输出 {Name, Response} ->Response end. loop(Name, Mod, State) -> receive %% 接收到请求后的处理 {From, Request} -> %% 调用Mod的handle函数进行处理 {Response, State1} = Mod:handle(Request, State), %% 将处理结果发送给请求者 From ! {Name, Response}, loop(Name, Mod, State1) end.
回调程序实现
-module(name_server). -import(server1, [rpc/2]). -export([init/0, add/2, whereis/1, handle/2]). %% 对外提供的功能函数, 会通过远程调用向指定的服务名发送请求 add(Name, Place) ->rpc(name_server, {add, Name, Place}). whereis(Name) ->rpc(name_server, {whereis, Name}). %% 模块初始化, 创建新的字典 init() ->dict:new(). %% 处理函数, 添加和查询 handle({add, Name, Place}, Dict) ->{ok, dict:store(Name, Place, Dict)}; handle({whereis, Name}, Dict) ->{dict:find(Name, Dict), Dict}.
运行结果
1> c(server1).
{ok,server1}
2> c(name_server).
{ok,name_server}
3> server1:start(name_server, name_server).
true
4> name_server:add(joe, "at home").
ok
5> name_server:whereis(joe).
{ok,"at home"}
处理流程:
1. server1:start(name_server, name_server).
启动进程, 完成name_server模块的初始化, 指定使用name_server的handle函数处理请求, 最后将进程注册为name_server
2. name_server:add(joe, "at home").
功能函数相当于是一个接口, 将数据封装为{add, joe, "at home"}格式, 调用rpc提交请求
3. rpc(Name, Request)
Name ! {self(), Request}
获取自身进程ID后将数据发送给name_server进程
4. loop(Name, Mod, State)
{Response, State1} = Mod:handle(Request, State)
接收到请求后调用name_server:handle处理请求
5. handle({add, Name, Place}, Dict)
回调函数, 相当于接口的具体实现, 向字典中添加数据
6. loop(Name, Mod, State)
From ! {Name, Response}
将处理结果发送给请求者
7. rpc(Name, Request)
{Name, Response} -> Response
打印处理结果
16.1.2 server2: 支持事务的服务器程序
服务端实现
-module(server2). -export([start/2, rpc/2]). %% 启动服务 start(Name, Mod) -> %% 注册进程名为Name, 并在启动进程时完成模块Mod的初始化并在调用loop进行监测 register(Name, spawn(fun() ->loop(Name, Mod, Mod:init()) end)). %% 远程调用 rpc(Name, Request) -> %% 向指定的进程发送请求, 使用self()获得自身的Pid, 用于识别请求者 Name ! {self(), Request}, receive %% 添加了出错处理 {Name, crash} ->exit(rpc); %% 接收到处理结果后输出 {Name, ok, Response} ->Response end. loop(Name, Mod, OldState) -> receive %% 接收到请求后的处理 {From, Request} -> %% 对Mod的handle函数调用添加异常处理 try Mod:handle(Request, OldState) of {Response, NewState} -> From ! {Name, ok, Response}, loop(Name, Mod, NewState) catch _:Why -> %% 发生异常则打印异常信息 log_the_error(Name, Request, Why), From ! {Name, crash}, %% 保留旧的状态 loop(Name, Mod, OldState) end end. log_the_error(Name, Request, Why) -> io:format("Server ~p request ~p ~n caused exception ~p~n", [Name, Request, Why]).
16.1.3 server3: 支持热代码替换的服务器程序
在服务端添加了替换代码的函数
swap_code(Name, Mod) ->rpc(Name, {swap_code, Mod}). loop(Name, Mod, OldState) -> receive {From, {swap_code, NewCallBackMod}} -> From ! {Name, ack}, %% 改变loop循环用于处理请求的模块 loop(Name, NewCallBackMod, OldState); {From, Request} -> {Response, NewState} = Mod:handle(Request, OldState), From ! {Name, Response}, loop(Name, Mod, NewState) end.
运行结果:
1> server3:start(name_server, name_server1).
true
2> name_server1:add(joe, "at home").
ok
3> name_server1:add(helen, "at work").
ok
4> c(new_name_server).
{ok,new_name_server}
5> server3:swap_code(name_server, new_name_server).
ack
6> new_name_server:all_names().
[joe,helen]
7> new_name_server:delete(joe).
ok
8> new_name_server:all_names().
[helen]
9> new_name_server:whereis(helen).
{ok,"at work"}
16.1.4 server4: 同时支持事务和热代码替换
显然, 将server2中的异常捕获处理添加到server3中即可。
16.1.5 server5: 压轴好戏
空服务器的实现
-module(server5). -export([start/0, rpc/2]). %% 启动进程 start() ->spawn(fun() ->wait() end). %% 等待接收指令成为某种服务 wait() -> receive {become, F} ->F() end. %% 远程调用 rpc(Pid, Q) -> %% 由服务进程处理请求 Pid ! {self(), Q}, receive %% 打印处理结果 {Pid, Reply} ->Reply end.
具体服务的一个实现
-module(my_fac_server). -export([loop/0]). %% 阶乘服务 loop() -> receive %% 可以接收数字计算其阶乘 {From, {fac, N}} -> From ! {self(), fac(N)}, loop(); %% 也可以变成另外一种服务 {become, Something} ->Something() end. fac(0) ->1; fac(N) ->N * fac(N-1).
运行结果:
1> c(server5).
{ok,server5}
2> Pid = server5:start().
<0.52.0>
3> c(my_fac_server).
{ok,my_fac_server}
4> Pid ! {become, fun my_fac_server:loop/0}.
{become,#Fun<my_fac_server.loop.0>}
5> server5:rpc(Pid, {fac, 30}).
265252859812191058636308480000000
运行流程:
1. 启动进程, 等待接收指令成为某种服务
Pid = server5:start().
start() -> spawn(fun() -> wait() end).
2. 接收指令, 成为可以计算阶乘的服务
Pid ! {become, fun my_fac_server:loop/0}.
3. 服务调用
server5:rpc(Pid, {fac, 30}).
4. rpc(Pid, Q)
%% 由服务进程处理请求
Pid ! {self(), Q}
5. 处理请求
loop() ->
receive
{From, {fac, N}} ->
From ! {self(), fac(N)}
...
6. 接收结果并输出
rpc(Pid, Q)
receive
{Pid, Reply} -> Reply
end.
16.2 gen_server起步
16.2.1 第一步: 确定回调模块的名称
模拟支付系统, 模块命名为my_bank
16.2.2 第二步: 写接口函数
%% 启动本地服务器 %% gen_server:start_link({local, Name}, Mod, _) start() ->gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). %% 远程调用 %% gen_server:call(Name, Term) stop() ->gen_server:call(?MODULE, stop). %% 开一个新账户 new_account(Who) ->gen_server:call(?MODULE, {new, Who}). %% 存钱 deposit(Who, Amount) ->gen_server:call(?MODULE, {add, Who, Amount}). %% 取钱 withdraw(Who, Amount) ->gen_server:call(?MODULE, {remove, Who, AMount}).
16.2.3 第三步: 编写回调函数
%% init([]) -> {ok, State} %% 必须实现的回调函数, 用于模块初始化 %% 这里返回一个ETS表 init([]) ->{ok, ets:new(?MODULE, [])}. %% handle_call(_Request, _From, State) -> {reply, Reply, State} %% 必须实现的回调函数, 用于gen_server:call时回调使用 %% 添加用户 handle_call({new, Who}, _From, Tab) -> %% 查询ETS表中相关用户是否存在 %% 不存在则插入ETS表并提示欢迎信息 %% 存在则提示已经存在此用户 Reply = case ets:lookup(Tab, Who) of [] ->ets:insert(Tab, {Who, 0}), {welcome, Who}; [_] ->{Who, you_already_are_a_customer} end, {reply, Reply, Tab}; %% 用户存钱 handle_call({add, Who, X}, _From, Tab) -> %% 首先查询用户是否存在 %% 存在则将存款累加后重新插入ETS表并给出提示信息 Reply = case ets:lookup(Tab, Who) of [] ->not_a_customer; [{Who, Balance}] -> NewBalance = Balance + X, ets:insert(Tab, {Who, NewBalance}), {thanks, Who, your_balance_is, NewBalance} end, {reply, Reply, Tab}; %% 用户取钱 handle_call({remove, Who, X}, _From, Tab) -> %% 首先查询用户是否存在 %% 存在则根据取款与存款的大小关系分别处理 Reply = case ets:lookup(Tab, Who) of [] ->not_a_customer; [{Who, Balance}] when X =< Balance -> NewBalance = Balance - X, ets:insert(Tab, {Who, NewBalance}), {thanks, Who, your_balance_is, NewBalance}; [{Who, Balance}] -> {sorry, Who, you_only_have, Balance, in_the_bank} end, {reply, Reply, Tab}; %% 停止服务 handle_call(stop, _From, Tab) -> {stop, normal, stopped, Tab}. %% 其它必须实现的回调函数 handle_cast(_Msg, State) ->{noreply, State}. handle_info(_Info, State) ->{noreply, State}. terminate(_Reason, _State) ->ok. code_change(_OldVsn, State, Extra) ->{ok, State}.
运行结果:
1> my_bank:start().
{ok,<0.70.0>}
2> my_bank:deposit("joe", 10).
not_a_customer
3> my_bank:new_account("joe").
{welcome,"joe"}
4> my_bank:deposit("joe", 10).
{thanks,"joe",your_balance_is,10}
5> my_bank:deposit("joe", 30).
{thanks,"joe",your_balance_is,40}
6> my_bank:withdraw("joe", 15).
{thanks,"joe",your_balance_is,25}
7> my_bank:withdraw("joe", 45).
{sorry,"joe",you_only_have,25,in_the_bank}
8> my_bank:stop().
stopped
16.3 gen_server回调的结构
16.3.1 启动服务器程序时发生了什么
%% 通过start_link启动 %% 创建名为Name的通用服务器程序 %% 回调模块为Mod %% Opts控制服务器程序的行为 %% 调用Mod:init(InitArgs)启动服务器程序 gen_server:start_link(Name, Mod, InitArgs, Opts)
16.3.2 调用服务器程序时发生了什么
%% 通过call调用服务端程序 %% 最终会调用回调模块中的handle_call/3函数 gen_server:call(Name, Request) %% Request 请求信息 %% From 发起调用的客户端进程ID %% State 客户端当前状态 Mod:handle_call(Request, From, State)
16.3.3 调用和通知
%% 通过cast实现通知 %% 最终会调用回调模块中的handle_cast/2函数 gen_server:cast(Name, Name) %% Msg 发送的消息 %% State 状态 Mod:handle_cast(Msg, State)
16.3.4 发送给服务器的原生消息
%% 接收其它进程或系统发送的消息 Mod:handle_info(Info, State)
16.3.5 Hasta La Vista, Baby
%% 这个标题, Joe真有点意思:） Mod:terminate(Reason, NewState)
16.3.6 热代码替换
code_change(OldVsn, State, Extra)