1         /// <summary>
 2         /// 交换器声明
 3         /// </summary>
 4         /// <param name="iModel"></param>
 5         /// <param name="exchange">交换器</param>
 6         /// <param name="type">交换器类型：
 7         /// 1、Direct Exchange – 处理路由键。需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全
 8         /// 匹配。这是一个完整的匹配。如果一个队列绑定到该交换机上要求路由键 “dog”，则只有被标记为“dog”的
 9         /// 消息才被转发，不会转发dog.puppy，也不会转发dog.guard，只会转发dog
10         /// 2、Fanout Exchange – 不处理路由键。你只需要简单的将队列绑定到交换机上。一个发送到交换机的消息都
11         /// 会被转发到与该交换机绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。Fanout
12         /// 交换机转发消息是最快的。
13         /// 3、Topic Exchange – 将路由键和某模式进行匹配。此时队列需要绑定要一个模式上。符号“#”匹配一个或多
14         /// 个词，符号“*”匹配不多不少一个词。因此“audit.#”能够匹配到“audit.irs.corporate”，但是“audit.*”
15         /// 只会匹配到“audit.irs”。</param>
16         /// <param name="durable">持久化</param>
17         /// <param name="autoDelete">自动删除</param>
18         /// <param name="arguments">参数</param>
19         private static void ExchangeDeclare(IModel iModel, string exchange, string type = ExchangeType.Direct,
20             bool durable = true,
21             bool autoDelete = false, IDictionary<string, object> arguments = null)
22         {
23             exchange = exchange.IsNullOrWhiteSpace() ? "" : exchange.Trim();
24             iModel.ExchangeDeclare(exchange, type, durable, autoDelete, arguments);
25         }
26
27         /// <summary>
28         /// 队列声明
29         /// </summary>
30         /// <param name="channel"></param>
31         /// <param name="queue">队列</param>
32         /// <param name="durable">持久化</param>
33         /// <param name="exclusive">排他队列，如果一个队列被声明为排他队列，该队列仅对首次声明它的连接可见，
34         /// 并在连接断开时自动删除。这里需要注意三点：其一，排他队列是基于连接可见的，同一连接的不同信道是可
35         /// 以同时访问同一个连接创建的排他队列的。其二，“首次”，如果一个连接已经声明了一个排他队列，其他连
36         /// 接是不允许建立同名的排他队列的，这个与普通队列不同。其三，即使该队列是持久化的，一旦连接关闭或者
37         /// 客户端退出，该排他队列都会被自动删除的。这种队列适用于只限于一个客户端发送读取消息的应用场景。</param>
38         /// <param name="autoDelete">自动删除</param>
39         /// <param name="arguments">参数</param>
40         private static void QueueDeclare(IModel channel, string queue, bool durable = true, bool exclusive = false,
41             bool autoDelete = false, IDictionary<string, object> arguments = null)
42         {
43             queue = queue.IsNullOrWhiteSpace() ? "UndefinedQueueName" : queue.Trim();
44             channel.QueueDeclare(queue, durable, exclusive, autoDelete, arguments);
45         }
46
47         /// <summary>
48         /// 获取Model
49         /// </summary>
50         /// <param name="exchange">交换机名称</param>
51         /// <param name="queue">队列名称</param>
52         /// <param name="routingKey"></param>
53         /// <param name="isProperties">是否持久化</param>
54         /// <returns></returns>
55         private static IModel GetModel(string exchange, string queue, string routingKey, bool isProperties = false)
56         {
57             return ModelDic.GetOrAdd(queue, key =>
58             {
59                 var model = _conn.CreateModel();
60                 ExchangeDeclare(model, exchange, ExchangeType.Fanout, isProperties);
61                 QueueDeclare(model, queue, isProperties);
62                 model.QueueBind(queue, exchange, routingKey);
63                 ModelDic[queue] = model;
64                 return model;
65             });
66         }
67
68         /// <summary>
69         /// 发布消息
70         /// </summary>
71         /// <param name="routingKey">路由键</param>
72         /// <param name="body">队列信息</param>
73         /// <param name="exchange">交换机名称</param>
74         /// <param name="queue">队列名</param>
75         /// <param name="isProperties">是否持久化</param>
76         /// <returns></returns>
77         public void Publish(string exchange, string queue, string routingKey, string body, bool isProperties = false)
78         {
79             var channel = GetModel(exchange, queue, routingKey, isProperties);
80
81             try
82             {
83                 channel.BasicPublish(exchange, routingKey, null, body.SerializeUtf8());
84             }
85             catch (Exception ex)
86             {
87                 throw ex.GetInnestException();
88             }
89         }        

        /// <summary>/// 获取Model/// </summary>/// <param name="queue">队列名称</param>/// <param name="isProperties"></param>/// <returns></returns>private static IModel GetModel(string queue, bool isProperties = false){return ModelDic.GetOrAdd(queue, value =>{var model = _conn.CreateModel();QueueDeclare(model, queue, isProperties);//每次消费的消息数model.BasicQos(0, 1, false);ModelDic[queue] = model;return model;});}    /// <summary>/// 接收消息/// </summary>/// <typeparam name="T"></typeparam>/// <param name="queue">队列名称</param>/// <param name="isProperties"></param>/// <param name="handler">消费处理</param>/// <param name="isDeadLetter"></param>public void Subscribe<T>(string queue, bool isProperties, Action<T> handler, bool isDeadLetter) where T : class{//队列声明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>();try{handler(msg);}catch (Exception ex){ex.GetInnestException().WriteToFile("队列接收消息", "RabbitMq");if (!isDeadLetter)PublishToDead<DeadLetterQueue>(queue, msgStr, ex);}finally{channel.BasicAck(ea.DeliveryTag, false);}};channel.BasicConsume(queue, false, consumer);}        

        /// <summary>/// 获取消息/// </summary>/// <typeparam name="T"></typeparam>/// <param name="exchange"></param>/// <param name="queue"></param>/// <param name="routingKey"></param>/// <param name="handler">消费处理</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);}}    

        /// <summary>/// RPC客户端/// </summary>/// <param name="exchange"></param>/// <param name="queue"></param>/// <param name="routingKey"></param>/// <param name="body"></param>/// <param name="isProperties"></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 name="T"></typeparam>/// <param name="exchange"></param>/// <param name="queue"></param>/// <param name="isProperties"></param>/// <param name="handler"></param>/// <param name="isDeadLetter"></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);}