首页 > 代码库 > Python开发【十一章】：RabbitMQ队列

Python开发【十一章】：RabbitMQ队列

2024-08-17 09:14:01 221人阅读

RabbitMQ队列

rabbitMQ是消息队列；想想之前的我们学过队列queue：threading queue（线程queue，多个线程之间进行数据交互）、进程queue（父进程与子进程进行交互或者同属于同一父进程下的多个子进程进行交互）；如果两个独立的程序，那么之间是不能通过queue进行交互的,这时候我们就需要一个中间代理即rabbitMQ

消息队列：

RabbitMQ
ZeroMQ
ActiveMQ
...........

原理：

技术分享

1、安装和基本使用

安装RabbitMQ服务 http://www.rabbitmq.com/install-standalone-mac.html

python安装RabbitMQ模块

pip install pika

or

easy_install pika

or

源码

https://pypi.python.org/pypi/pika

2、实现最简单的队列通信

发送端：

#!/usr/bin/env python

# -*- coding:utf-8 -*-

#-Author-Lian

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(‘localhost‘))

channel = connection.channel() #声明一个管道(管道内发消息)

channel.queue_declare(queue=‘lzl‘) #声明queue队列

channel.basic_publish(exchange=‘‘,

routing_key=‘lzl‘, #routing_key 就是queue名

body=‘Hello World!‘

)

print("Sent ‘Hello,World!‘")

connection.close() #关闭

接收端：

#!/usr/bin/env python

# -*- coding:utf-8 -*-

#-Author-Lian

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(‘localhost‘))

channel = connection.channel()

channel.queue_declare(queue=‘lzl‘)

def callback(ch,method,properties,body):

print(ch,method,properties)

#ch:<pika.adapters.blocking_connection.BlockingChannel object at 0x002E6C90> 管道内存对象地址

#methon:<Basic.Deliver([‘consumer_tag=ctag1.03d155a851b146f19cee393ff1a7ae38‘, #具体信息

# ‘delivery_tag=1‘, ‘exchange=‘, ‘redelivered=False‘, ‘routing_key=lzl‘])>

#properties:<BasicProperties>

print("Received %r"%body)

channel.basic_consume(callback, #如果收到消息，就调用callback函数处理消息

queue="lzl",

no_ack=True)

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

channel.start_consuming() #开始收消息

注：windows连linux上的rabbitMQ会出现报错，需要提供用户名密码

3、RabbitMQ消息分发轮询

先启动消息生产者，然后再分别启动3个消费者，通过生产者多发送几条消息，你会发现，这几条消息会被依次分配到各个消费者身上

技术分享

在这种模式下，RabbitMQ会默认把p发的消息公平的依次分发给各个消费者(c),跟负载均衡差不多

pubulish.py

consume.py

通过执行pubulish.py和consume.py可以实现上面的消息公平分发，那假如c1收到消息之后宕机了，会出现什么情况呢？rabbitMQ是如何处理的？现在我们模拟一下

publish.py

consume.py

在consume.py的callback函数里增加了time.sleep模拟函数处理，通过上面程序进行模拟发现，c1接收到消息后没有处理完突然宕机，消息就从队列上消失了，rabbitMQ把消息删除掉了；如果程序要求消息必须要处理完才能从队列里删除，那我们就需要对程序进行处理一下

publish.py

consume.py

通过把consume.py接收端里的no_ack=True去掉之后并在callback函数里面添加ch.basic_ack(delivery_tag = method.delivery_tag，就可以实现消息不被处理完不能在队列里清除

查看消息队列数：

技术分享

4、消息持久化

如果消息在传输过程中rabbitMQ服务器宕机了，会发现之前的消息队列就不存在了，这时我们就要用到消息持久化，消息持久化会让队列不随着服务器宕机而消失，会永久的保存下去

发送端：

#!/usr/bin/env python

# -*- coding:utf-8 -*-

#-Author-Lian

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(‘localhost‘))

channel = connection.channel() #声明一个管道(管道内发消息)

channel.queue_declare(queue=‘lzl‘,durable=True) #队列持久化

channel.basic_publish(exchange=‘‘,

routing_key=‘lzl‘, #routing_key 就是queue名

body=‘Hello World!‘,

properties=pika.BasicProperties(

delivery_mode = 2 #消息持久化

)

print("Sent ‘Hello,World!‘")

connection.close() #关闭

接收端：

#!/usr/bin/env python

# -*- coding:utf-8 -*-

#-Author-Lian

import pika,time

connection = pika.BlockingConnection(pika.ConnectionParameters(‘localhost‘))

channel = connection.channel()

channel.queue_declare(queue=‘lzl‘,durable=True)

def callback(ch,method,properties,body):

print("->>",ch,method,properties)

time.sleep(15) # 模拟处理时间

print("Received %r"%body)

ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(callback, #如果收到消息，就调用callback函数处理消息

queue="lzl",

)

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

channel.start_consuming() #开始收消息

5、消息公平分发

如果Rabbit只管按顺序把消息发到各个消费者身上，不考虑消费者负载的话，很可能出现，一个机器配置不高的消费者那里堆积了很多消息处理不完，同时配置高的消费者却一直很轻松。为解决此问题，可以在各个消费者端，配置perfetch=1,意思就是告诉RabbitMQ在我这个消费者当前消息还没处理完的时候就不要再给我发新消息了

技术分享

1	`channel.basic_qos(prefetch_count=1)`

带消息持久化+公平分发

pubulish.py

consume.py

6、Publish\Subscribe(消息发布\订阅)　

之前的例子都基本都是1对1的消息发送和接收，即消息只能发送到指定的queue里，但有些时候你想让你的消息被所有的Queue收到，类似广播的效果，这时候就要用到exchange了，

An exchange is a very simple thing. On one side it receives messages from producers and the other side it pushes them to queues. The exchange must know exactly what to do with a message it receives. Should it be appended to a particular queue? Should it be appended to many queues? Or should it get discarded. The rules for that are defined by the exchange type.

Exchange在定义的时候是有类型的，以决定到底是哪些Queue符合条件，可以接收消息

fanout: 所有bind到此exchange的queue都可以接收消息
direct: 通过routingKey和exchange决定的那个唯一的queue可以接收消息
topic:所有符合routingKey(此时可以是一个表达式)的routingKey所bind的queue可以接收消息

headers: 通过headers 来决定把消息发给哪些queue

表达式符号说明：#代表一个或多个字符，*代表任何字符

例：#.a会匹配a.a，aa.a，aaa.a等
*.a会匹配a.a，b.a，c.a等
注：使用RoutingKey为#，Exchange Type为topic的时候相当于使用fanout　

fanout接收所有广播：

技术分享

publish.py

consume.py

有选择的接收消息 direct:

RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据关键字判定应该将数据发送至指定队列

技术分享

publish.py

consume.py

更细致的消息过滤 topic:

Although using the direct exchange improved our system, it still has limitations - it can‘t do routing based on multiple criteria.

In our logging system we might want to subscribe to not only logs based on severity, but also based on the source which emitted the log. You might know this concept from the syslog unix tool, which routes logs based on both severity (info/warn/crit...) and facility (auth/cron/kern...).

That would give us a lot of flexibility - we may want to listen to just critical errors coming from ‘cron‘ but also all logs from ‘kern‘

技术分享

publish.py

consume.py

RPC（Remote procedure call ）双向通信

To illustrate how an RPC service could be used we‘re going to create a simple client class. It‘s going to expose a method named call which sends an RPC request and blocks until the answer is received:

技术分享

rpc client：

#!/usr/bin/env python

# -*- coding:utf-8 -*-

#-Author-Lian

import pika

import uuid,time

class FibonacciRpcClient(object):

def __init__(self):

self.connection = pika.BlockingConnection(pika.ConnectionParameters(

host=‘localhost‘))

self.channel = self.connection.channel()

result = self.channel.queue_declare(exclusive=True)

self.callback_queue = result.method.queue

self.channel.basic_consume(self.on_response, #只要收到消息就执行on_response

no_ack=True, #不用ack确认

queue=self.callback_queue)

def on_response(self, ch, method, props, body):

if self.corr_id == props.correlation_id: #验证码核对

self.response = body

def call(self, n):

self.response = None

self.corr_id = str(uuid.uuid4())

print(self.corr_id)

self.channel.basic_publish(exchange=‘‘,

routing_key=‘rpc_queue‘,

properties=pika.BasicProperties(

reply_to=self.callback_queue, #发送返回信息的队列name

correlation_id=self.corr_id, #发送uuid 相当于验证码

),

body=str(n))

while self.response is None:

self.connection.process_data_events() #非阻塞版的start_consuming

print("no messages")

time.sleep(0.5) #测试

return int(self.response)

fibonacci_rpc = FibonacciRpcClient() #实例化

print(" [x] Requesting fib(30)")

response = fibonacci_rpc.call(30) #执行call方法

print(" [.] Got %r" % response)

rpc server：

#!/usr/bin/env python

# -*- coding:utf-8 -*-

#-Author-Lian

import pika

import time

connection = pika.BlockingConnection(pika.ConnectionParameters(

host=‘localhost‘))

channel = connection.channel()

channel.queue_declare(queue=‘rpc_queue‘)

def fib(n):

if n == 0:

return 0

elif n == 1:

return 1

else:

return fib(n - 1) + fib(n - 2)

def on_request(ch, method, props, body):

n = int(body)

print(" [.] fib(%s)" % n)

response = fib(n)

ch.basic_publish(excha

RabbitMQ队列

nge=‘‘,

routing_key=props.reply_to, #回信息队列名

properties=pika.BasicProperties(correlation_id=

props.correlation_id),

body=str(response))

ch.basic_ack(delivery_tag=method.delivery_tag)

#channel.basic_qos(prefetch_count=1)

channel.basic_consume(on_request,

queue=‘rpc_queue‘)

print(" [x] Awaiting RPC requests")

channel.start_consuming()

Python开发【十一章】：RabbitMQ队列

声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉：投诉/举报工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。

联系
我们

首页 > 代码库 > Python开发【十一章】：RabbitMQ队列

Python开发【十一章】：RabbitMQ队列

RabbitMQ队列

看完仍有疑问？有类似问题直接问程序猿