RabbitMQ -- part1[Hello World]

RabbitMQ 是一个消息中间件,它接受并转发消息。也可以认为是一个邮局,将需要投递的信件放到信箱中,确保你的收件人最终会收到此邮件。在这个过程中,RabbitMQ 充当了信箱、邮局和邮递员的角色

1. 相关概念

  • Producing 一个发送消息的程序:
producer
  • queue RabbitMQ中的一个信箱的名字。虽然消息流在RabbitMQ和应用程序之间传输,但只能存储在一个队列(queue)中。queue 仅受到主机内存与磁盘的限制,它本质就是个大的消息缓冲区。多个producers可以发送消息到一个队列中,并且多个consumers可以从一个队列中获取数据:
queue
  • Consuming 主要是等待接受消息的程序:
consumer

2. Python Client [pika]

  • 安装: pip install pika

3. Hello World

hello

上图中,"P"代表生产者(producer),"C"代表消费者(consumer),中间的box是队列(queue)--消息缓冲区。producer发送"hello"到队列中,consumer从队列中接收消息。

  • Sending
sending

首先编辑send.py程序去发送消息到队列中:

#!/usr/bin/env python3
import pika

# 链接本地的RabbitMQ服务器
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

# 创建hello队列。在发送消息之前,首先确保队列存在,如果将消息发送至一个不存在的队列,RabbitMQ将会丢弃消息
channel.queue_declare(queue='hello')

# 发送"Hello World!"消息到"hello"队列中。在RabbitMQ中,消息不能直接发送到队列中,需要通过"exchange"将消息交到队列中。
channel.basic_publish(exchange='',  # 空字符表示使用默认exchange
                      routing_key='hello',  # 指定队列名称
                      body='Hello World!')  # 要发送的消息
print(" [x] Sent 'Hello World!'")

# 在退出程序之前,需要确保网络缓冲区已刷新并且消息被成功提交到RabbitMQ,所以要关闭连接
connection.close()

注: 如果第一次执行之后,没有看到"Sent"消息,可能是中间件磁盘空间不足(默认最少需要200MB)导致拒绝接受消息,需要检查中间件日志来排查错误。

  • Receiving

然后编辑receive.py程序,从队列中接受消息并且打印到终端

#!/usr/bin/env python3
import pika

# 连接本地RabbitMQ服务器
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

# 如之前,需要确保队列存在,"queue_declare"操作是幂等的,可以执行多次,最终结果只是确保队列被创建
channel.queue_declare(queue='hello')

# 定义回调函数,当接收到消息后,此函数被Pika库调用
def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)

# 通知RabbitMQ从hello队列中接受到消息后调用callback回调函数
channel.basic_consume(callback,     # 回调函数名称
                      queue='hello',    # 从哪个队列中获取消息
                      no_ack=True)      # 不需要ACK

# 不停地循环等待接受数据并调用回调函数
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

在发送和接受消息中重复申明了队列 (channel.queue_declare(queue='hello')),是为了确保队列已存在。如果可以确定队列已经存在,则不必申明,例如在send.py中申明后,确保receiving.py在send.py之后启动,就不需要再次申明队列。

  • 运行

启动consumer:

> python3 receive.py
[*] Waiting for messages. To exit press CTRL+C
[x] Received 'Hello World!'

启动producer,producer每次执行后会自动退出

> python3 send.py
[x] Sent 'Hello World!'
  • 查看RabbitMQ队列中有多少消息

rabbitmqctl list_queues


参考文档: http://www.rabbitmq.com/tutorials/tutorial-one-python.html


最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

推荐阅读更多精彩内容

  • Spring Cloud为开发人员提供了快速构建分布式系统中一些常见模式的工具(例如配置管理,服务发现,断路器,智...
    卡卡罗2017阅读 134,904评论 18 139
  • 1. 历史 RabbitMQ是一个由erlang开发的AMQP(Advanced Message Queue )的...
    高广超阅读 6,116评论 3 51
  • 来源 RabbitMQ是用Erlang实现的一个高并发高可靠AMQP消息队列服务器。支持消息的持久化、事务、拥塞控...
    jiangmo阅读 10,399评论 2 34
  • 本文转载自http://dataunion.org/?p=9307 背景介绍Kafka简介Kafka是一种分布式的...
    Bottle丶Fish阅读 5,491评论 0 34
  • 注:这份文档是我和几个朋友学习后一起完成的。 目录 RabbitMQ 概念 exchange交换机机制什么是交换机...
    Mooner_guo阅读 33,357评论 8 97