谈谈RPC中的异步调用设计

小程序：扫一扫查出行
【扫一扫了解最新限行尾号】
复制小程序

RPC（远过程调用）在分布式系统中是很常用的基础通讯手段，核心思想是将不同进程之间的通讯抽象为函数调用，基本的过程是调用端通过将参数序列化到流中并发送给服务端，服务端从流中反序列化出参数并完成实际的处理，然后将结果序列化后返回给调用端。通常的RPC由接口形式来定义，接口定义服务的名字，接口方法定义每个请求的输入参数和返回结果。RPC内部的序列化、网络通讯等实现细节则由框架来完成，对用户来说是完全透明的。之前我使用.net开发过一套轻量级的分布式框架（PPT看这里，视频看这里），这套框架经过2年多的持续开发和改进已经运用到数款产品中（包括网络游戏和分布式应用），取得了不错的效果，等未来框架成熟后会考虑开源，本文讨论的RPC基于这套框架展开。

通常我们的函数调用都是同步的，也就是调用方在发起请求后就能得到结果（成功返回结果失败则抛出异常），中间不能去干其他事情，与这种模式对应的RPC称之为请求-响应式模式。请求-响应式的优点在于时序清晰，逻辑简单，和普通的函数调用完全等价。比如我们可以这样定义RPC接口：

[Protocol(ID=)]
public interface ICalculate
{
[DispId()]
int Add(int p1, int p2);
}

客户端就可以像这样使用接口：

var calculate = new ICalculateProxy();//ICalculateProxy为框架生成的接口代理类
calculate.Connect(url);
var result = calculate.Add(, );

但是在分布式中这种模式的缺点也是非常的明显，第一个问题是网络通讯的延迟会严重的制约请求-响应式RPC的响应速度，使得请求吞吐量无法满足性能需要，大量的CPU时间会阻塞在等待请求的响应上；第二个问题是请求-响应式只有由客户端向服务端发起请求，服务端不能主动向客户端发送事件通知，也就是缺乏一种callback机制。

针对请求-响应式的缺点我们可以用双向通讯机制来改进，首先去掉请求的返回值，所有的方法请求都定义为无返回结果，这样调用方在发出请求之后就可以继续干后面的事情了，而不需要再等待服务返回结果。同时针对服务接口定义一个Callback接口用于服务端向客户端发送请求结果和事件通知，这样服务器就可以主动向客户端发送消息了。这种RPC模式可以称之为双向会话式，接口可以这样定义：

[Protocol(ID=), Callback(typeof(ICalculateCallback))]
public interface ICalculate
{
[DispId()]
void Add(int p1, int p2);
} public interface ICalculateCallback : IServiceCallback
{
[DispId()]
void OnAdd(int result);
}

服务端可以这样实现服务接口：

public class CaculateService : ICaculateImpl //这里ICaculateImpl为框架生成的服务实现接口
{
ICaculateImpl.Add(Session session, int p1, int p2)
{
var result = p1 + p2;
session.Cllback.OnAdd(result);
}
}

双向会话式解决了请求的异步处理以及服务器的双向通讯问题，但是也给调用者带来了一些不便，例如上例中如果调用方发起多个Add请求，在收到OnAdd消息后如何将结果与请求关联起来呢？一种解决方案是在Add请求中多加一个request id参数，服务器在处理完Add之后将request id放到OnAdd方法中和结果一起传给客户端，客户端根据request id来关联请求与结果。这种手工处理的方式代码写起来很麻烦，那么有没有一种更好的RPC模式来解决这个问题呢？这就是下面给大家介绍的支持异步调用的RPC设计。

异步调用的主要设计思想是在双向会话式的基础上让调用方通过一个回调函数来获得请求的结果，而不再通过Callback接口来获得结果。采用回调函数的好处在于我们可以使用闭包来隐式的关联请求和响应之间的上下文，这样就不需要显式的传递request id来手工关联上下文了。并且服务器仍然可以通过Callback接口向客户端主动发送消息，保留了原来双向通讯的优点。但是需要注意的是由于请求在服务器上可能是异步执行的，所以服务器不保证请求的响应是按顺序返回的，这可能造成一些隐含的乱序问题，需要客户端在调用时特别注意。如果响应需要严格的按照请求顺序返回客户端，那么服务端需要同步处理请求，或者引入队列机制对异步的响应进行排队有序返回响应。

之前的ICalculate就可以这样定义：

[Protocol(ID=), Callback(typeof(ICalculateCallback))]
public interface ICalculate
{
[DispId(), Async]
void Add(int p1, int p2, Action<int> OnResult, Action<int,string> OnError = null);
}

用Async这个标签表示这个请求为异步请求，调用者用OnResult回调函数来接收请求的结果，OnError则为返回错误的回调函数，如果调用者不关心错误返回，那么可以不传递OnError，而在IServiceCallback的OnError方法中接收错误信息。
调用者可以很方便的使用闭包来处理结果，同时隐藏异步的实现细节，像这样：

void TestAdd(ICalculateProxy calculate, int p1, int p2)
{
calculate.Add(p1, p2, result => MessageBox.Show(string.Format("{0} + {1} = {2}", p1, p2, result), (errCode, errMsg) => MessageBox.Show("Add failed:" + errMsg));
}

服务器端的实现是这样的：

public class CaculateService : ICaculateImpl
{
ICaculateImpl.Add(Session session, int p1, int p2, ICaculate_AddAsyncReply reply)
{
try
{
var result = p1 + p2;
reply.OnResult(result);
}
catch(OverflowException e)
{
reply.OnError(-, e.Message);
}
}
}

ICaculate_AddAsyncReply为框架生成的返回异步结果的对象，有一个OnResult和一个OnError方法。有了这个reply对象之后，服务器的请求处理也可以实现异步处理，客户端请求不需要在请求函数里一次完成，而是可以放到其他线程或者异步方法中处理，稍后在通过reply向客户端返回结果。

下面我们来看看框架在背后为我们做的一些实现细节，首先是客户端的Proxy：

//在Proxy中使用一个RequestContext结构保存请求的上下文信息，上下文中记录某个请求的唯一id，在调用时一起发送到服务器：
struct RequestContext
{
public int reqId;
public Delegate OnResult;
public Action<int, string> OnError; public RequestContext(int id, Delegate onResult, Action<int, string> onError)
{
reqId = id;
OnResult = onResult;
OnError = onError;
}
} //服务器返回响应之后proxy就找出reqId对应的请求上下文，然后调用对应的回调函数传递结果
void OnAddReply(BinaryStreamReader __reader)
{
int reqId;
int ret;
__reader.Read(out reqId);
__reader.Read(out ret);
if(ret == )
{
int p0;
__reader.Read(out p0);
RequestContext ctx = PopAsyncRequest(reqId);
var __onResult = ctx.OnResult as Action<int>;
__onResult(p0);
}
else
{
RequestContext ctx = PopAsyncRequest(reqId);
string msg;
__reader.Read(out msg);
if(ctx.OnError != null)
ctx.OnError(ret, msg);
else
_handler.OnError(ret, msg);
}
}

服务端的一些实现细节：

//框架生成请求对应的异步响应类
public class ICaculate_AddAsyncReply : AsyncReply
{
public ICaculate_AddAsyncReply(int reqId, Connection conn)
{
_reqId = reqId;
_connection = conn;
} public void OnError(int error, string msg)
{
var stream = new BinaryStreamWriter();
stream.Write();
stream.Write(_reqId);
stream.Write(error);
stream.Write(msg);
_connection.Write(stream.BuildSendBuffer());
}
public void OnResult(int result)
{
var stream = new BinaryStreamWriter();
stream.Write();
stream.Write(_reqId);
stream.Write();
stream.Write(result);
_connection.Write(stream.BuildSendBuffer());
}
}

框架生成的Stub类将收到的请求数据进行解析然后调用具体服务类来处理请求：

void AddInvoke(ICaculateImpl __service, Session __client, BinaryStreamReader __reader)
{
int p1;
int p2;
int __reqId;
__reader.Read(out __reqId);
__reader.Read(out p1);
__reader.Read(out p2);
var reply = new ICaculate_AddAsyncReply(__reqId, __client.Connection);
try
{
__service.Add(__client, p1, p2, reply);
}
catch(ServiceException e)
{
reply.OnError(e.ErrCode, e.Message);
Log.Info("Service Invoke Failed. clientId:{0} error message:{1}", __client.ID, e.Message);
}
catch(Exception e)
{
reply.OnError((int)ServiceErrorCode.Generic, "generic service error.");
Log.Error("Generic Service Invoke Failed, clientId:{0} error message:{1}\nCall Stack: {2}", __client.ID, e.Message, e.StackTrace);
}
}