C#使用RabbitMQ-3_发布订阅模式（扇形交换机）

今天药忘吃喽~ 2024-04-23 20:23 66阅读 0赞

## 简介 ##

发布订阅模式允许一个生产者向多个消费者发送消息。在RabbitMQ中实现发布订阅模式通常涉及以下几个关键组件：

1.  **生产者**：负责生产并发送消息到RabbitMQ的Exchange（路由器）。
2.  **Exchange**：作为消息的中转站，根据不同的规则将消息路由到一个或多个队列。
3.  **队列**：存储消息的缓冲区，每个消费者有自己的独立队列。
4.  **消费者**：从自己的队列中接收并消费消息。

在这种模式下，生产者发送的消息不是直接发送到特定的队列，而是发送给Exchange。Exchange根据配置的规则决定如何处理这些消息。例如，它可以将消息路由到一个特定的队列，也可以将消息路由到多个队列，或者在某些情况下废弃消息。

在实际应用中，发布订阅模式常用于构建实时通信系统、通知服务、日志系统等场景，其中多个消费者需要接收来自同一生产者的消息。这种模式的优势在于能够实现一对多的通信，使得消息的分发更加灵活和高效。

![ec5c9cbe82734376aa1694f3a8fdcc4a.png][]

## 生产者代码 ##

在前面的模式中，我们使用了channel.QueueDeclare()来声明队列。这里不需要了

channel.QueueDeclare("hello", true, false, false, null);

在发布订阅模式中，生产者只需要将消息发送到交换机上，然后由交换机根据绑定规则将消息路由到一个或多个队列中。消费者则可以从自己的队列中获取并处理这些消息。

因此，我们这里只声明一个扇形交换机，并将消息发布到该交换机上即可。而具体的队列声明和绑定操作可以在消费者端进行。

在前面的第一章和第二章中，我们都没有声明交换机，这是因为RabbitMQ中有一个默认的交换机，称为空字符串名称的默认交换机。当生产者发送消息但没有指定交换机时，消息会被发送到这个默认交换机。同样，当创建队列但没有指定队列与交换机的绑定关系时，队列会自动绑定到默认交换机上。RabbitMQ提供了几种内置的交换机类型，如直接交换（direct）、扇形交换（fanout）、主题交换（topic）和头交换（headers）等，以满足不同的业务场景需求。

class MyClass
    {
        public static void Main(string[] args)
        {
            var factory = new ConnectionFactory();
            factory.HostName = "localhost"; //RabbitMQ服务在本地运行
            factory.UserName = "guest"; //用户名
            factory.Password = "guest"; //密码
    
            //创建连接
            using (var connection = factory.CreateConnection())
            {
                //创建通道
                using (var channel = connection.CreateModel())
                {
                    //声明了一个扇形交换机（fanout exchange），命名为"hello"
                    channel.ExchangeDeclare("hello", "fanout");
    
                    string msg;
                    Console.WriteLine("请输入要发送的消息内容：");
                    while (!string.IsNullOrEmpty(msg = Console.ReadLine()))
                    {
                        var body = Encoding.UTF8.GetBytes(msg);
    
                        channel.BasicPublish("hello", "", null, body); //开始传递
                        Console.WriteLine("已发送： {0}", msg);
                    }
                }
            }
        }
    }

## 消费者代码 ##

我们通过channel.QueueDeclare().QueueName;声明一个新的队列，如果这个方法声明队列，RabbitMQ会自动为你生成一个独一无二的队列名称

var queueName = channel.QueueDeclare().QueueName;
    channel.QueueBind(queueName, "hello", "");

然后将队列的名称赋值给变量`queueName`。将该队列绑定到之前声明的扇形交换机hello上，使用空字符串作为路由键。每次运行这个项目时，都会创建一个新的队列并将其绑定到交换机上。这样，多个消费者可以同时连接到同一个交换机，并从不同的队列中接收消息。

class MyClass
    {
        static void Main(string[] args)
        {
            //创建连接工厂
            var factory = new ConnectionFactory();
            factory.HostName = "localhost";
            factory.UserName = "guest";
            factory.Password = "guest";
            //创建连接
            using (var connection = factory.CreateConnection())
            {
                //创建通道
                using (var channel = connection.CreateModel())
                {
                    //声明了一个扇形交换机（fanout exchange），命名为"hello"
                    channel.ExchangeDeclare("hello", "fanout");
                    
                    //声明一个新的队列，并将这个队列的名称赋值给变量 queueName
                    var queueName = channel.QueueDeclare().QueueName;
                    channel.QueueBind(queueName, "hello", "");
                    //事件的基本消费者
                    var consumer = new EventingBasicConsumer(channel);
    
                    consumer.Received += (model, ea) =>
                    {
                        var body = ea.Body.ToArray();
                        var message = Encoding.UTF8.GetString(body);
    
                        Console.WriteLine("已接收： {0}", message);
                    };
    
                    channel.BasicConsume(queueName, true, consumer);
                    Console.ReadKey();
                }
            }
        }
    }

## 代码演示 ##

和我前面文章的步骤一样，将消费者先进行发布打包，双击.exe文件运行多次项目。

![0b04783611fb483cb97c7f4bc8a90c61.png][]

我们前面提到的使用channel.QueueDeclare().QueueName;声明一个新的队列，在RabbitMQ管理界面可以看到有三个自动生成名称的队列。

![6fd83129fdb242439c18a4ea2834db20.png][]

然后我们启动生产者，并随机发送几条消息

![fe7251678e02421f92ebf952012c4aa6.png][]

再回到消费者，我们运行的三个消费端都同时的收到了消息！

![35bf5fdb7c804066903cacb1d6d25e08.png][]

[ec5c9cbe82734376aa1694f3a8fdcc4a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/23/eb106eaca6214fe391702c31a7c0b648.png
[0b04783611fb483cb97c7f4bc8a90c61.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/23/bb0e39bdc98042439943c33d5c0a7791.png
[6fd83129fdb242439c18a4ea2834db20.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/23/f4e24ed867df4c50a099623ee222b583.png
[fe7251678e02421f92ebf952012c4aa6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/23/61ab45a641144ec5b631bc994ee1c66e.png
[35bf5fdb7c804066903cacb1d6d25e08.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/23/3f4465b122324c50a1688447b905a73d.png