小编给大家分享一下RabbitMQ如何实现Publish和Subscribe,相信大部分人都还不怎么了解,因此分享这篇文章给大家参考一下,希望大家阅读完这篇文章后大有收获,下面让我们一起去了解一下吧!
1、消息交换机【Exchange】
在教程的前面部分,我们从队列中发送和接收消息。在RabbitMQ中,现在是时候引入全消息模型。
让我们快速看看我们以前的教程讲了什么:
【生产者】:就是一个用于发送消息的用户程序
【消费者】:就是一个用于接收和使用消息的用户程序
【队列】:是一个暂存消息的缓存区
RabbitMQ消息传递模型的核心思想是,【生产者】不直接发送任何信息到队列。事实上,【生产者】根本就不知道消息是否会被传送到任何队列。
相反,【生产者】只能发送消息到【消息交换机】。交换是件很简单的事。一方面它接收来自【生产者】的消息,另一方面是将接收到消息推送到队列中。【消息交换机】必须知道它如何处理接收消息的确切方法。是否应该发送到特定队列?它应该被发送到多个队列呢?或者它应该被丢弃。该规则由【消息交换机】的类型来定义。
这里有一些可用的【消息交换机】的类型:【Direct】直接,【Topic】主题,【Headers】标题和【Fanout】扇出。我们将集中关注最后一个-【Fanout】扇出。让我们创建一个这种类型的【消息交换机】,并给它命名为Logs:
channel.ExchangeDeclare("logs", "fanout");
【Fanout】类型的【消息交换机】非常简单。正如你从名字可能猜出的,它只是传播它收到的所有消息去它知道所有的队列中。这正是我们需要我们的日志记录器。
显示【消息交换机】的列表:
使用Rabbitmqctl列出在服务器上可以运行的最有用的【消息交换机】
sudo rabbitmqctl list_exchanges
在这个列表中会有一些amq.*【消息交换机】和默认(未命名)消息交换机。这些都是默认创建的,但现在不太可能需要使用它们。
默认的消息交换机
在教程前面的部分我们队【消息交换机】是一无所知,但是我依然可以发送消息去想去的队列,那是因为我们使用了默认的【消息交换机】,这些默认的消息 交换机我用使用两个双引号“”来标识。
我们回忆一下以前是如何发送消息的:
var message = GetMessage(args);var body = Encoding.UTF8.GetBytes(message); channel.BasicPublish(exchange: "", routingKey: "hello", basicProperties: null, body: body);
第一个参数是【消息交换机】的名称。空字符串表示默认或未命名的消息交换机:消息会被路由到指定的routingkey名称的队列,如果它存在的话。
现在,我们可以发布到我们命名的【消息交换机】:
var message = GetMessage(args);var body = Encoding.UTF8.GetBytes(message); channel.BasicPublish(exchange: "logs", routingKey: "", basicProperties: null, body: body);
2、临时队列
也许你还记得以前我们使用的队列所指定的名称(记得Hello和task_queue吗?)对我们来说,能够给一个队列指定名称是至关重要的--因为我们需要把【Worker】指向同一个队列。如果要在【生产者】和【消费者】之间共享队列,给队列命名是很重要的。
但这不是我们的日志记录器的情况。我们想听到所有的日志消息,而不仅仅是其中的一个子集。我们也只对当前刚刚收到的消息感兴趣,而不是对旧的。为了解决上述问题,我们需要做两件事。
首先,无论何时当我们连接到Rabbit的时候,我们都需要一个新的并且是空的队列。要做到这一点,我们可以创建一个具有随机名称的队列,或者,甚至更好一点-让服务器为我们选择一个随机队列名称。
其次,一旦我们断开与【消费者】的队列就应该自动删除该队列。
在.NET客户端中,当我们没有为queueDeclare()提供参数时,我们创建了一个具有生成名称的非持久,排他,自动删除队列:
var queueName = channel.QueueDeclare().QueueName;
在这点上,QueueName包含随机队列名称。例如,它可能看起来像amq.gen-jzty20brgko-hjmujj0wlg。
3、绑定【Binding】
我们已经创建了一个【Fanout】类型的【消息交换机】和队列。现在我们需要告诉【消息交换机】向我们的队列发送消息。【消息交换机】和【队列】之间的关系称为绑定。
channel.QueueBind(queue: queueName, exchange: "logs", routingKey: "");
从现在开始,日志的【消息交换机】就可以将消息推送到我们定义的队列中去了。
我们可以通过以下语句查看【binding】列表数据:
rabbitmqctl list_bindings
4、把所有的代码整合到一起
【生产者】的程序,它发出的日志消息,看起来并没有和以前的教程有很大的不同。最重要的变化是,我们现在想发送的消息是到达我们指定名称的日志【消息交换机】,而不是无名的。我们在发送消息的时候需要提供一个routingkey表示的名称,但【Fanout】类型的【消息交换机】会容忽视该routingKey的值的。
以上是“RabbitMQ如何实现Publish和Subscribe”这篇文章的所有内容,感谢各位的阅读!相信大家都有了一定的了解,希望分享的内容对大家有所帮助,如果还想学习更多知识,欢迎关注亿速云行业资讯频道!
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。