RabbitMQ教程C#版 - 发布订阅

先决条件
本教程假定 RabbitMQ 已经安装，并运行在localhost标准端口（5672）。如果你使用不同的主机、端口或证书，则需要调整连接设置。

从哪里获得帮助
如果您在阅读本教程时遇到困难，可以通过邮件列表 联系我们。

发布/订阅#

（使用 .NET Client）

在 教程[2] 中，我们创建了一个工作队列，假设在工作队列中的每一个任务都只被分发给一个 Worker。那么在这一章节，我们要做与之完全不同的事，那就是我们将要把一条消息分发给多个消费者。这种模式被称为“发布/订阅”。

为了说明、体现这种模式，我们将会建一个简单的日志系统。它将会包含两个程序 - 第一个用来发送日志消息，第二个用来接收并打印它们。

在我们建立的日志系统中，每个接收程序的运行副本都会收到消息。这样我们就可以运行一个接收程序接收消息并将日志写入磁盘；同时运行另外一个接收程序接收消息并将日志打印到屏幕上。

实质上，发布的日志消息将会被广播给所有的接收者。

交换器#

在教程的前几部分，我们是发送消息到队列并从队列中接收消息。现在是时候介绍 Rabbit 中完整的消息传递模型了。

让我们快速回顾一下前面教程中的内容：

  • 生产者是发送消息的用户应用程序。
  • 队列是存储消息的缓冲区。
  • 消费者是接收消息的用户应用程序。

在 RabbitMQ 中，消息传递模型的核心理念是生产者从来不会把任何消息直接发送到队列，其实，通常生产者甚至不知道消息是否会被分发到任何队列中。

然而，生产者只能把消息发送给交换器。交换器非常简单，一方面它接收来自生产者的消息，另一方面又会把接收的消息推送到队列中。交换器必须明确知道该如何处理收到的消息，应该追加到一个特定队列中？还是应该追加到多个队列中？或者应该把它丢弃？这些规则都被定义在交换器类型中。

目前交换器类型有这几种：direct，topic，headers和fanout。我们先重点关注最后一个fanout，我们创建一个这种类型的交换器，将其命名为logs：

Copy

channel.ExchangeDeclare("logs", "fanout");

fanout类型交换器非常简单，正如您可能从名字中猜出的那样，它会把收到的所有消息广播到它已知的所有队列中。这恰巧是我们的日志系统目前所需要的。

列举交换器
要列举出服务器上的交换器，您可以使用非常有用的rabbitmqctl命令行工具：

Copy

sudo rabbitmqctl list_exchanges

执行上述命令后，出现的列表中将会有一些amq.*交换器和默认（未命名）交换器。这些是默认创建的，不过目前您可能用不到它们。

默认交换器
在教程的前些部分，我们对交换器这一概念还一无所知，但仍然可以把消息发送到队列。之所以这样，是因为我们使用了一个用空字符串("")标识的默认交换器。

回顾一下我们之前如何发布消息：

Copy

var message = GetMessage(args);
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(exchange: "",
                     routingKey: "hello",
                     basicProperties: null,
                     body: body);

第一个参数就是交换器的名称，空字符串表示默认或匿名交换器：将消息路由到routingKey指定的队列（如果存在）中。

现在，我们可以把消息发布到我们指定的交换器：

Copy

var message = GetMessage(args);
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(exchange: "logs",
                     routingKey: "",
                     basicProperties: null,
                     body: body);

临时队列#

您是否还记得之前我们使用过的队列，它们都有一个特定的名称（记得应该是hello和task_queue吧）。给队列命名对我们来说是至关重要的 -- 因为我们可能需要多个 Worker 指向同一个队列；当您想要在生产者和消费者之间共享队列时，给队列一个名称也是非常重要的。

但是，我们创建的日志系统并不希望如此。我们希望监听所有的日志消息，而不仅仅是其中一部分。我们也只对目前流动的消息感兴趣，而不是旧消息。为解决这个问题，我们需要做好两件事。

首先，我们无论何时连接 Rabbit，都需要一个新的、空的队列。要做到这一点，我们可以使用随机名称来创建队列，或许，甚至更好的方案是让服务器为我们选择一个随机队列名称。

其次，一旦我们与消费者断开连接，与之相关的队列应该被自动删除。

在 .NET 客户端中，如果不向QueueDeclare()方法提供任何参数，实际上就是创建了一个非持久化、独占、且自动删除的随机命名队列：

Copy

var queueName = channel.QueueDeclare().QueueName;

您可以在 队列指南 中了解更多关于exclusive参数和其他队列属性的信息。

此时，queueName包含一个随机队列名称。例如，它看起来可能像amq.gen-JzTY20BRgKO-HjmUJj0wLg。

绑定#

我们已经创建好了一个fanout交换器和一个队列。现在我们需要告诉交换器把消息发送到我们的队列。而交换器和队列之间的关系就称之为绑定。

Copy

// 把一个队列绑定到指定交换器。
channel.QueueBind(queue: queueName,
                  exchange: "logs",
                  routingKey: "");

从现在起，logs交换器会把消息追加到我们的队列中。

列举绑定
您可以使用（您或许已经猜到了），列举出现有的绑定。

Copy

sudo rabbitmqctl list_bindings

组合在一起#

生产者程序负责分发消息，这与之前的教程看起来没有太大区别。

最重要的变化是我们现在想把消息发布到我们的logs交换器，而不是匿名交换器。在发送时我们需要提供一个路由键routingKey，但是对于fanout交换器，它的值可以被忽略。这里是EmitLog.cs文件的代码：

Copy

using System;
using RabbitMQ.Client;
using System.Text;

class EmitLog
{
    public static void Main(string[] args)
    {
        var factory = new ConnectionFactory() { HostName = "localhost" };
        using(var connection = factory.CreateConnection())
        using(var channel = connection.CreateModel())
        {
            channel.ExchangeDeclare(exchange: "logs", type: "fanout");

            var message = GetMessage(args);
            var body = Encoding.UTF8.GetBytes(message);
            channel.BasicPublish(exchange: "logs",
                                 routingKey: "",
                                 basicProperties: null,
                                 body: body);
            Console.WriteLine(" [x] Sent {0}", message);
        }

        Console.WriteLine(" Press [enter] to exit.");
        Console.ReadLine();
    }

    private static string GetMessage(string[] args)
    {
        return ((args.Length > 0)
               ? string.Join(" ", args)
               : "info: Hello World!");
    }
}

（EmitLog.cs 源码）

如你所见，在建立连接后，我们声明了交换器。这一步非常有必要，因为发布消息到一个不存在的交换器，这种情况是被禁止的。

如果没有队列绑定到交换器上，消息将会丢失，但这对我们来说并没有什么没问题；如果没有消费者正在监听，我们是可以放心地把消息丢弃的。

ReceiveLogs.cs的代码：

Copy

using System;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;

class ReceiveLogs
{
    public static void Main()
    {
        var factory = new ConnectionFactory() { HostName = "localhost" };
        using(var connection = factory.CreateConnection())
        using(var channel = connection.CreateModel())
        {
            channel.ExchangeDeclare(exchange: "logs", type: "fanout");

            var queueName = channel.QueueDeclare().QueueName;
            channel.QueueBind(queue: queueName,
                              exchange: "logs",
                              routingKey: "");

            Console.WriteLine(" [*] Waiting for logs.");

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body;
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine(" [x] {0}", message);
            };
            channel.BasicConsume(queue: queueName,
                                 autoAck: true,
                                 consumer: consumer);

            Console.WriteLine(" Press [enter] to exit.");
            Console.ReadLine();
        }
    }
}

（ReceiveLogs.cs 源码）

按照 教程[1]中的设置说明生成EmitLogs和ReceiveLogs 项目。

如果您想把日志保存到文件中，只需打开一个控制台并输入：

Copy

cd ReceiveLogs
dotnet run > logs_from_rabbit.log

如果你想在屏幕上看到日志，我可以新开一个终端并运行：

Copy

cd ReceiveLogs
dotnet run

当然，分发日志需要输入：

Copy

cd EmitLog
dotnet run

使用rabbitmqctl list_bindings命令，您可以验证代码是否真正创建了我们想要的绑定和队列。当有两个ReceiveLogs.cs程序运行时，您应该看到如下所示的内容：

Copy

sudo rabbitmqctl list_bindings
# => Listing bindings ...
# => logs    exchange        amq.gen-JzTY20BRgKO-HjmUJj0wLg  queue           []
# => logs    exchange        amq.gen-vso0PVvyiRIL2WoV3i48Yg  queue           []
# => ...done.

对执行结果的解释简洁明了：来自logs交换器的数据转发到了两个由服务器随机分配名称的队列。这正是我们期待的结果。

想要了解如何监听消息的这一块内容，让我们继续阅读 教程[4]。

写在最后#

本文翻译自 RabbitMQ 官方教程 C# 版本。如本文介绍内容与官方有所出入，请以官方最新内容为准。水平有限，翻译的不好请见谅，如有翻译错误还请指正。

原文地址：https://www.cnblogs.com/Alex80/p/10848180.html

时间： 2024-11-10 05:30:08

RabbitMQ教程C#版 - 发布订阅的相关文章

RabbitMQ（三） ——发布订阅

RabbitMQ(三) --发布订阅 (转载请附上本文链接--linhxx) 一.概述 RabbitMQ的发布订阅(Publish/Subscribe),其将生产者和消费者进一步解耦,生产者生产消息后,交付给交换机,消费者上线后,主动主动去队列中取数据进行处理.该模式也符合上一节工作队列中的ack.预取等规则. 发布订阅模式如下图所示: 二.交换机(exchange) 生产者生产完消息之后,都是将消息通过channel交给交换机,即生产者并不直接和队列联系.在没有定义交换机的时候,RabbitM

RabbitMQ系列教程之三：发布/订阅（Publish/Subscribe）

(本教程是使用Net客户端,也就是针对微软技术平台的) 在前一个教程中,我们创建了一个工作队列.工作队列背后的假设是每个任务会被交付给一个[工人].在这一部分我们将做一些完全不同的事情--我们将向多个[消费者]传递信息.这种模式被称为"发布/订阅". 为了说明这种模式,我们将构建一个简单的日志系统.它将包括两个程序,第一个将发出日志消息,第二个将接收并打印它们. 在我们的日志系统中每个接收程序的运行副本都会得到消息.这样我们就可以运行一个接收者程序,将日志记录到磁盘:同时我们可以运行另

rabbitmq系列三 之发布/订阅

1.发布/订阅 在上篇教程中,我们搭建了一个工作队列,每个任务只分发给一个工作者(worker).在本篇教程中,我们要做的跟之前完全不一样 -- 分发一个消息给多个消费者(consumers).这种模式被称为"发布/订阅". 为了描述这种模式,我们将会构建一个简单的日志系统.它包括两个程序--第一个程序负责发送日志消息,第二个程序负责获取消息并输出内容. 在我们的这个日志系统中,所有正在运行的接收方程序都会接受消息.我们用其中一个接收者(receiver)把日志写入硬盘中,另外一个接受

python使用rabbitMQ介绍三（发布订阅模式）

一.模式介绍 在前面的例子中,消息直接发送到queue中. 现在介绍的模式,消息发送到exchange中,消费者把队列绑定到exchange上. 发布-订阅模式是把消息广播到每个消费者,每个消费者接收到的消息都是相同的. 一个生产者,多个消费者,每一个消费者都有自己的一个队列,生产者没有将消息直接发送到队列,而是发送到了交换机,每个队列绑定交换机,生产者发送的消息经过交换机,到达队列,实现一个消息被多个消费者获取的目的.需要注意的是,如果将消息发送到一个没有队列绑定的exchange上面,那么该

RabbitMQ指南之三：发布/订阅模式（Publish/Subscribe）

在上一章中,我们创建了一个工作队列,工作队列模式的设想是每一条消息只会被转发给一个消费者.本章将会讲解完全不一样的场景: 我们会把一个消息转发给多个消费者,这种模式称之为发布-订阅模式. 为了阐述这个模式,我们将会搭建一个简单的日志系统,它包含两种程序:一种发送日志消息,另一种接收并打印日志消息.在这个日志系统里,每一个运行的消费者都可以获取到消息,在这种情况下,我们可以实现这种需求:一个消费者接收消息并写入磁盘,另一个消费者接收消息并打印在电脑屏幕上.简单来说,生产者发布的消息将会以广播的形式

Redis教程03——Redis 发布/订阅（Pub/Sub）

Pub/Sub 订阅,取消订阅和发布实现了发布/订阅消息范式(引自wikipedia),发送者(发布者)不是计划发送消息给特定的接收者(订阅者).而是发布的消息分到不同的频道,不需要知道什么样的订阅者订阅.订阅者对一个或多个频道感兴趣,只需接收感兴趣的消息,不需要知道什么样的发布者发布的.这种发布者和订阅者的解耦合可以带来更大的扩展性和更加动态的网络拓扑. F为了订阅foo和bar,客户端发出一个订阅的频道名称: SUBSCRIBE foo bar 其他客户端发到这些频道的消息将会被推送到所有订

rabbitMQ学习笔记(四) 发布/订阅消息

前面都是一条消息只会被一个消费者处理. 如果要每个消费者都处理同一个消息,rabbitMq也提供了相应的方法. 在以前的程序中,不管是生产者端还是消费者端都必须知道一个指定的QueueName才能发送.获取消息.  而rabbitMQ消息模型的核心思想是生产者不会将消息直接发送给队列. 因为,生产者通常不会知道消息将会被哪些消费者接收. 生产者的消息虽然不是直接发送给Queue,但是消息会交给Exchange,所以需要定义Exchange的消息分发模式 ,之前的程序中,有如下一行代码: chan

RabbitMQ官网教程---发布/订阅

(使用python客户端pika 0.9.8) 在前面的教程中我们创建了一个工作队列.假设在一个工作队列后面是每一个被传递给正确的工作者的任务.在这部分中我们将做一些完全不同的事情--我们将给多个消费者传递一个消息.这种模式被称为"发布/订阅". 为了阐明这个模式,我们将构建一个简单的日志系统.它将由两个程序构成--第一个将发出日志消息并且第二个将接收并且打印它们. 在我们的日志系统中每个运行的接收程序副本将获得这个消息.用这种方式我们将可以运行一个接收器并且直接日志到磁盘:而且同时我

RabbitMQ入门教程(五)：扇形交换机发布/订阅(Publish/Subscribe)

原文:RabbitMQ入门教程(五):扇形交换机发布/订阅(Publish/Subscribe) 版权声明:本文为博主原创文章,遵循CC 4.0 BY-SA版权协议,转载请附上原文出处链接和本声明. 本文链接:https://blog.csdn.net/vbirdbest/article/details/78628659 分享一个朋友的人工智能教程.比较通俗易懂,风趣幽默,感兴趣的朋友可以去看看. 简介 本节主要演示交换机的广播类型fanout,广播类型不需要routingKey,交换机会将所有