RabbitMQ---3、c#实现

1、EasyNetQ组件的使用

EasyNetQ组件的使用方式比较简单，跟很多组件都类似，例如：建立连接，进行操作做等等，对于EasyNetQ组件也是如此。（mq的升级，用于简化rabbitmq应用代码）

在.NET中使用EasyNetQ，要求至少基于 .NET4.5的框架基础上进行开发，可以直接在VS项目上使用NuGet的程序包进行添加EasyNetQ的引用。

一般添加引用后，至少包含了下面图示的几个引用DLL。

2、相关概念

  • ConnectionFactory(连接工厂): 生产Connection的的工厂

  • Connection(连接):是RabbitMQ的socket的长链接，它封装了socket协议相关部分逻辑

  • Channel(频道|信道): 是建立在Connection连接之上的一种轻量级的连接，我们大部分的业务操作是在Channel这个接口中完成的，包括定义队列的声明queueDeclare、交换机的声明exchangeDeclare、队列的绑定queueBind、发布消息basicPublish、消费消息basicConsume等。如果把Connection比作一条光纤电缆的话，那么Channel信道就比作成光纤电缆中的其中一束光纤。一个Connection上可以创建任意数量的Channel。 
    这里写图片描述

    这里写图片描述

  • Producer(生产者)：生产者用于发布消息

  • Exchange(交换机)：生产者会将消息发送到交换机，然后交换机通过路由策略(规则)将消息路由到匹配的队列中去

  • Routing Key(路由键)：一个String值，用于定义路由规则，在队列绑定的时候需要指定路由键，在生产者发布消息的时候需要指定路由键，当消息的路由键和队列绑定的路由键匹配时，消息就会发送到该队列。

  • Queue(队列):用于存储消息的容器，可以看成一个有序的数组，生产者生产的消息会发送到交换机中，最终交换机将消息存储到某个或某些队列中，队列可被消费者订阅，消费者从订阅的队列中获取消息。

  • Binding(绑定): Binding并不是一个概念，而是一种操作，RabbitMQ中通过绑定，以路由键作为桥梁将Exchange与Queue关联起来(Exchange—>Routing Key—>Queue)，这样RabbitMQ就知道如何正确地将消息路由到指定的队列了，通过queueBind方法将Exchange、Routing Key、Queue绑定起来

  • Consumer(消费者)：用于从队列中获取消息，消费者只需关注队列即可，不需要关注交换机和路由键，消费者可以通过basicConsume(订阅模式可以从队列中一直持续的自动的接收消息)或者basicGet(先订阅消息，然后获取单条消息，再然后取消订阅，也就是说basicGet一次只能获取一条消息，如果还想再获取下一条还要再次调用basicGet)来从队列中获取消息

  • vhost(虚拟主机): RabbitMQ 通过虚拟主机（virtual host）来分发消息, 拥有自己独立的权限控制，不同的vhost之间是隔离的，单独的。vhost是权限控制的基本单位，用户只能访问与之绑定的vhost，默认vhost：”/” ,默认用户”guest” 密码“guest”，来访问默认的vhost。

下图说明了生产者将消息发送到交换机中，然后又路由到指定的队列中 
这里写图片描述

3、分发与订阅使用

4、代码编写
为了测试发布和订阅消息，我们可以建立几个不同的项目来进行测试，如发布放在一个Winform项目，订阅放在一个Winform项目，另外一个项目放置共享的消息对象定义，如下所示。
Producter 使用

1）发送消息代码：

  1. /// <summary>
  2.  /// 连接配置
  3.  /// </summary>
  4.  private static readonly ConnectionFactory rabbitMqFactory = new ConnectionFactory(){
  5.  HostName ="192.168.1.8",UserName="hao",Password="abc123",Port= 5672
  6.  };
  7. /// <summary>
  8. /// 路由名称
  9. /// </summary>
  10. const string TopExchangeName = "topic.justin.exchange";
  11. //队列名称
  12. const string TopQueueName = "topic.justin.queue";
  13. public static void TopicExchangeSendMsg()
  14.  {
  15. using (IConnection conn = rabbitMqFactory.CreateConnection())
  16.  {
  17. using (IModel channel = conn.CreateModel())
  18.  {
  19.  channel.ExchangeDeclare(TopExchangeName, "topic", durable: false, autoDelete: false, arguments: null);
  20.  channel.QueueDeclare(TopQueueName, durable: false, autoDelete: false, exclusive: false, arguments: null);
  21.  channel.QueueBind(TopQueueName, TopExchangeName, routingKey: TopQueueName);
  22. //var props = channel.CreateBasicProperties();
  23. //props.Persistent = true;
  24. string vadata = Console.ReadLine();
  25. while (vadata != "exit")
  26.  {
  27. var msgBody = Encoding.UTF8.GetBytes(vadata);
  28.  channel.BasicPublish(exchange: TopExchangeName, routingKey: TopQueueName, basicProperties: null, body: msgBody);
  29.  Console.WriteLine(string.Format("***发送时间:{0}，发送完成，输入exit退出消息发送", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")));
  30.  vadata = Console.ReadLine();
  31.  }
  32.  }
  33.  }
  34.  }

2）测试验证（登入rabbitmq服务器）

Customer

1）接收消息代码：

  1.  /// <summary>
  2.  /// 连接配置
  3.  /// </summary>
  4.  private static readonly ConnectionFactory rabbitMqFactory = new ConnectionFactory() {
  5.  HostName = "192.168.1.8", UserName = "hao", Password = "abc123", Port = 5672
  6.  };
  7.   
  8.  /// <summary>
  9.  /// 路由名称
  10.  /// </summary>
  11.  const string TopExchangeName = "topic.justin.exchange";
  12.   
  13.  //队列名称
  14.  const string TopQueueName = "topic.justin.queue";
  15.   
  16.  public static void TopicAcceptExchange()
  17.  {
  18.  using (IConnection conn = rabbitMqFactory.CreateConnection())
  19.  {
  20.  using (IModel channel = conn.CreateModel())
  21.  {
  22.  channel.ExchangeDeclare(TopExchangeName, "topic", durable: false, autoDelete: false, arguments: null);
  23.  channel.QueueDeclare(TopQueueName, durable: false, autoDelete: false, exclusive: false, arguments: null);
  24.  channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
  25.  channel.QueueBind(TopQueueName, TopExchangeName, routingKey: TopQueueName);
  26.  var consumer = new EventingBasicConsumer(channel);
  27.  consumer.Received += (model, ea) =>
  28.  {
  29.  var msgBody = Encoding.UTF8.GetString(ea.Body);
  30.  Console.WriteLine(string.Format("***接收时间:{0}，消息内容：{1}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"), msgBody));
  31.  int dots = msgBody.Split('.').Length - 1;
  32.  System.Threading.Thread.Sleep(dots * 1000);
  33.  Console.WriteLine(" [x] Done");
  34.  channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
  35.  };
  36.  channel.BasicConsume(TopQueueName, noAck: false, consumer: consumer);
  37.   
  38.  Console.WriteLine("按任意值，退出程序");
  39.  Console.ReadKey();
  40.  } 
  41. }
  42.  }

2）查看queue中的messages字段情况

 参考资料：https://cloud.tencent.com/developer/article/1151202

.net api应用资料：

http://www.rabbitmq.com/dotnet-api-guide.html#recovery  ；

https://tw.wxwenku.com/d/101521853  ；

https://blog.csdn.net/WuLex/article/details/62043912;https://www.codeproject.com/Articles/309786/Rabbit-Mq-Shovel-Example  ;

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/xiaohua19920/p/9583074.html