首页 > 其他分享 >RabbitMQ学习笔记02:Hello World!

RabbitMQ学习笔记02:Hello World!

时间:2022-12-30 14:44:23浏览次数:62  
标签:02 队列 RabbitMQ python World Hello channel

参考资料:RabbitMQ tutorial - "Hello world!" — RabbitMQ 

 

 

前言

RabbitMQ是一个中间人,它接受和转发消息。我们可以把它想象成一个邮局:当你把邮件投入邮箱的时候,你可以确信它最终会被投递到收件人的手中。RabbitMQ就是那个邮箱、邮局和邮差。区别就在于RabbitMQ投递的是二进制的消息数据。

这里有一些术语需要说明。

发送、产生消息的程序我们称之为生产者producer,使用此图标表示

队列queue就是一个有名字的邮箱,虽然消息会在RabbitMQ和你的应用程序之间流动,但是它们可以持久保存在队列中。队列会收到主机的内存和磁盘容量的限制。多个生产者可以向一个队列发送消息,同时多个消费者consumer也可以从一个队列中消费/接收消息。

队列的图标如下

接收、消费消息的程序我们称之为消费者consumer,使用此图标表示

注意,生产者、消费者以及RabbitMQ不需要位于同一台服务器上。事实上,绝大多数情况中,它们都分别位于不同的服务器上。生产者和消费者可以是同一个程序实现。

这里我们的实验环境,生产者和消费者是独立的程序文件,但是它们以及RabbitMQ都位于同一台服务器上。

 

 

Hello World!

学习RabbitMQ,其中的生产者和消费者需要用户自己使用编程语言来实现。我本人是一名运维工程师,有过bash, awk的经验,2017年学习廖雪峰的Python学了一半,近期又学习了B站高淇老师的Java前三章(包含面向对象),对于Python和Java语言算是一知半解,大概看得懂又写不出来的水平。我自己尝试了下,即使没有编程基础,只要我们严格按照官方的教程指导,也是可以将这些代码实现的。

这部分我们使用python手写生产者和消费者,使用RabbitMQ自带的guest账户。生产者会发送一条简单的消息给到名为hello的队列,消费者从队列会收到这条消息并将其打印出来。

简单的流程图如图所示,队列就有点类似于代表了消费者去接收了这条消息。

我们的操作系统自带了python 3.9,无论是键入python又或者是python3,都是指向python3,这点可以从字符链接里面看出来。

[root@rabbitmq-01 ~]# ls -l $(which python)
lrwxrwxrwx. 1 root root 9 Dec 12 20:51 /usr/bin/python -> ./python3
[root@rabbitmq-01 ~]# ls -l $(which python3)
lrwxrwxrwx. 1 root root 9 Dec 12 20:42 /usr/bin/python3 -> python3.9

我们需要使用到 Pika 库来使得我们的python代码可以连接RabbitMQ,这个库也是官方推荐的。

python -m pip install pika --upgrade

如果没有安装pip的话,可以使用

python -m ensurepip --upgrade

接下来就可以开始正式写python代码了。

Sending

我们的第一个程序send.py会发送一条消息到队列hello中。首先我们需要建立连接。

#!/usr/bin/env python
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

连接建立完毕,在发送消息之前,我们需要确保队列hello必须存在,否则我们发送出去的消息就会被丢弃。因此我们创建队列。

channel.queue_declare(queue='hello')

此时我们可以开始发送消息了,我们计划发送消息Hello World!到队列hello中去。

在mq中,消息无法直接发送到队列中去,必须要经过exchange。目前我们还不需要对其进行展开理解,大家直到这么个概念即可,后续在讲解RabbitMQ tutorial - Publish/Subscribe — RabbitMQ的时候会涉及到。

我们现在只需要将其发送给默认的exchange,使用空字符串来识别它。它是一个特殊的exchange,允许我们指定要发送消息的队列,队列使用routing_key参数来指定。

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

这里的print只是用来告诉用户我们做了什么。

消息发送完毕了,在我们退出程序前,我们应该确保网络buffer已经被清空,我们的消息真的发出去了。这里通过优雅地关闭连接来实现。

connection.close()

发送到这里就结束了,如果消息没有发送成功的话,可以考虑从日志入手排查问题。

Receiving

第二个程序是消费者receive.py,它将会从队列中接收消息并将其打印出来。

首先我们需要连接到服务器上,这部分代码和生产者代码相同。

接下来,同样我们需要确保队列的存在性。使用queue_declare创建队列是幂等(idempotent)的,即使执行多次也只会创建一个队列。

channel.queue_declare(queue='hello')

之所以在这里重复创建队列,是为了

  • 我们可能不知道生产者和消费者程序,哪个会先运行,因此最好在两端都创建队列。
  • 创建队列的操作需要具备幂等性。

从队列中接收一个消息会更加复杂,它工作方式是订阅一个callback函数到队列上。每当队列中有消息出现的时候,callback函数都会被Pika库调用。在本案例中,callback函数会打印消息在屏幕上。

def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)

接下来我们告诉mq,这个特定的callback函数需要从队列hello接收消息。这一步需要确保队列已存在,好在我们上面的代码已经确保过了。

channel.basic_consume(queue='hello',
                      auto_ack=True,
                      on_message_callback=callback)

auto_ack参数将会在RabbitMQ tutorial - Work Queues — RabbitMQ中讲解。

接下来我们进入无限循环,在循环中我们等待队列中一旦出现新的消息,我们就会将其消费掉然后输出消息的内容。

直到用户输入Ctrl+C来停止程序。

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print('Interrupted')
        try:
            sys.exit(0)
        except SystemExit:
            os._exit(0)

Putting it all together

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, sys, os

def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
    channel = connection.channel()

    channel.queue_declare(queue='hello')

    def callback(ch, method, properties, body):
        print(" [x] Received %r" % body.decode())

    channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=True)

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

if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print('Interrupted')
        try:
            sys.exit(0)
        except SystemExit:
            os._exit(0)

把python代码文件放到服务器上,准备开始测试。

首先打开第一个终端,运行消费者程序receive.py,它会占用前端一直运行下去,一旦有消息就会打印,直到来自用户的终止指令Ctrl+C

[root@rabbitmq-01 code]# python receive.py 
 [*] Waiting for messages. To exit press CTRL+C

打开第二个终端,运行生产者程序send.py,每运行一次它都会向队列hello发送一条消息Hello World!同时输出到终端,随后它就会退出,不会占用前端资源。

[root@rabbitmq-01 code]# python send.py 
 [x] Sent 'Hello World!'

此时我们回到消费者终端,它会按照程序中说的消费掉队列中的消息并将其输出。

[root@rabbitmq-01 code]# python receive.py 
 [*] Waiting for messages. To exit press CTRL+C
 [x] Received 'Hello World!'

send.py可以反复执行。

[root@rabbitmq-01 code]# python send.py 
 [x] Sent 'Hello World!'
[root@rabbitmq-01 code]# python send.py 
 [x] Sent 'Hello World!'
[root@rabbitmq-01 code]# python send.py 
 [x] Sent 'Hello World!'
[root@rabbitmq-01 code]# python send.py 
 [x] Sent 'Hello World!'

消费者程序都可以捕获。

[root@rabbitmq-01 code]# python receive.py 
 [*] Waiting for messages. To exit press CTRL+C
 [x] Received 'Hello World!'
 [x] Received 'Hello World!'
 [x] Received 'Hello World!'
 [x] Received 'Hello World!'

我们可以通过rabbitmqctl list_queues来查看mq实例中当前存在的队列以及队列中的消息数量。

[root@rabbitmq-01 rabbitmq_server-3.11.5]# ./sbin/rabbitmqctl list_queues
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
name	messages
hello	0

因为消息在一瞬间就被消费掉了,所以我们看到的消息数量都会是0。

想要退出的话就是在消费中终端执行Ctrl+C

 [x] Received 'Hello World!'
 [x] Received 'Hello World!'
^CInterrupted

 

本片文章的内容就到此为止了,我们学会了RabbitMQ的基本概念(生产者、消费者和队列)。

标签:02,队列,RabbitMQ,python,World,Hello,channel
From: https://www.cnblogs.com/alongdidi/p/rabbitmq_tutorial_one.html

相关文章

  • 《程序与算法》课程设计(论文)指导书[2022-12-30]
    《程序与算法》课程设计(论文)指导书[2022-12-30]《程序与算法》课程设计(论文)指导书一、设计目的使学生具备理论联系实际的程序设计思想,掌握数据与结构中线性表和二叉树和......
  • 2021届华为提前批面试记录
    文章目录2020年06月09日参加华为提前批AICloud云计算部门技术面试记录(下午5:00-6:30面试官最后说时长要自己把握一下,不要太长)总体是先结合项目提问,最后手撕代码。用......
  • 2020朝花夕拾-不务正业的大学生做了什么比赛?
    朝花夕拾,旧事重提之意。恰逢年终征文,所以写作本文为年终总结,​​实为大学四年的总结,和一起成长参与各类竞赛的回忆​​。四年里和小伙伴们一起辛苦奋斗做了那么多比赛,当比......
  • NOIP2022 游记
    \[\mathcal{I\;\,COME\;\,FROM\;\,THE\;\,SKY}\]\[\mathcal{BORNE\;\,WITH\;\,THE\;\,WINDS\;\,OF\;\,FATE}\]\[\mathcal{I\;\,CAST\;\,OFF\;\,MY\;\,CAGE}\]\[\mathcal......
  • KSN2021
    我是印度尼西亚老哥!D1T1题意给定一个长度为\(N\)字符集为\(\{{\texttt?},{\textttA},{\textttB}\}\)的字符串,求将每个\(\texttt?\)替换为\(\textttA\)或\(......
  • INTENT2022--一道包含12个反调试反虚拟机操作的ctf题解
    作者:selph查看全文请去公众号:极安御信安全研究院查看原文。从一道Re题学习12种反调试反虚拟技术题目:AntiDebuggingEmporium来源:INTENTCTF2022Re这个题目很有意思,里......
  • 2022 12 15 HW
    Okay.Shallwestartordoweneedtowait?Ithinkwecanstart,yeah.Wewantedtocontinuewiththeschematicreviewactually,isthefirsttopic.Or,Idon'tk......
  • 中国各地级市绿色金融指数数据(2000-2022)
    中国各地级市绿色金融指数数据(2000-2022)中国各地级市绿色金融指数数据(2000-2022)中国各地级市绿色金融指数数据(2000-2022)最新版数据已整理为Excel格式,数据的时间区间为2000-2......
  • 1024·致敬 | 迟到的小温暖,感谢不平凡的你们
    1024是2的10次方是电脑运行、存储,甚至分辨率的进制更是技术同学们最特别的节日今年的XTransfer1024线下活动由于疫情影响延后至今天举办在短暂的居家办公后大家重新相聚在办......
  • 20220627外星人2015 R2硬盘更换记录
    外星人换硬盘太贵,自个动手丰衣足食外星人alienware15R2更换硬盘记录,参考资料为:​​​https://downloads.dell.com/manuals/all-products/esuprt_laptop/esuprt_alienwa......