热门标签 | HotTags
当前位置:  开发笔记 > 编程语言 > 正文

开发笔记:python中的rabbitmq

篇首语:本文由编程笔记#小编为大家整理,主要介绍了python中的rabbitmq相关的知识,希望对你有一定的参考价值。

篇首语:本文由编程笔记#小编为大家整理,主要介绍了python中的rabbitmq相关的知识,希望对你有一定的参考价值。



介绍

RabbitMQ是一个消息代理:它接受和转发消息。你可以把它想象成一个邮局:当你把你想要发布的邮件放在邮箱中时,你可以确定邮差先生最终将邮件发送给你的收件人。在这个比喻中,RabbitMQ是邮政信箱,邮局和邮递员。

RabbitMQ和邮局的主要区别在于它不处理纸张,而是接受,存储和转发二进制数据块 -- 消息。

请注意,生产者,消费者和消息代理不必驻留在同一主机上; 实际上在大多数应用程序中它们不是同一主机上。


Hello World!

(using the Pika Python client)


pip3 install pika

在本教程的这一部分,我们将使用Python编写两个小程序; 发送单个消息的生产者(发送者),以及接收消息并将其打印出来的消费者(接收者)。这是一个消息传递的“Hello World”。

在下图中,“P”是我们的生产者,“C”是我们的消费者。中间的盒子是一个队列 - RabbitMQ代表消费者保存的消息缓冲区。

我们的整体设计将如下所示:

 技术分享图片

 

生产者将消息发送到“hello”队列,消费者接收来自该队列的消息。

发送

 技术分享图片

 

我们的第一个程序 send.py 会向队列发送一条消息。我们需要做的第一件事是与RabbitMQ服务器建立连接。

#!/usr/bin/env python
import pika
cOnnection= pika.BlockingConnection(pika.ConnectionParameters(‘localhost‘))
channel = connection.channel()

我们现在连接到本地上的的代理 - 因此是 ‘localhost‘。如果我们想连接到另一台机器上的代理,我们只需在此指定其名称或IP地址。

接下来,在发送之前,我们需要确保收件人队列存在。如果我们发送消息到不存在的位置,RabbitMQ将只删除该消息。我们来创建一个将传递消息的 hello 队列:

channel.queue_declare(queue=‘hello‘)

此时我们准备发送消息。我们的第一条消息将只包含一个字符串 "Hello World!"我们想把它发送给我们的 hello 队列。

在RabbitMQ中,消息永远不会直接发送到队列,它总是需要经过交换。我们现在需要知道的是如何使用由空字符串标识的默认交换。这种交换是特殊的 - 它允许我们准确地指定消息应该到达哪个队列。队列名称需要在routing_key参数中指定:

channel.basic_publish(exchange=‘‘,routing_key=‘hello‘,body=‘Hello World!‘)
print(" [x] Sent ‘Hello World!‘")

在退出程序之前,我们需要确保网络缓冲区被刷新,并且我们的消息被实际传送到RabbitMQ。我们可以通过轻轻关闭连接来完成。

connection.close()

接收

 技术分享图片

 

我们的第二个程序 receive.py 将接收队列中的消息并将它们打印在屏幕上。

再次,我们首先需要连接到RabbitMQ服务器。负责连接到Rabbit的代码与以前相同。

下一步,就像以前一样,要确保队列存在。使用queue_declare创建一个队列是幂等的 - 我们可以根据需要多次运行该命令,并且只会创建一个。

channel.queue_declare()

您可能会问为什么我们再次声明队列 - 我们已经在之前的代码中声明了它。如果我们确信队列已经存在,我们可以避免这种情况。例如,如果 send.py 程序之前运行过。但我们还不确定首先运行哪个程序。在这种情况下,重复在两个程序中重复声明队列是一种很好的做法。

列出队列
您可能希望看到RabbitMQ有什么队列以及它们中有多少条消息。您可以使用rabbitmqctl工具(作为特权用户)执行此操作:
> sudo rabbitmqctl list_queues
在Windows上,省略sudo:
> rabbitmqctl.bat list_queues

从队列接收消息更为复杂。它通过向队列订阅 回调函数 来工作。每当我们收到一条消息,这个回调函数就被皮卡库调用。在我们的例子中,这个函数会在屏幕上打印消息的内容。

def callback(ch, method, propertites, body):
print(" [x] Received {}".format(body))

接下来,我们需要告诉RabbitMQ这个特定的回调函数应该从我们的hello队列接收消息:

channel.basic_consume(callable, queue=‘hello‘, no_ack=True)

为了让这个命令成功,我们必须确保我们想要订阅的队列存在。幸运的是,我们对此有信心 - 我们已经使用queue_declare创建了一个队列。

NO_ACK参数,后面(几篇之后)会有解释。

最后,我们进入一个永无止境的循环,等待数据并在必要时运行回调。

print(‘ [*] Waiting for messages. To exit press CTRL+C‘)
channel.start_consuming()

把它放在一起

send.py的完整代码:

#!/usr/bin/env python
import pika
cOnnection= pika.BlockingConnection(pika.ConnectionParameters(host=‘localhost‘))
channel = connection.channel()
channel.queue_declare(queue=‘hello‘)
channel.basic_publish(exchange=‘‘, routing_key=‘hello‘,body=‘Hello World!‘)
print(" [x] Sent ‘Hello World!‘")
connection.close()

receive.py的完整代码:

#!/usr/bin/env python
import pika
cOnnection= pika.BlockingConnection(pika.ConnectionParameters(‘localhost‘))
channel = connection.channel()
channel.queue_declare(queue=‘hello‘)
def callback(ch, method, propertites, body):
print(" [x] Received {}".format(body))
channel.basic_consume(callable,queue=‘hello‘,no_ack=True)
print(‘ [*] Waiting for messages. To exit press CTRL+C‘)
channel.start_consuming()

现在我们可以在终端上试用我们的程序。首先,让我们开始一个消费者,它将持续运行等待交付:

python receive.py
# => [*] Waiting for messages. To exit press CTRL+C
# => [x] Received ‘Hello World!‘

现在开始制作。生产者计划将在每次运行后停止:

python send.py
# => [x] Sent ‘Hello World!‘

欢呼!我们能够通过RabbitMQ发送我们的第一条消息。正如您可能已经注意到的,receive.py 程序不会退出。它会随时准备接收更多消息,并可能会被Ctrl-C中断。

尝试在新终端中再次运行 send.py

 技术分享图片

 


推荐阅读
  • 本文档介绍了如何使用ESP32开发板在STA模式下实现与TCP服务器的通信,包括环境搭建、代码解析及实验步骤。 ... [详细]
  • 本文详细介绍了Oracle 11g中的创建表空间的方法,以及如何设置客户端和服务端的基本配置,包括用户管理、环境变量配置等。 ... [详细]
  • 在 Ubuntu 22.04 LTS 上部署 Jira 敏捷项目管理工具
    Jira 敏捷项目管理工具专为软件开发团队设计,旨在以高效、有序的方式管理项目、问题和任务。该工具提供了灵活且可定制的工作流程,能够根据项目需求进行调整。本文将详细介绍如何在 Ubuntu 22.04 LTS 上安装和配置 Jira。 ... [详细]
  • mysql 授权!!
    为什么80%的码农都做不了架构师?MySQL的权限系统围绕着两个概念:认证-确定用户是否允许连接数据库服务器授权-确定用户是否拥有足够的权限执 ... [详细]
  • 本文介绍了如何使用Node.js通过两种不同的方法连接MongoDB数据库,包括使用MongoClient对象和连接字符串的方法。每种方法都有其特点和适用场景,适合不同需求的开发者。 ... [详细]
  • 本文探讨了如何在PHP与MySQL环境中实现高效的分页查询,包括基本的分页实现、性能优化技巧以及高级的分页策略。 ... [详细]
  • Docker安全策略与管理
    本文探讨了Docker的安全挑战、核心安全特性及其管理策略,旨在帮助读者深入理解Docker安全机制,并提供实用的安全管理建议。 ... [详细]
  • Web动态服务器Python基本实现
    Web动态服务器Python基本实现 ... [详细]
  • 本文介绍了如何利用X_CORBA实现远程对象调用,并通过多个示例程序展示了其功能与应用,包括基础的Hello World示例、文件传输工具以及一个完整的聊天系统。 ... [详细]
  • 本文介绍了实时流协议(RTSP)的基本概念、组成部分及其与RTCP的交互过程,详细解析了客户端请求格式、服务器响应格式、常用方法分类及协议流程,并提供了SDP格式的深入解析。 ... [详细]
  • 本文详细介绍了Elasticsearch中的分页查询机制,包括基本的分页查询流程、'from-size'浅分页与'scroll'深分页的区别及应用场景,以及两者在性能上的对比。 ... [详细]
  • 本文探讨了 Java 中 net.minecraft.client.multiplayer.PlayerControllerMP 类下的 getCurrentGameType() 方法的详细使用方法,并提供了多个实际应用的代码示例。 ... [详细]
  • C# 中创建和执行存储过程的方法
    本文详细介绍了如何使用 C# 创建和调用 SQL Server 存储过程,包括连接数据库、定义命令类型、设置参数等步骤。 ... [详细]
  • 如何在U8系统中连接服务器并获取数据
    本文介绍了如何在U8系统中通过不同的方法连接服务器并获取数据,包括使用MySQL客户端连接实例的方法,如非SSL连接和SSL连接,并提供了详细的步骤和注意事项。 ... [详细]
  • 网络安全实验:Telnet与SSH服务对比及抓包分析
    本实验旨在对比Telnet和SSH两种安全通信协议的服务差异,并通过搭建服务器和使用Wireshark抓包工具进行详细分析。 ... [详细]
author-avatar
君奇均政29
这个家伙很懒,什么也没留下!
PHP1.CN | 中国最专业的PHP中文社区 | DevBox开发工具箱 | json解析格式化 |PHP资讯 | PHP教程 | 数据库技术 | 服务器技术 | 前端开发技术 | PHP框架 | 开发工具 | 在线工具
Copyright © 1998 - 2020 PHP1.CN. All Rights Reserved | 京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有