ay的RabbitMQ研究报告-第7章-C#代码实战[3]-发布与订阅

时间：2016年06月01日 | 作者 : aaronyang | 分类 : C#开发 | 浏览: 2245次 | 评论 0 人

（国内第一个rabbitMQ中文系列讲解------引进外国技术，我是AY，安徽合肥的杨洋，1991年。我学习技术，我开心）

为了给自己站点增加流量，望各位抓包的，手下留情，自己去外国下载书籍，回来翻译着看，也不容易。谢谢你了。

====================www.ayjs.net 杨洋 wpfui.com ayui ay aaronyang=======请不要转载谢谢了。

虽然有多个消费者，但是一个消息只会有一个消费者来处理。而订阅和发布则是每个订阅该消息的消费者都会收到这个消息。RabbitMQ的路由机制让我们实现这个功能轻而易举

要了解RabbitMQ的路由机制，exchange是一个关键。exchange可以叫做交换机，也似乎可以叫做路由器，反正它是用来选择路由的。前文说到，RabbitMQ的核心思想就是消息的发布者不是直接把消息发送到目标队列中的，事实上，通常它并不知道消息要发到哪个队列中，它只知道把消息队列发送到exchange中。exchange一边接收发送者发过来的消息，而另一边则把消息发送到目标队列中去。exchange一定知道哪些队列需要接收这个消息，是加到一个队列里还是加到好几个队列里，还是直接扔掉

新建解决方案文件夹3.Publisher and Subscriber，新建2个控制台程序，然后引用common和RabbitMQ 客户端dll

publisher代码如下：

using Common;
using RabbitMQ.Client;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Pulisher
{
    class Program
    {
        private static ConnectionFactory _factory;
        private static IConnection _connection;
        private static IModel _model;
        private const string ExchangeName = "PublishSubscribe_Exchange";
        private static void CreateConnection()
        {
            _factory = new ConnectionFactory
            {
                HostName = "localhost",
                UserName = "guest",
                Password = "guest"
            };
            _connection = _factory.CreateConnection();
            _model = _connection.CreateModel();
            _model.ExchangeDeclare(ExchangeName, "fanout");
        }

        private static void SendMessage(Payment message)
        {
            {
                _model.BasicPublish(ExchangeName, "", null, message.Serialize());
                Console.WriteLine(" Payment Sent {0}, £{1}", message.CardNumber,
                message.AmountToPay);
            }
        }

        static void Main(string[] args)
        {
            var payment1 = new Payment
            {
                AmountToPay = 25.0m,
                CardNumber = "1234123412341234"
            };
            var payment2 = new Payment
            {
                AmountToPay = 5.0m,
                CardNumber = "1234123412341234"
            };
            var payment3 = new Payment
            {
                AmountToPay = 2.0m,
                CardNumber = "1234123412341234"
            };
            var payment4 = new Payment
            {
                AmountToPay = 17.0m,
                CardNumber = "1234123412341234"
            };
            var payment5 = new Payment
            {
                AmountToPay = 300.0m,
                CardNumber =
                    "1234123412341234"
            };
            var payment6 = new Payment
            {
                AmountToPay = 350.0m,
                CardNumber =
                    "1234123412341234"
            };
            var payment7 = new Payment
            {
                AmountToPay = 295.0m,
                CardNumber =
                    "1234123412341234"
            };
            var payment8 = new Payment
            {
                AmountToPay = 5625.0m,
                CardNumber =
                    "1234123412341234"
            };
            var payment9 = new Payment
            {
                AmountToPay = 5.0m,
                CardNumber = "1234123412341234"
            };
            var payment10 = new Payment
            {
                AmountToPay = 12.0m,
                CardNumber =
                    "1234123412341234"
            };
            CreateConnection();
            SendMessage(payment1);
            SendMessage(payment2);
            SendMessage(payment3);
            SendMessage(payment4);
            SendMessage(payment5);
            SendMessage(payment6);
            SendMessage(payment7);
            SendMessage(payment8);
            SendMessage(payment9);
            SendMessage(payment10);
        }
    }
}

====================www.ayjs.net 杨洋 wpfui.com ayui ay aaronyang=======请不要转载谢谢了。=========

这块代码跟我们在Worker Queue的Demo是几乎一样的。查看

唯一不一样的，我们创建了一个fanout类型的exchange路由器。

_model.ExchangeDeclare(ExchangeName, "fanout");

_model.BasicPublish(ExchangeName, "", null, message.Serialize());

接下来添加订阅的消费者

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Common;
using RabbitMQ.Client;

namespace Subscriber
{
    class Program
    {
        private static ConnectionFactory _factory;
        private static IConnection _connection;
        private static QueueingBasicConsumer _consumer;
        private const string ExchangeName = "PublishSubscribe_Exchange";
        private static string DeclareAndBindQueueToExchange(IModel channel)
        {
            channel.ExchangeDeclare(ExchangeName, "fanout");
            var queueName = channel.QueueDeclare().QueueName;
            channel.QueueBind(queueName, ExchangeName, "");
            _consumer = new QueueingBasicConsumer(channel); 
            return queueName;
        }
        static void Main()
        {
            _factory = new ConnectionFactory
            {
                HostName = "localhost",
                UserName = "guest",
                Password = "guest"
            };
            using (_connection = _factory.CreateConnection())
            {
                using (var channel = _connection.CreateModel())
                {
                    var queueName = DeclareAndBindQueueToExchange(channel);
                    channel.BasicConsume(queueName, true, _consumer);
                    while (true)
                    {
                        var ea = _consumer.Queue.Dequeue();
                        var message = (Payment)ea.Body.DeSerialize();
                        Console.WriteLine("----- Payment Processed {0} : {1}",
                        message.CardNumber, message.AmountToPay);
                    }
                }
            }

      
        }
    }
}

消费者代码ay的讲解：

ExchangeDeclare 路由器声明：RabbitMQ存在就返回存在的，不存在就创建新的返回，跟QueueDeclare是一样的。

这里我们指定queue的名字，跟以前的例子不太一样，对于订阅程序，queuename有点特殊，需要系统的方式生成queue的name。

例如

amq.gen-TsdoX9qziswm9QCbdkp9Zw

每次启动订阅端，都会创建一个channel，每一个channel都会按照PublishSubscribe_Exchange这个名字找到exchange，然后binding一个queue，然后RabbitMQ指定名字，然后消费。

运行2个订阅者