RabbitMQ 实现RPC

实现RPC

首先要弄明白，RPC是个什么东西。

(RPC) Remote Procedure Call Protocol 远程过程调用协议

在一个大型的公司，系统由大大小小的服务构成，不同的团队维护不同的代码，部署在不同的机器。但是在做开发时候往往要用到其它团队的方法，因为已经有了实现。但是这些服务部署不同的机器上，想要调用就需要网络通信，这些代码繁琐且复杂，一不小心就会写的很低效。RPC协议定义了规划，其它的公司都给出了不同的实现。比如微软的wcf，以及现在火热的WebApi。

在RabbitMQ中RPC的实现也是很简单高效的，现在我们的客户端、服务端都是消息发布者与消息接收者。

首先客户端通过RPC向服务端发出请求

服务端拿到了请求，开始处理并返回





一些繁琐的细节rabbitmq已经为我们封装了，简单的SimpleRpcServer与SimpleRpcClient让Rpc实现的更为方便，这里可以先看一下server端

使用默认的交换机，创建一个SimpleRpcServer的实例，这里需要注意的是，SimpleRpcServer的处理应该是根据业务来的，也就是自己的。在给出的类中没有任何的实现，如果我们创建一个自己的RpcServer并且给出实现

//创建返回一个新的频道
using (var channel = RabbitMqHelper.GetConnection().CreateModel())
{

//创建一个rpc queue
channel.QueueDeclare("RpcQueue", true, false, false, null);

SimpleRpcServer rpc = new SmsSimpleRpcServer(new Subscription(channel, "RpcQueue"));

Console.WriteLine("服务端启动成功");

rpc.MainLoop();

Console.ReadKey();

}

这里是自己的一个RpcServer，在HandleSimpleCall方法里返回对处理的回调消息，在ProcessRequest中做出具体的处理逻辑

/// <summary>
/// 发送短信的Rpc
/// </summary>
public class SmsSimpleRpcServer : SimpleRpcServer
{
public SmsSimpleRpcServer(Subscription subscription) : base(subscription)
{
}

/// <summary>
/// 执行完成后进行h回调
/// </summary>
/// <param name="isRedelivered"></param>
/// <param name="requestProperties"></param>
/// <param name="body"></param>
/// <param name="replyProperties"></param>
/// <returns></returns>
public override byte[] HandleSimpleCall(bool isRedelivered, IBasicProperties requestProperties, byte[] body, out IBasicProperties replyProperties)
{
replyProperties = null;
return Encoding.UTF8.GetBytes($"给{Encoding.UTF8.GetString(body)}发送短信成功");
}

/// <summary>
/// 进行处理
/// </summary>
/// <param name="evt"></param>
public override void ProcessRequest(BasicDeliverEventArgs evt)
{
// todo.....
base.ProcessRequest(evt);
}
}

回到client端，这里的代码也是非常容易的。创建一个SimpleRpcClient，然后指定了交换机类型，因为用的是默认的，所以exchange传的是null， routingkey是我们的rpcqueue。最后调用call方法

using (var channel = RabbitMqHelper.GetConnection().CreateModel())
{
//创建client的rpc
SimpleRpcClient client = new SimpleRpcClient(channel, new PublicationAddress(exchangeType: ExchangeType.Direct, exchangeName: string.Empty, routingKey: "RpcQueue"));
bool flag = true;
var sendmsg = "";
while (flag)
{
Console.WriteLine("请输入要发送的消息");
sendmsg = Console.ReadLine();
if (string.IsNullOrWhiteSpace(sendmsg))
{
Console.Write("请输入消息");
continue;
}

var msg = client.Call(Encoding.UTF8.GetBytes(sendmsg));

Console.WriteLine(Encoding.UTF8.GetString(msg));

}

Console.ReadKey();

}

把程序运行起来





后面说一些内部的东西，其实上在创建一次SimpleRpcClient的时候都会创建一个回调队列，这个队列在程序关闭后会自动消失，所以这些建议创建一次就够了，都使用这个。如果创建多次会影响性能









在回调的时候，通过源码也可以看到判断了correlation_id的一致性





在server端也可以看到在执行Process后会发布消息到回调队列