python系列之 RabbitMQ - hello world

不念不忘少年蓝@ 2022-07-29 10:40 26阅读 0赞

## **介绍** ##

RabbitMQ 是一个消息中间件，它的主要功能非常简单:  接收和转发消息。  
 你可以认为 RabbitMQ 就是一个邮局，当你将邮件投递到邮箱后你非常确认邮递员会将你邮件最终投递到你的接收者。 通过这个比喻，MQ就是一个邮箱、邮局和邮递员  
 RabbitMQ与邮局的最大不同点是 MQ不处理纸张，而是存储和转发二进制数据(消息)

RabbitMQ有一些基本术语：   
Producer（生产者）： 一个程序/应用 发送消息就是一个producer, 以下我们简称为 "P"

![Center][]

Queue（队列）：queue就是一个"邮箱"的简称，它存在于RabbitMQ内部。虽然消息在RabbitMQ和你的应用之间传输，但他们只能保存在一 个队列(queue)里面。队列是不受任何限制的约束，你可以想存多少就存多少 - 它本质上是一个无限的缓冲区。

![Center 1][]

Consumer（消费者）：类似一个接收者, 一个 Consumer 是一个等待接收消息的程序，以下我们简称 “C"

![Center 2][]

注意： 一个Producer、Consuer和中间件（MQ)不一定必须在同一台机器上，现实使用也是这样

## Hello World ##

(使用官方推荐的 pika python 客户端）

“hello world”这里并不复杂-- 我们发送一个消息、接收消息然后输出到屏幕，为了演示这个我们需要两个程序，一个发送、一个接收并打印 我们的设计如下图：

![Center 3][]

我们这里讲的RabbitMQ采用的是AMQP 0.9.1 ,一个开放的、通用的消息协议，在不同的语言中有很多的不同的RabbitMQ客户端，我们下面使用的是pika， 这个是RabbitMQ小组推荐的python客户端

## **Sending：** ##

![Center 4][]

我们的第一个程序是send.py 用来发送消息到队列，第一件事就是要建立一个RabbitMQ的连接到Server端

import pika
    
    s_connec = pika.BlockingConnection(pika.ConnectionParameters('loaclhost'))
    chan = s_connec.channel()

我们现在就建立了连接了，由于我们中间件在本机所以这里的设置的是“localhost",如果我们想连接到一个不同的服务器，只 需要简单的将”localhost" 改为 服务器的主机名或IP地址.

下一步，在发送消息之前我们要确定接收队列存在，如果我们发送一个消息到一个不存在的目的地，RabbitMQ将丢弃这条消息， 我们先创建一个消息发送的目的队列，这里命名为  
"hello"

chan.queue_declare(queue='hello')

到这点我们就可以发送一个消息了，我们的第一个消息将正式包含一个字符串“hello world"，将这个消息发送到 ”hello“队列  
在RabbitMQ中，一个消息不能直接发送到一个队列中，通常需要通过一个交换（exchange），我们在后面的部分将详细讲解 exchange  
，现在所有我们需要知道的是如何利用一个空字符串标识默认的exchange。这个一个特殊的exchange -- 它使我们能够确切地指定消息应该到哪个队列去。 这里需要在 routing\_key  
参数中指定发送的队列名：

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

在我们退出应用程序之前，我们需要确认网络缓冲区已经flush而且消息已经确认发送到了RabbitMQ中，我们可以使用下面的代码来关闭连接

s_connec.close()

## Receiving  ##

![Center 5][]

我们的第二个程序 receive.py 将从队列中接收消息并输出到屏幕 同样我们的第一步也是连接到RabbitMQ服务器，代码和服务端的一样：

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

下一步和之前一样也需要确认队列是否存在，使用 queue\_declare 创建一个队列。  
你可能会问为什么还需要在创建一次队列，我们在上面的代码中已经创建了一次了。因为我们不能确认队列是否已经存在了，如果我们的服务端还未启动，队列也就没有创建，这个时候客户端启动了，就找不到队列。为了避免这个问题就需要在客户端和服务端均创建一下，这样就能保证无论是服务端还是客户端先启动队列都存在

channel.queue_declare(queue='hello')

可以在服务器上通过执行以下命令查看队列信息 
    $ sudo rabbitmqctl list_queues

从队列接收消息要更复杂一些，它需要为队列订阅一个 callback 函数来进行接收。当我们接收一个消息后，这个 callback 函数将会被 pika函数库自动调用， 在我们的这个实例里面这个函数将用来打印接收的消息内容到屏幕

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

下一步我们需要告诉RabbitMQ这个特殊的 callback 函数需要从我们的hello队列接收消息

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

no\_ack 参数将在后面的部分讲解

最终，我们执行永不停止的循环来等待数据和调用callbacks进行数据处理

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

完整代码：

Producer： send.py

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()

Consumer：receive.py

import pika
    
    def callback(ch, method, properties, body):
        print(" [x] Received %r" % body)
    
    connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='hello')
    channel.basic_consume(callback,
                          queue='hello',
                          no_ack=True)
    print(' [*] Waiting for messages. To exit press CTRL+C')
    channel.start_consuming()

更多信息查看:http://www.rabbitmq.com/tutorials/tutorial-one-python.html

[Center]: /images/20220728/a66f89cf2f6b45c2a73abf65542eb54f.png
[Center 1]: /images/20220728/5ed7cadc27f4425293faccd03285582e.png
[Center 2]: /images/20220728/30b16d5b1145462c98b425b12589a497.png
[Center 3]: /images/20220728/d2d6712f830e4c359e7df7dbcc655a96.png
[Center 4]: /images/20220728/da2e142ebe5c48f2bcd5e441be0857a4.png
[Center 5]: /images/20220728/988a39d1acb2484ba186b1a05b368581.png