.net平台的rabbitmq使用封装demo详解(3)

/// <summary> /// 获取消息 /// </summary> /// <typeparam></typeparam> /// <param></param> /// <param></param> /// <param></param> /// <param>消费处理</param> private void Poll<T>(string exchange, string queue, string routingKey, Action<T> handler) where T : class { var channel = GetModel(exchange, queue, routingKey); var result = channel.BasicGet(queue, false); if (result.IsNull()) return; var msg = result.Body.DeserializeUtf8().FromJson<T>(); try { handler(msg); } catch (Exception ex) { ex.GetInnestException().WriteToFile("队列接收消息", "RabbitMq"); } finally { channel.BasicAck(result.DeliveryTag, false); } }

　　快的时候有1.8K/s，稳定是1.5K/S

Rpc（远程调用）的封装

　　首先说明下，RabbitMq只是提供了这个RPC的功能，但是并不是真正的RPC，为什么这么说：

　　1、传统Rpc隐藏了调用细节，像调用本地方法一样传参、抛出异常

　　2、RabbitMq的Rpc是基于消息的，消费者消费后，通过新队列返回响应结果。

/// <summary> /// RPC客户端 /// </summary> /// <param></param> /// <param></param> /// <param></param> /// <param></param> /// <param></param> /// <returns></returns> public string RpcClient(string exchange, string queue, string routingKey, string body, bool isProperties = false) { var channel = GetModel(exchange, queue, routingKey, isProperties); var consumer = new QueueingBasicConsumer(channel); channel.BasicConsume(queue, true, consumer); try { var correlationId = Guid.NewGuid().ToString(); var basicProperties = channel.CreateBasicProperties(); basicProperties.ReplyTo = queue; basicProperties.CorrelationId = correlationId; channel.BasicPublish(exchange, routingKey, basicProperties, body.SerializeUtf8()); var sw = Stopwatch.StartNew(); while (true) { var ea = consumer.Queue.Dequeue(); if (ea.BasicProperties.CorrelationId == correlationId) { return ea.Body.DeserializeUtf8(); } if (sw.ElapsedMilliseconds > 30000) throw new Exception("等待响应超时"); } } catch (Exception ex) { throw ex.GetInnestException(); } } /// <summary> /// RPC服务端 /// </summary> /// <typeparam></typeparam> /// <param></param> /// <param></param> /// <param></param> /// <param></param> /// <param></param> public void RpcService<T>(string exchange, string queue, bool isProperties, Func<T, T> handler, bool isDeadLetter) { //队列声明 var channel = GetModel(queue, isProperties); var consumer = new EventingBasicConsumer(channel); consumer.Received += (model, ea) => { var body = ea.Body; var msgStr = body.DeserializeUtf8(); var msg = msgStr.FromJson<T>(); var props = ea.BasicProperties; var replyProps = channel.CreateBasicProperties(); replyProps.CorrelationId = props.CorrelationId; try { msg = handler(msg); } catch (Exception ex) { ex.GetInnestException().WriteToFile("队列接收消息", "RabbitMq"); } finally { channel.BasicPublish(exchange, props.ReplyTo, replyProps, msg.ToJson().SerializeUtf8()); channel.BasicAck(ea.DeliveryTag, false); } }; channel.BasicConsume(queue, false, consumer); }

　　可以用，但不建议去用。可以考虑其他的RPC框架。grpc、thrift等。

结尾

　　本篇文章，没有过多的写RabbitMq的知识点，因为园子的学习笔记实在太多了。下面把我的代码奉上 https://github.com/SkyChenSky/Sikiro.Mq.Rabbit。如果有发现写得不对的地方麻烦在评论指出，我会及时修改以免误导别人。

到此这篇关于.net平台的rabbitmq使用封装的文章就介绍到这了,更多相关.net使用rabbitmq内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章: