Python에서 RabbitMQ 사용에 대한 심층적인 이해
- 高洛峰원래의
- 2017-03-15 14:16:346130검색
서문: 이번에 RabbitMQ에 대한 글을 정리하고 작성했는데, 오히려 RabbitMQ가 훨씬 어렵다고 느꼈습니다. 이 블로그에는 영어 설명이 일부 삽입되어 있지만 이해하기 어렵지는 않습니다. RabbitMQ 다운로드 및 설치는 redis&rabbitMQ 설치를 참고하세요.
rabbitMQ는 메시지 큐입니다. 이전에 큐 큐에 대해 배운 내용을 생각해 보세요. 스레딩 큐(스레드 큐, 여러 스레드 간의 데이터 상호 작용), 프로세스 큐 (상위 프로세스가 하위 프로세스와 상호 작용하거나 동일한 상위 프로세스에 속한 여러 하위 프로세스가 상호 작용함) 두 개의 독립적인 프로그램이 대기열을 통해 상호 작용할 수 없는 경우 중간 에이전트, 즉 RabbitMQ가 필요합니다.
1. 간단한 RabbitMQ 큐 통신
by 위 그림에서 볼 수 있듯이 데이터가 먼저 전송됩니다. 교환 교환기로 전송된 후 해당 대기열로 교환이 전송됩니다. pika 모듈은 Python의 RabbitMQ용 API인터페이스입니다. 수신 측에는 콜백 함수 가 있는데, 데이터를 수신하자마자 가 호출됩니다. 소비자가 메시지를 받은 후 해당 메시지는 대기열 에서 삭제됩니다. 자, 위의 지식을 이해한 후 먼저 간단한 RabbitMQ 대기열 통신을 살펴보겠습니다.
sendend:
import pika
#连上rabbitMQ
connection=pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel=connection.channel() #生成管道，在管道里跑不同的队列
#声明queue
channel.queue_declare(queue='hello1')
#n RabbitMQ a message can never be sent directly to the queue,it always needs to go through an exchange.
#向队列里发数据
channel.basic_publish(exchange='', #先把数据发给exchange交换器,exchage再发给相应队列
routing_key='hello1', #向"hello'队列发数据
body='HelloWorld!!') #发的消息
print("[x]Sent'HelloWorld!'")
connection.close()receive end:import pika
connection=pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel=connection.channel()
# You may ask why we declare the queue again ‒ we have already declared it in our previous code.
# We could avoid that if we were sure that the queue already exists. For example if send.py program
# was run before. But we're not yet sure which program to run first. In such cases it's a good
# practice to repeat declaring the queue in both programs.
channel.queue_declare(queue='hello1')#声明队列，保证程序不出错
def callback(ch,method,properties,body):
print("-->ch",ch)
print("-->method",method)
print("-->properties",properties)
print("[x] Received %r" % body) #一条消息被一个消费者接收后，该消息就从队列删除
channel.basic_consume(callback, #回调函数，一接收到消息就调用回调函数
queue='hello1',
no_ack=False) #消费完毕后向服务端发送一个确认，默认为False
print('[*] Waiting for messages.To exit press CTRL+C')
channel.start_consuming()
실행 결과:(위 코드는 제가 쓴 주석에 해당합니다. 이해가 되리라 믿습니다~)
rabbitMQ_1_send.py [x] Sent 'Hello World!' rabbitMQ_2_receive.py [*] Waiting for messages. To exit press CTRL+C -->ch <pika.adapters.blocking_connection.BlockingChannel object at 0x000000000250AEB8> -->method <Basic.Deliver(['consumer_tag=ctag1.f9533f4c8c59473c8096817670ad69d6', 'delivery_tag=1', 'exchange=', 'redelivered=False', 'routing_key=hello1'])> -->properties <BasicProperties> [x] Received b'Hello World!!'
심층 테스트 후 다음 두 가지 결과를 얻었습니다.
rabbitMQ_1_send를 먼저 실행합니다. py로 데이터를 보내는데,rabbitMQ_2_receive.py가 실행되고 있지 않습니다. 수신이 실행 중일 때 데이터를 계속 수신할 수 있는 것으로 나타났습니다.
데이터를 수신하기 위해 여러(예: 3) 클라이언트를 실행한 후 클라이언트 1이 데이터를 수신하고 클라이언트 2를 실행합니다. 데이터를 수신한 다음 송신자를 실행하고 클라이언트 3이 데이터를 수신합니다.
로드 밸런싱과 유사합니다.
2. Full English ack
위 예시를 보면 no_ack=False( 소비 완료 후 서버에 확인 보내기, 기본값은 False) 내 영어 레벨 4인데 다음 설명을 보니 글을 쓰고 싶은 기분이 든다. ack 아주 멋지네요!! 그래서 공유합니다:
소비자 중 한 명이 오랫동안 작업을 시작하면 작업을 수행하는 데 몇 초가 걸릴 수 있습니다. 현재 코드를 사용하면 RabbitMQ가 고객에게 메시지를 전달하면 즉시 메모리에서 해당 메시지가 제거됩니다. 이 경우 작업자를 죽이면 메시지가 손실됩니다. 처리 중 이 특정 작업자에게 전달되었지만 아직dl처리되지 않은 메시지도 모두 손실됩니다.
하지만 원하지 않습니다. 작업자가 사망하면 작업이 다른 작업자에게 전달되기를 바랍니다.
메시지가 절대 오지 않도록 주문하세요. 손실된 경우 RabbitMQ는 소비자로부터 응답(승인)을 다시 전송하여 특정 메시지가 수신 및 처리되었으며 RabbitMQ가 해당 메시지를 자유롭게 삭제할 수 있음을 RabbitMQ에 알립니다.
Ack를 보내지 않고 소비자가 사망하는 경우(채널이 닫히거나, 연결이 닫히거나, TCP 연결이 끊김) RabbitMQ는 메시지가 완전히 처리되지 않았음을 이해하고 다시 대기열에 넣습니다. 같은 시간에 온라인에 다른 소비자가 있으면 신속하게 다른 소비자에게 다시 전달합니다. 이렇게 하면 작업자가 때때로 사망하더라도 메시지가 손실되지 않는다는 것을 확신할 수 있습니다.
메시지 시간 초과가 없습니다. RabbitMQ는 소비자가 죽으면 메시지를 다시 전달합니다. 메시지 처리에 아주 오랜 시간이 걸리더라도 괜찮습니다.
메시지 확인은 기본적으로 켜져 있습니다. 이전 예에서는 no_ack=True 플래그를 통해 이를 명시적으로 껐습니다. 이제 작업이 완료되면 이 플래그를 제거하고 작업자로부터 적절한 승인을 보낼 때입니다.
이 코드를 사용하면 작업자를 죽이더라도 이를 확신할 수 있습니다. 메시지를 처리하는 동안 Ctrl+C를 사용하면 아무것도 손실되지 않습니다. 직원이 사망한 후 곧 확인되지 않은 모든 메시지가 다시 전달됩니다.
我把发送端 and接收端分别比生产者与消费者。生产者发送任务A ，消费者接收任务A并处理，处理完后生产者将消息队列中的任务A删除。程中突然宕机了。而此时生产者将消息队列中的任务A删除。实际上任务A并未成功处理完，丢失了任务/消息。为解决这个问题，应使消费者接收任务并成功处理完后发送一个ack到生产者!生产者收到ack后就明白任务A已被成功处理，这时才从消息队列中将任务A删除，如果没有收到ack，就需要把任务A发送给下一个消费者，直到任务A被成功处理。
三、消息持久化
前面已经知道，生产者生产数据，消费者再启动是可以接收数据的.
但是，生产者生产数据，然后重启rabbitMQ，消费者是无法接收数据。
예:消息在传输过程中rabbitMQ服务器宕机了，会发现之前的消息队列就不存在了，这时我们就要用到消息持久化，消息持久化会让队列不随着服务器宕机而消失，会永久的保存下去。下面看下关于消息持久化的英文讲解:
우리는 이를 확인하는 방법을 배웠습니다. 소비자가 사망하더라도 작업은 손실되지 않습니다(기본적으로 비활성화하려면 no_ack=True 사용). 그러나 RabbitMQ 서버top
RabbitMQ가 종료되거나 충돌하면 사용자가 알려주지 않는 한 대기열과 메시지를 잊어버리게 됩니다. 그렇지 않아요. 메시지가 손실되지 않도록 필요두 가지가 있습니다. 즉, 대기열과 메시지를 모두 내구성으로 표시해야 합니다.
먼저 RabbitMQ가 대기열을 잃지 않도록 해야 합니다. 그러기 위해서는 내구성이 있음을 선언해야 합니다:
1 channel.queue_declare(queue='hello', Durable=True)
이 명령 자체는 정확하지만 설정에서는 작동하지 않습니다. 그 이유는 내구성이 없는 hello라는 대기열을 이미 정의했기 때문입니다. RabbitMQ에서는 다른 매개변수를 사용하여 기존 대기열을 재정의하는 것을 허용하지 않으며 이를 시도하는 모든 프로그램에 오류(会曝错)를 반환합니다. 하지만 빠른 해결 방법이 있습니다. 예를 들어 task_queue와 같이 다른 이름으로 대기열을 선언해 보겠습니다.
1channel.queue_declare(queue='task_queue', Durable=True)
이 queue_declare 변경 사항은 생산자 코드와 소비자 코드 모두에 적용되어야 합니다.
At that point we're sure that the task_queue queue won't be lost even if RabbitMQ restarts. Now we need to mark our messages as persistent - by supplying a delivery_mode property with a value 2.
1 channel.basic_publish(exchange='',
2 routing_key="task_queue",
3 body=message,
4 properties=pika.BasicProperties(
5 delivery_mode = 2, # make message persistent
6 ))
上面的英文对消息持久化讲得很好。消息持久化分为两步:
持久化队列。通过代码实现持久化hello队列:channel.queue_declare(queue='hello', durable=True)
持久化队列中的消息。通过代码实现:properties=pika.BasicProperties( delivery_mode = 2, )
这里有个点要注意下:
如果你在代码中已实现持久化hello队列与队列中的消息。那么你重启rabbitMQ后再次运行代码可能会爆错!
因为: RabbitMQ doesn't allow you to redefine an existing queue with different parameters and will return an error.
为了解决这个问题，可以声明一个与重启rabbitMQ之前不同的队列名(queue_name).
四、消息公平分发
如果Rabbit只管按顺序把消息发到各个消费者身上，不考虑消费者负载的话，很可能出现，一个机器配置不高的消费者那里堆积了很多消息处理不完，同时配置高的消费者却一直很轻松。为解决此问题，可以在各个消费者端，配置perfetch=1,意思就是告诉RabbitMQ在我这个消费者当前消息还没处理完的时候就不要再给我发新消息了。
带消息持久化+公平分发的完整代码
生产者端:
import pika
import sys
connection =pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True) #队列持久化
message = ' '.join(sys.argv[1:]) or"Hello World!"
channel.basic_publish(exchange='',
routing_key='task_queue',
body=message,
properties=pika.BasicProperties(
delivery_mode = 2, # make message persistent消息持久化
))
print(" [x] Sent %r" % message)
connection.close()消费者端:
#!/usr/bin/env python
import pika
import time
connection =pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
time.sleep(body.count(b'.'))
print(" [x] Done")
ch.basic_ack(delivery_tag =method.delivery_tag)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback,
queue='task_queue')
channel.start_consuming()我在运行上面程序时对消费者端里回调函数的一句代码(ch.basic_ack(delivery_tag =method.delivery_tag))十分困惑。这句代码去掉消费者端也能照样收到消息啊。这句代码有毛线用处??
生产者端消息持久后,需要在消费者端加上(ch.basic_ack(delivery_tag =method.delivery_tag)): 保证消息被消费后，消费端发送一个ack，然后服务端从队列删除该消息.
五、消息发布与订阅
之前的例子都基本都是1对1的消息发送和接收，即消息只能发送到指定的queue里，但有些时候你想让你的消息被所有的queue收到，类似广播的效果，这时候就要用到exchange了。PS:有兴趣的了解redis的发布与订阅，可以看看我写的博客python之redis。
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可以接收消息
表达式符号说明： #代表一个或多个字符，*代表任何字符
例：#.a会匹配a.a，aa.a，aaa.a等
*.a会匹配a.a，b.a，c.a等
注：使用RoutingKey为#，Exchange Type为topic的时候相当于使用fanout
下面我分别讲下fanout,direct,topic:
1、fanout
fanout: 所有bind到此exchange的queue都可以接收消息
send端:
import pika
import sys
connection=pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel=connection.channel()
channel.exchange_declare(exchange='logs',
type='fanout')
message=''.join(sys.argv[1:])or"info:HelloWorld!"
channel.basic_publish(exchange='logs',
routing_key='', #fanout的话为空(默认)
body=message)
print("[x]Sent%r"%message)
connection.close()receive端:
import pika
connection=pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel=connection.channel()
channel.exchange_declare(exchange='logs',type='fanout')
#不指定queue名字(为了收广播),rabbit会随机分配一个queue名字,
#exclusive=True会在使用此queue的消费者断开后,自动将queue删除
result=channel.queue_declare(exclusive=True)
queue_name=result.method.queue
#把声明的queue绑定到交换器exchange上
channel.queue_bind(exchange='logs',
queue=queue_name)
print('[*]Waitingforlogs.ToexitpressCTRL+C')
def callback(ch,method,properties,body):
print("[x]%r"%body)
channel.basic_consume(callback,
queue=queue_name,
no_ack=True)
channel.start_consuming()有两个点要注意下:
fanout-广播，send端的routing_key='', #fanout的话为空(默认)
receive端有一句代码:result=channel.queue_declare(exclusive=True)，作用:不指定queue名字(为了收广播),rabbitMQ会随机分配一个queue名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除。
2、有选择的接收消息(exchange type=direct)
RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据 关键字 判定应该将数据发送至指定队列。
send端:
import pika
import sys
connection =pika.BlockingConnection(pika.ConnectionParameters(
host='localh'))ost
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs',
type='direct')
severity = sys.argv[1] iflen(sys.argv) > 1 else 'info'
message = ' '.join(sys.argv[2:]) or'Hello World!'
channel.basic_publish(exchange='direct_logs',
routing_key=severity, #关键字不为空，告知消息发送到哪里(info,error~)
body=message)
print(" [x] Sent %r:%r" % (severity, message))
connection.close()receive端:
import pika
import sys
connection =pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs',
type='direct')
result =channel.queue_declare(exclusive=True)
queue_name = result.method.queue
severities = sys.argv[1:]
if not severities:
sys.stderr.write("Usage: %s [info] [warning] [error]\n" %sys.argv[0])
sys.exit(1)
for severity in severities:
channel.queue_bind(exchange='direct_logs',
queue=queue_name,
routing_key=severity)
print(' [*] Waiting for logs. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(" [x] %r:%r" %(method.routing_key, body))
channel.basic_consume(callback,
queue=queue_name,
no_ack=True)
channel.start_consuming()其实最开始我看代码是一脸懵逼的~ 下面是我在cmd进行测试的截图(配合着截图看会容易理解些)，一个send端，两个receive端(先起receive端，再起receive端):
send端:
receive端-1:
receive端-2:
3、更细致的消息过滤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'.
感觉我英文水平不高啊~，我对照着垃圾有道翻译，加上自己的理解，大概知道上面在讲什么。
举例: 如果是系统的错误，就把信息发送到A，如果是ＭySQL的错误，就把信息发送到B。但是对B来说，想实现接收MySQL的错误信息，可以用有选择的接收消息(exchange type=direct)，让关键字为error就实现了啊!现在B有个需求:不是所有的错误信息都接收，只接收指定的错误。在某种信息再进行过滤，这就是更细致的消息过滤topic。
send端:
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs',
type='topic') #类型为topic
routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(exchange='topic_logs',
routing_key=routing_key,
body=message)
print(" [x] Sent %r:%r" % (routing_key, message))
connection.close()receive端:
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs',
type='topic')
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
binding_keys = sys.argv[1:]
if not binding_keys:
sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0])
sys.exit(1)
for binding_key in binding_keys:
channel.queue_bind(exchange='topic_logs',
queue=queue_name,
routing_key=binding_key)
print(' [*] Waiting for logs. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(" [x] %r:%r" % (method.routing_key, body))
channel.basic_consume(callback,
queue=queue_name,
no_ack=True)
channel.start_consuming()六、RPC(Remote Procedure Call)
RPC的概念可看我百度的(其实就类似我之前做的FTP，我从客户端发一个指令，服务端返回相关信息):
RPC采用客户机/服务器模式。请求程序就是一个客户机，而服务提供程序就是一个服务器。首先，客户机调用进程发送一个有进程参数的调用信息到服务进程，然后等待应答信息。在服务器端，进程保持睡眠状态直到调用信息的到达为止。当一个调用信息到达，服务器获得进程参数，计算结果，发送答复信息，然后等待下一个调用信息，最后，客户端调用进程接收答复信息，获得进程结果，然后调用执行继续进行。
下面重点讲下RPC通信，我刚开始学挺难的，学完之后感觉RPC通信的思想很有启发性，代码的例子写得也很牛!!
client端发的消息被server端接收后，server端会调用callback函数，执行任务后，还需要把相应的信息发送到client，但是server如何将信息发还给client?如果有多个client连接server，server又怎么知道是要发给哪个client？？
RPC-server默认监听rpc_queue.肯定不能把要发给client端的信息发到rpc_queue吧(rpc_queue是监听client端发到server端的数据)。
合理的方案是server端另起一个queue，通过queue将信息返回给对应client。但问题又来了，queue是server端起的，故client端肯定不知道queue_name，连queue_name都不知道，client端接收毛线的数据？？
解决方法:
客户端在发送指令的同时告诉服务端:任务执行完后，数据通过某队列返回结果。客户端监听该队列就OK了。
client端:
import pika
import uuid
class FibonacciRpcClient(object):
def init(self):
self.connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
self.channel = self.connection.channel()
#随机建立一个queue，为了监听返回的结果
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,
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()) ##可用来标识指令(顺序)
self.channel.basic_publish(exchange='',
routing_key='rpc_queue', #client发送指令，发到rpc_queue
properties=pika.BasicProperties(
reply_to=self.callback_queue, #将指令执行结果返回到reply_to队列
correlation_id=self.corr_id,
),
body=str(n))
while self.response is None:
self.connection.process_data_events() #去queue接收数据(不阻塞)
return int(self.response)
fibonacci_rpc = FibonacciRpcClient()
print(" [x] Requesting fib(30)")
response = fibonacci_rpc.call(30)
print(" [.] Got %r" % response)server端:
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(exchange='', ##服务端发送返回的数据到props.reply_to队列(客户端发送指令时声明)
routing_key=props.reply_to, #correlation_id (随机数)每条指令都有随机独立的标识符
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, #一接收到消息就调用on_request
queue='rpc_queue')
print(" [x] Awaiting RPC requests")
channel.start_consuming()위 내용은 Python에서 RabbitMQ 사용에 대한 심층적인 이해의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!