简单入门Rabbitmq

相关概念说明

Broker：标识消息队列服务器实体.相当于安装的RabbitMQ。

Virtual Host：虚拟主机。拥有自己的队列、交换器、绑定和权限机制。相当于在一个broker中实现多租户，一个虚拟主机给一个用户使用。

Queue：消息队列，用来保存消息直到被消费者消费掉（取出消息并回复）。

Exchange：交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。
一般消息提供者不会直接将消息发送到指定的队列中，而是先将消息发送至交换器，交换器再将消息按照交换规则发送到指定队列。

Banding：绑定，用于消息队列和交换机之间的关联。一个绑定就是基于路由键将交换机和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。

Channel：信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内地虚拟链接，AMQP命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，
这些动作都是通过信道完成。因为对于操作系统来说，建立和销毁TCP都是非常昂贵的开销，所以引入了信道的概念，以复用一条TCP连接。

Connection：连接，就是与rabbitmq服务的一个socket连接。

Publisher：消息的生产者，也就是是一个向交换器发布消息的服务或者程序。

Consumer：消息的消费者，表示一个从一个消息队列中取得消息的服务或者程序。

 1        public void PublishDirectMessage(string queueName, string exchangeName, string routeKey, string message)
 2         {
 3 
 4             var factory = new ConnectionFactory()
 5             {
 6                 HostName = _config.Host,
 7                 VirtualHost = _config.VirtualHost,
 8                 UserName = _config.UserName,
 9                 Password = _config.Password,
10             };
11             using (var connection = factory.CreateConnection())
12             {
13                 using (var channel = connection.CreateModel())
14                 {
15                     //定义交换机
16                     channel.ExchangeDeclare(exchangeName, ExchangeType.Direct, true, false, null);
17                     //定义队列
18                     channel.QueueDeclare(queueName, true, false, false, null);
19                     //绑定
20                     channel.QueueBind(queueName, exchangeName, routeKey, null);
21                     byte[] data = Encoding.UTF8.GetBytes(message);
22                     //发送消息
23                     channel.BasicPublish(exchangeName, routeKey, null, data);
24                 }
25             }
26         }

 1        public void PublishTopicMessage(string queueName, string exchangeName, string routeKey, string message)
 2         {
 3 
 4             var factory = new ConnectionFactory()
 5             {
 6                 HostName = _config.Host,
 7                 VirtualHost = _config.VirtualHost,
 8                 UserName = _config.UserName,
 9                 Password = _config.Password,
10             };
11             using (var connection = factory.CreateConnection())
12             {
13                 using (var channel = connection.CreateModel())
14                 {
15                     channel.QueueDeclare(queueName, true, false, false, null);
16                     channel.ExchangeDeclare(exchangeName, ExchangeType.Topic, true, false, null);
17                     channel.QueueBind(queueName, exchangeName, routeKey, null);
18                     byte[] data = Encoding.UTF8.GetBytes(message);
19                     channel.BasicPublish(exchangeName, routeKey, null, data);
20                 }
21             }
22         }

 1        public void PublishFanoutMessage(string queueName, string exchangeName, string message)
 2         {
 3 
 4             var factory = new ConnectionFactory()
 5             {
 6                 HostName = _config.Host,
 7                 VirtualHost = _config.VirtualHost,
 8                 UserName = _config.UserName,
 9                 Password = _config.Password,
10             };
11             using (var connection = factory.CreateConnection())
12             {
13                 using (var channel = connection.CreateModel())
14                 {
15                     channel.QueueDeclare(queueName, true, false, false, null);
16                     channel.ExchangeDeclare(exchangeName, ExchangeType.Fanout, true, false, null);
17 
18                     byte[] data = Encoding.UTF8.GetBytes(message);
19                     channel.BasicPublish(exchangeName, "", null, data);
20                 }
21             }
22         }

 1         public void HandleMessage(string queueName)
 2         {
 3 
 4             //创建工厂--建立连接---创建channel--定义事件消费者并绑定事件--指定消费队列
 5             var factory = new ConnectionFactory()
 6             {
 7                 HostName = _config.Host,
 8                 VirtualHost = _config.VirtualHost,
 9                 UserName = _config.UserName,
10                 Password = _config.Password
11             };
12             var connection = factory.CreateConnection();
13             var channel = connection.CreateModel();
14             channel.QueueDeclare(queueName,true,false,false,null);
15             // channel.BasicQos(1,1,false);
16             ///定义事件消费者
17             EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
18             consumer.Received += Consumer_Received;
19             //消费
20             channel.BasicConsume(queueName, false, consumer);
21             //此处不释放channel,应该在事件处理中完成后释放，不然可能存在事件未处理完，而channel释放，导致消息没能被移除
22             //channel.Dispose();
23             //connection.Close();
24         }
25 
26         private void Consumer_Received(object sender, BasicDeliverEventArgs e)
27         {
28             var message = Encoding.UTF8.GetString(e.Body.ToArray());
29             #region 业务处理
30             Console.WriteLine(message);
31             #endregion
32             EventingBasicConsumer consumer = sender as EventingBasicConsumer;
33             //string exchangeName = e.Exchange;
34             //string routeKey = e.RoutingKey; 
35             //设置已经被消费
36             consumer.Model.BasicAck(e.DeliveryTag, false);
37             //释放Channel
38             consumer.Model.Dispose();
39 
40         }