1 #生产者 2 3 #!/usr/bin/env python 4 # -*- coding:utf-8 -*- 5 #-Author-soloLi 6 import pika 7 8 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) #连接到rabbitmq服务器，因为是在本地测试，所以就用localhost就可以了 9 channel = connection.channel() #声明一个管道(管道内发消息)10 channel.queue_declare(queue='solo') #声明消息队列，消息将在这个队列中进行传递。如果将消息发送到不存在的队列，rabbitmq将会自动清除这些消息。11 12 channel.basic_publish(exchange='', #exchange表示交换器，能精确指定消息应该发送到哪个队列13 routing_key='solo', #routing_key 就是queue队列的名称14 body='Hello World!' #发送内容15 )16 print("Sent 'Hello,World!'")17 connection.close() #关闭
[聚合文章] Python学习笔记整理总结【RabbitMQ队列】
消息系统
2017-11-28
19 阅读
1 #消费者 2 #!/usr/bin/env python 3 # -*- coding:utf-8 -*- 4 #-Author-soloLi 5 import pika 6 7 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) 8 channel = connection.channel() 9 channel.queue_declare(queue='solo')#如果知道queue='solo'已经存在且运行可以省略，但有可能你不知道10 11 def callback(ch,method,properties,body):12 print(ch,method,properties)13 #ch:<pika.adapters.blocking_connection.BlockingChannel object at 0x002E6C90> #管道内存对象地址14 #methon:<Basic.Deliver(['consumer_tag=ctag1.03d155a851b146f19cee393ff1a7ae38', #具体信息15 # 'delivery_tag=1', 'exchange=', 'redelivered=False', 'routing_key=solo'])>16 #properties:<BasicProperties>17 print("Received %r"%body)18 19 channel.basic_consume(callback, #如果收到消息，就调用callback函数处理消息20 queue="solo",21 no_ack=True) #接受到消息后不返回ack，无论本地是否处理完消息都会在队列中消失22 print(' [*] Waiting for messages. To exit press CTRL+C')23 channel.start_consuming() #开始接收信息，并进入阻塞状态，队列里有信息才会调用callback进行处理
2.一个生产者，多个消费者模型（一对多）
①默认生产者轮询发布消息给各个消费者,跟负载均衡差不多（消息公平分发）
1 #生产者 2 3 #!/usr/bin/env python 4 # -*- coding:utf-8 -*- 5 #-Author-soloLi 6 import pika 7 8 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) #连接到rabbitmq服务器，因为是在本地测试，所以就用localhost就可以了 9 channel = connection.channel() #声明一个管道(管道内发消息)10 channel.queue_declare(queue='solo') #声明消息队列，消息将在这个队列中进行传递。如果将消息发送到不存在的队列，rabbitmq将会自动清除这些消息。11 12 channel.basic_publish(exchange='', #exchange表示交换器，能精确指定消息应该发送到哪个队列13 routing_key='solo', #routing_key 就是queue队列的名称14 body='Hello World!' #发送内容15 )16 print("Sent 'Hello,World!'")17 connection.close() #关闭
1 #消费者1 2 #!/usr/bin/env python 3 # -*- coding:utf-8 -*- 4 #-Author-soloLi 5 import pika 6 7 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) 8 channel = connection.channel() 9 channel.queue_declare(queue='solo')#如果知道queue='solo'已经存在且运行可以省略，但有可能你不知道10 11 def callback(ch,method,properties,body):12 print(ch,method,properties)13 #ch:<pika.adapters.blocking_connection.BlockingChannel object at 0x002E6C90> #管道内存对象地址14 #methon:<Basic.Deliver(['consumer_tag=ctag1.03d155a851b146f19cee393ff1a7ae38', #具体信息15 # 'delivery_tag=1', 'exchange=', 'redelivered=False', 'routing_key=solo'])>16 #properties:<BasicProperties>17 print("Received %r"%body)18 19 channel.basic_consume(callback, #如果收到消息，就调用callback函数处理消息20 queue="solo",21 no_ack=True) #接受到消息后不返回ack，无论本地是否处理完消息都会在队列中消失22 print(' [*] Waiting for messages. To exit press CTRL+C')23 channel.start_consuming() #开始接收信息，并进入阻塞状态，队列里有信息才会调用callback进行处理
1 #消费者2 2 #!/usr/bin/env python 3 # -*- coding:utf-8 -*- 4 #-Author-soloLi 5 import pika 6 7 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) 8 channel = connection.channel() 9 channel.queue_declare(queue='solo') #如果知道queue='solo'已经存在且运行可以省略，但有可能你不知道10 11 def callback(ch,method,properties,body):12 print(ch,method,properties)13 #ch:<pika.adapters.blocking_connection.BlockingChannel object at 0x002E6C90> #管道内存对象地址14 #methon:<Basic.Deliver(['consumer_tag=ctag1.03d155a851b146f19cee393ff1a7ae38', #具体信息15 # 'delivery_tag=1', 'exchange=', 'redelivered=False', 'routing_key=solo'])>16 #properties:<BasicProperties>17 print("Received %r"%body)18 19 channel.basic_consume(callback, #如果收到消息，就调用callback函数处理消息20 queue="solo",21 no_ack=True) #接受到消息后不返回ack，无论本地是否处理完消息都会在队列中消失22 print(' [*] Waiting for messages. To exit press CTRL+C')23 channel.start_consuming() #开始接收信息，并进入阻塞状态，队列里有信息才会调用callback进行处理
②如果Rabbit只管按顺序把消息发到各个消费者身上，不考虑消费者负载的话，很可能出现，一个机器配置不高的消费者那里堆积了很多消息处理不完，同时配置高的消费者却一直很轻松。
解决方案：可以在各个消费者端，配置perfetch=1,意思就是告诉RabbitMQ在我这个消费者当前消息还没处理完的时候就不要再给我发新消息了
1 #channel.basic_qos(prefetch_count=1)
③消费者接收到消息后没有处理完突然宕机，消息就从队列上消失了，rabbitMQ把消息删除掉了
解决方案：消费者里的no_ack=True去掉之后并在callback函数里面添加ch.basic_ack(delivery_tag = method.delivery_tag，就可以实现消息不被处理完不能在队列里清除。
④如果消息在传输过程中rabbitMQ服务器宕机了，会发现之前的消息队列就不存在了
解决方案：消息持久化，消息持久化会让队列不随着服务器宕机而消失，会永久的保存下去
1 #channel.queue_declare(queue='solo',durable=True) //队列持久化2 3 #channel.basic_publish(4 # properties=pika.BasicProperties(5 # delivery_mode = 2 //消息持久化6 # )
⑤公平分发+消息持久化
1 #生产者 2 3 #!/usr/bin/env python 4 # -*- coding:utf-8 -*- 5 #-Author-soloLi 6 import pika 7 8 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) #连接到rabbitmq服务器，因为是在本地测试，所以就用localhost就可以了 9 channel = connection.channel() #声明一个管道(管道内发消息)10 channel.queue_declare(queue='solo',durable=True) #声明消息队列，消息将在这个队列中进行传递。如果将消息发送到不存在的队列，rabbitmq将会自动清除这些消息。//增加队列持久化11 12 channel.basic_publish(exchange='', #exchange表示交换器，能精确指定消息应该发送到哪个队列13 routing_key='solo', #routing_key 就是queue队列的名称14 body='Hello World!' #发送内容15 properties=pika.BasicProperties(16 delivery_mode = 2 #消息持久化17 )18 )19 print("Sent 'Hello,World!'")20 connection.close() #关闭
1 #消费者1 2 #!/usr/bin/env python 3 # -*- coding:utf-8 -*- 4 #-Author-soloLi 5 import pika 6 7 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) 8 channel = connection.channel() 9 channel.queue_declare(queue='solo')#如果知道queue='solo'已经存在且运行可以省略，但有可能你不知道10 11 def callback(ch,method,properties,body):12 print(ch,method,properties)13 #ch:<pika.adapters.blocking_connection.BlockingChannel object at 0x002E6C90> #管道内存对象地址14 #methon:<Basic.Deliver(['consumer_tag=ctag1.03d155a851b146f19cee393ff1a7ae38', #具体信息15 # 'delivery_tag=1', 'exchange=', 'redelivered=False', 'routing_key=solo'])>16 #properties:<BasicProperties>17 print("Received %r"%body)18 ch.basic_ack(delivery_tag=method.delivery_tag) #④解决宕机消息未发送完19 20 channel.basic_consume(callback, #如果收到消息，就调用callback函数处理消息21 queue="solo",22 ) #接受到消息后不返回ack，无论本地是否处理完消息都会在队列中消失23 print(' [*] Waiting for messages. To exit press CTRL+C')24 channel.start_consuming() #开始接收信息，并进入阻塞状态，队列里有信息才会调用callback进行处理
1 #消费者2 2 #!/usr/bin/env python 3 # -*- coding:utf-8 -*- 4 #-Author-soloLi 5 import pika 6 7 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost')) 8 channel = connection.channel() 9 channel.queue_declare(queue='solo')#如果知道queue='solo'已经存在且运行可以省略，但有可能你不知道10 11 def callback(ch,method,properties,body):12 print(ch,method,properties)13 #ch:<pika.adapters.blocking_connection.BlockingChannel object at 0x002E6C90> #管道内存对象地址14 #methon:<Basic.Deliver(['consumer_tag=ctag1.03d155a851b146f19cee393ff1a7ae38', #具体信息15 # 'delivery_tag=1', 'exchange=', 'redelivered=False', 'routing_key=solo'])>16 #properties:<BasicProperties>17 print("Received %r"%body)18 ch.basic_ack(delivery_tag=method.delivery_tag) #④解决宕机消息未发送完19 20 channel.basic_consume(callback, #如果收到消息，就调用callback函数处理消息21 queue="solo",22 ) #接受到消息后不返回ack，无论本地是否处理完消息都会在队列中消失23 print(' [*] Waiting for messages. To exit press CTRL+C')24 channel.start_consuming() #开始接收信息，并进入阻塞状态，队列里有信息才会调用callback进行处理
3、Publish/Subscribe(消息发布/订阅)
接收者绑定一个频道就能接收这个频道的所有消息，但无过滤相关操作。使用RabbitMQ就能达到过滤等操作，更细致的接收消息
之前的例子都基本都是1对1的消息发送和接收，即消息只能发送到指定的queue里，但有些时候你想让你的消息被所有的Queue收到，类似广播的效果，这时候就要用到exchange了，
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接收所有广播：广播表示当前消息是实时的，如果没有一个消费者在接受消息，消息就会丢弃，在这里消费者的no_ack已经无用，因为fanout不会管你处理消息结束没有，发过的消息不会重发，记住广播是实时的。
1 #消息发布者 2 #!/usr/bin/env python 3 # -*- coding:utf-8 -*- 4 #-Author-soloLi 5 import pika 6 import sys 7 8 connection = pika.BlockingConnection(pika.ConnectionParameters( 9 host='localhost'))10 channel = connection.channel()11 12 channel.exchange_declare(exchange='logs',13 type='fanout') #fanout类型的转发器14 15 message = "info: Hello World!"16 channel.basic_publish(exchange='logs',17 routing_key='', #广播不用声明queue18 body=message)19 print(" [x] Sent %r" % message)20 connection.close()
1 #消息订阅者 2 #!/usr/bin/env python 3 # -*- coding:utf-8 -*- 4 #-Author-soloLi 5 import pika 6 7 connection = pika.BlockingConnection(pika.ConnectionParameters( 8 host='localhost')) 9 channel = connection.channel()10 channel.exchange_declare(exchange='logs',11 type='fanout') #fanout类型的转发器12 result = channel.queue_declare(exclusive=True) # 不指定queue名字,rabbit会随机分配一个名字,13 # exclusive=True会在使用此queue的消费者断开后,自动将queue删除14 queue_name = result.method.queue15 channel.queue_bind(exchange='logs', # 绑定转发器，接收转发器上面的数据16 queue=queue_name)17 18 print(' [*] Waiting for logs. To exit press CTRL+C')19 def callback(ch, method, properties, body):20 print(" [x] %r" % body)21 channel.basic_consume(callback,22 queue=queue_name,23 no_ack=True)24 channel.start_consuming()
② 有选择的接收消息 direct: 同fanout一样，no_ack在此要设置为True，不然队列里数据不会清空（虽然也不会重发）
RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据 关键字 判定应该将数据发送至指定队列
注：此方式接收所有的error、info、warning
1 #消息发布者 2 #!/usr/bin/env python 3 # -*- coding:utf-8 -*- 4 #-Author-soloLi 5 import pika 6 import sys 7 8 connection = pika.BlockingConnection(pika.ConnectionParameters( 9 host='localhost'))10 channel = connection.channel()11 12 channel.exchange_declare(exchange='direct_logs',13 type='direct') #direct类型的转发器14 15 severity = sys.argv[1] if len(sys.argv) > 1 else 'info' #严重程度,什么级别的日志//通过脚本传入参数（默认传入参数）16 message = ' '.join(sys.argv[2:]) or 'Hello World!' #发出的消息17 channel.basic_publish(exchange='direct_logs',18 routing_key=severity, #哪些队列绑定了这些级别，哪些队列就可以收该消息19 body=message)20 print(" [x] Sent %r:%r" % (severity, message))21 connection.close()
1 #消息订阅者 2 #!/usr/bin/env python 3 # -*- coding:utf-8 -*- 4 #-Author-soloLi 5 import pika 6 import sys 7 8 connection = pika.BlockingConnection(pika.ConnectionParameters( 9 host='localhost'))10 channel = connection.channel()11 12 channel.exchange_declare(exchange='direct_logs',13 type='direct')14 15 result = channel.queue_declare(exclusive=True)16 queue_name = result.method.queue17 18 severities = sys.argv[1:] #传入的所有参数19 if not severities:20 sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])21 sys.exit(1)22 23 for severity in severities: #循环severity列表24 channel.queue_bind(exchange='direct_logs', #绑定（传入几个绑定几个）25 queue=queue_name,26 routing_key=severity)27 28 print(' [*] Waiting for logs. To exit press CTRL+C')29 30 def callback(ch, method, properties, body):31 print(" [x] %r:%r" % (method.routing_key, body))32 33 channel.basic_consume(callback,34 queue=queue_name,35 no_ack=True)36 37 channel.start_consuming()
③ 更细致的消息过滤 topic（动态匹配）
所有符合routingKey(此时可以是一个表达式)的routingKey所bind的queue可以接收消息
表达式符号说明：#代表一个或多个字符，*代表任何字符
例：#.a会匹配a.a，aa.a，aaa.a等
*.a会匹配a.a，b.a，c.a等
注：使用RoutingKey为#，Exchange Type为topic的时候相当于使用fanout
注：此方式接收为分类别接收：MySQL的所有error、info、warning；阿帕奇的所有error、info、warning。。。
1 #消息发布者 2 #!/usr/bin/env python 3 # -*- coding:utf-8 -*- 4 #-Author-soloLi 5 import pika 6 import sys 7 8 connection = pika.BlockingConnection(pika.ConnectionParameters( 9 host='localhost'))10 channel = connection.channel()11 12 channel.exchange_declare(exchange='topic_logs',13 type='topic')14 15 routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'16 message = ' '.join(sys.argv[2:]) or 'Hello World!'17 channel.basic_publish(exchange='topic_logs',18 routing_key=routing_key,19 body=message)20 print(" [x] Sent %r:%r" % (routing_key, message))21 connection.close()
1 #消息订阅者 2 #!/usr/bin/env python 3 # -*- coding:utf-8 -*- 4 #-Author-soloLi 5 import pika 6 import sys 7 8 connection = pika.BlockingConnection(pika.ConnectionParameters( 9 host='localhost'))10 channel = connection.channel()11 12 channel.exchange_declare(exchange='topic_logs',13 type='topic')14 15 result = channel.queue_declare(exclusive=True)16 queue_name = result.method.queue17 18 binding_keys = sys.argv[1:]19 if not binding_keys:20 sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0])21 sys.exit(1)22 23 for binding_key in binding_keys:24 channel.queue_bind(exchange='topic_logs',25 queue=queue_name,26 routing_key=binding_key)27 28 print(' [*] Waiting for logs. To exit press CTRL+C')29 30 def callback(ch, method, properties, body):31 print(" [x] %r:%r" % (method.routing_key, body))32 33 channel.basic_consume(callback,34 queue=queue_name,35 no_ack=True)36 37 channel.start_consuming()
4、RPC（双向通信）
注：也是实时的
（一）接收端将接收到的消息处理后再返回给发送端（发送端在发送信息前，产生一个接收消息的临时队列，并且在发送请求的时候将该队列赋值给reply_to，该队列用来接收返回的结果。）；
（二）发送的消息和返回的结果一一对应起来（correlation id）
①使用python的uuid来产生唯一的correlation_id。
②发送计算请求时，设定参数correlation_id。
③定义一个字典来保存返回的数据，并且键值为相应线程产生的correlation_id。
1 #########rpc client####### 2 #!/usr/bin/env python 3 # -*- coding:utf-8 -*- 4 #-Author-soloLi 5 import pika 6 import uuid,time 7 8 9 class FibonacciRpcClient(object):10 def __init__(self):11 self.connection = pika.BlockingConnection(pika.ConnectionParameters(12 host='localhost'))13 14 self.channel = self.connection.channel()15 16 result = self.channel.queue_declare(exclusive=True) #生成一个随机的callback_queue17 self.callback_queue = result.method.queue18 19 self.channel.basic_consume(self.on_response, #只要收到消息就执行on_response回调函数20 no_ack=True, #不用ack确认21 queue=self.callback_queue)22 23 def on_response(self, ch, method, props, body):24 if self.corr_id == props.correlation_id: #验证码核对，确保发送和返回的消息一一对应25 self.response = body26 27 28 def call(self, n):29 self.response = None30 self.corr_id = str(uuid.uuid4()) 31 print(self.corr_id)32 self.channel.basic_publish(exchange='',33 routing_key='rpc_queue',34 properties=pika.BasicProperties(35 reply_to=self.callback_queue, #发送返回信息的队列name//携带新的信息reply_to，不影响发送的消息本身36 correlation_id=self.corr_id, #发送uuid 相当于验证码37 ),38 body=str(n))39 while self.response is None:40 self.connection.process_data_events() #非阻塞版的start_consuming//有消息接收消息，没有消息继续往下走41 print("no messages")42 time.sleep(0.5) #测试//可以允许客户再次发送数据，多条数据可能不按顺序完成，返回消息的结果可能不一一对应43 return int(self.response)44 45 46 fibonacci_rpc = FibonacciRpcClient() #实例化47 print(" [x] Requesting fib(30)")48 response = fibonacci_rpc.call(30) #执行call方法49 print(" [.] Got %r" % response)
1 ########rpc server########## 2 3 #!/usr/bin/env python 4 # -*- coding:utf-8 -*- 5 #-Author-soloLi 6 import pika 7 import time 8 9 connection = pika.BlockingConnection(pika.ConnectionParameters(10 host='localhost'))11 12 channel = connection.channel()13 14 channel.queue_declare(queue='rpc_queue')15 16 17 def fib(n):18 if n == 0:19 return 020 elif n == 1:21 return 122 else:23 return fib(n - 1) + fib(n - 2)24 25 26 def on_request(ch, method, props, body):27 n = int(body)28 29 print(" [.] fib(%s)" % n)30 response = fib(n)31 32 ch.basic_publish(exchange='',33 routing_key=props.reply_to, #回信息队列名34 properties=pika.BasicProperties(correlation_id= #客户端发来的id35 props.correlation_id),36 body=str(response))37 ch.basic_ack(delivery_tag=method.delivery_tag) #确认消息被消费（任务完成）38 39 40 #channel.basic_qos(prefetch_count=1)41 channel.basic_consume(on_request, #收到消息触发on_request回调函数42 queue='rpc_queue')43 44 print(" [x] Awaiting RPC requests")45 channel.start_consuming()
注：本文内容来自互联网，旨在为开发者提供分享、交流的平台。如有涉及文章版权等事宜，请你联系站长进行处理。