Rabbitmqpython连接rabbitmq

创建用户

&＃96;&＃96;&＃96;sript
rabbitmqctl add_user admin 123123
rabbitmqctl set_user_tags admin administrator
rabbitmqctl set_permissions -p &＃39;/&＃39; admin &＃39;.&＃39; &＃39;.&＃39; &＃39;.&＃39;
&＃96;&＃96;&＃96;

简单示例

send.py

#!/usr/bin/env python
import pikacredentials &＃61; pika.PlainCredentials(&＃39;admin&＃39;,&＃39;123123&＃39;)
connection &＃61; pika.BlockingConnection(pika.ConnectionParameters(&＃39;192.168.49.135&＃39;,5673,&＃39;/&＃39;,credentials))
channel &＃61; connection.channel()channel.queue_declare(queue&＃61;&＃39;hello&＃39;)channel.basic_publish(exchange&＃61;&＃39;&＃39;, routing_key&＃61;&＃39;hello&＃39;, body&＃61;&＃39;Hello World!&＃39;)
print(" [x] Sent &＃39;Hello World!&＃39;")
connection.close()


receive.py

#!/usr/bin/env python
import pikacredentials &＃61; pika.PlainCredentials(&＃39;admin&＃39;,&＃39;123123&＃39;)
connection &＃61; pika.BlockingConnection(pika.ConnectionParameters(&＃39;192.168.49.135&＃39;,5673,&＃39;/&＃39;,credentials))
channel &＃61; connection.channel()channel.queue_declare(queue&＃61;&＃39;hello&＃39;)def callback(ch, method, properties, body):print(" [x] Received %r" % body)channel.basic_consume(queue&＃61;&＃39;hello&＃39;, on_message_callback&＃61;callback, auto_ack&＃61;True)print(&＃39; [*] Waiting for messages. To exit press CTRL&＃43;C&＃39;)
channel.start_consuming()


一个生产者&＃xff0c;两个消费者轮询消费

send.py

#!/usr/bin/env python
import pika
import syscredentials &＃61; pika.PlainCredentials(&＃39;admin&＃39;,&＃39;123123&＃39;)
connection &＃61; pika.BlockingConnection(pika.ConnectionParameters(&＃39;192.168.49.135&＃39;,5673,&＃39;/&＃39;,credentials))
channel &＃61; connection.channel()channel.queue_declare(queue&＃61;&＃39;hello&＃39;)message &＃61; &＃39; &＃39;.join(sys.argv[1:]) or "Hello World!"channel.basic_publish(exchange&＃61;&＃39;&＃39;,routing_key&＃61;&＃39;hello&＃39;,body&＃61;message)
print(" [x] Sent %r" % message)
connection.close()


receive.py 两个消费端相同

#!/usr/bin/env python
import pika
import timecredentials &＃61; pika.PlainCredentials(&＃39;admin&＃39;,&＃39;123123&＃39;)
connection &＃61; pika.BlockingConnection(pika.ConnectionParameters(&＃39;192.168.49.135&＃39;,5673,&＃39;/&＃39;,credentials))
channel &＃61; connection.channel()channel.queue_declare(queue&＃61;&＃39;hello&＃39;)def callback(ch, method, properties, body):print(" [x] Received %r" % body)time.sleep(body.count(b&＃39;.&＃39;))print(" [x] Done")channel.basic_consume(queue&＃61;&＃39;hello&＃39;, on_message_callback&＃61;callback, auto_ack&＃61;True)print(&＃39; [*] Waiting for messages. To exit press CTRL&＃43;C&＃39;)
channel.start_consuming()


结果

发送&＃xff1a;
[root&＃64;node1 3]# python send.py 1[x] Sent &＃39;1&＃39;
[root&＃64;node1 3]# python send.py 2[x] Sent &＃39;2&＃39;
[root&＃64;node1 3]# python send.py 3[x] Sent &＃39;3&＃39;
[root&＃64;node1 3]# python send.py 4[x] Sent &＃39;4&＃39;receive1&＃xff1a;[x] Received &＃39;2&＃39;[x] Done[x] Received &＃39;4&＃39;[x] Donereceive2&＃xff1a;[x] Received &＃39;1&＃39;[x] Done[x] Received &＃39;3&＃39;[x] Done


消息确认

确认消息必须在接收方的channel上回复&＃xff0c;使用不同的channel将导致异常

send.py

#!/usr/bin/env python
import pika
import syscredentials &＃61; pika.PlainCredentials(&＃39;admin&＃39;,&＃39;123123&＃39;)
connection &＃61; pika.BlockingConnection(pika.ConnectionParameters(&＃39;192.168.49.135&＃39;,5673,&＃39;/&＃39;,credentials))
channel &＃61; connection.channel()channel.queue_declare(queue&＃61;&＃39;hello&＃39;)message &＃61; &＃39; &＃39;.join(sys.argv[1:]) or "Hello World!"channel.basic_publish(exchange&＃61;&＃39;&＃39;,routing_key&＃61;&＃39;hello&＃39;,body&＃61;message)
print(" [x] Sent %r" % message)
connection.close()

receive.py

#coding: utf-8
#!/usr/bin/env python
import pika
import timecredentials &＃61; pika.PlainCredentials(&＃39;admin&＃39;,&＃39;123123&＃39;)
connection &＃61; pika.BlockingConnection(pika.ConnectionParameters(&＃39;192.168.49.135&＃39;,5673,&＃39;/&＃39;,credentials))
channel &＃61; connection.channel()channel.queue_declare(queue&＃61;&＃39;hello&＃39;)def callback(ch, method, properties, body):print(" [x] Received %r" % body)time.sleep(10) #模拟处理消息print(" [x] Done")ch.basic_ack(delivery_tag &＃61; method.delivery_tag) #手动确认消息处理完毕channel.basic_consume(queue&＃61;&＃39;hello&＃39;, on_message_callback&＃61;callback)#queue&＃61;&＃39;hello&＃39;, on_message_callback&＃61;callback, auto_ack&＃61;True) #auto_ack&＃61;True 表示消费完以后主动把状态通知rabbitmq 默认是Falseprint(&＃39; [*] Waiting for messages. To exit press CTRL&＃43;C&＃39;)
channel.start_consuming()


查看当前未被确认的消息rabbitmqctl list_queues name messages_ready messages_unacknowledged

消息持久化

虽然&＃xff0c;已经做到了消费端挂掉后消息不丢失&＃xff0c;但当rabbitmq重启后&＃xff0c;会丢失队列和消息&＃xff0c;导致消息丢失。除非显示声明。
确保消息不丢失需要做两件事&＃xff1a;

• 队列持久化

  channel.queue_declare(queue&＃61;&＃39;hello&＃39;, durable&＃61;True)

• 消息持久化
  python channel.basic_publish(exchange&＃61;&＃39;&＃39;, routing_key&＃61;"task_queue", body&＃61;message, properties&＃61;pika.BasicProperties( delivery_mode &＃61; 2, # make message persistent ))

将消息持久化也并不能完全保证消息不丢失。&＃xff08;消息有可能保存到系统缓存&＃xff0c;但还未同步到磁盘&＃xff09;
如果需要更强的保证可以使用publisher confirm

rabbitmq 对于已经存在的队列可以重复声明&＃xff0c;但需要确保参数相同

公平调度

公平调度。在一个消费者未处理完一个消息之前不要分发新的消息给它&＃xff0c;而是将这个新消息分发给另一个不是很忙的消费者进行处理。为了解决这个问题我们可以在消费者代码中使用 channel.basic.qos ( prefetch_count &＃61; 1 )&＃xff0c;将消费者设置为公平调度。

通过 BasicQos 方法设置prefetchCount &＃61; 3。这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理3个Message。换句话说&＃xff0c;在接收到该Consumer的ack前&＃xff0c;他它不会将新的Message分发给它。

send.py

#!/usr/bin/env python
import pika
import sysconnection &＃61; pika.BlockingConnection(pika.ConnectionParameters(host&＃61;&＃39;localhost&＃39;))
channel &＃61; connection.channel()channel.queue_declare(queue&＃61;&＃39;task_queue&＃39;, durable&＃61;True)message &＃61; &＃39; &＃39;.join(sys.argv[1:]) or "Hello World!"
channel.basic_publish(exchange&＃61;&＃39;&＃39;,routing_key&＃61;&＃39;task_queue&＃39;,body&＃61;message,properties&＃61;pika.BasicProperties(delivery_mode&＃61;2, # make message persistent))
print(" [x] Sent %r" % message)
connection.close()

receive.py

#!/usr/bin/env python
import pika
import timeconnection &＃61; pika.BlockingConnection(pika.ConnectionParameters(host&＃61;&＃39;localhost&＃39;))
channel &＃61; connection.channel()channel.queue_declare(queue&＃61;&＃39;task_queue&＃39;, durable&＃61;True)
print(&＃39; [*] Waiting for messages. To exit press CTRL&＃43;C&＃39;)def callback(ch, method, properties, body):print(" [x] Received %r" % body)time.sleep(body.count(b&＃39;.&＃39;))print(" [x] Done")ch.basic_ack(delivery_tag&＃61;method.delivery_tag)channel.basic_qos(prefetch_count&＃61;1)
channel.basic_consume(queue&＃61;&＃39;task_queue&＃39;, on_message_callback&＃61;callback)channel.start_consuming()

exchange类型

• fanout
  • 把发送到该交换器的消息路由到所有与该交换器绑定的队列中
• direct
  • 把消息路由到RoutingKey与BindingKey完全匹配的队列中
• topic
  • 与direct相似&＃xff0c;不过topic类型可以进行类似正则匹配
• headers
  • 不依赖于路由键的匹配规则来路由消息&＃xff0c;而是根据发送的消息内容中的headers属性进行匹配&＃xff08;性能差&＃xff0c;用的少&＃xff09;

fanout发布和订阅

发布/订阅&＃xff1a;向多个消费者传递同一条消息

send.py

#coding: utf-8
#!/usr/bin/env python
import pika
import syscredentials &＃61; pika.PlainCredentials(&＃39;admin&＃39;,&＃39;123123&＃39;)
connection &＃61; pika.BlockingConnection(pika.ConnectionParameters(&＃39;192.168.49.135&＃39;,5673,&＃39;/&＃39;,credentials))
channel &＃61; connection.channel()channel.exchange_declare(exchange&＃61;&＃39;logs&＃39;, exchange_type&＃61;&＃39;fanout&＃39;)message &＃61; &＃39; &＃39;.join(sys.argv[1:]) or "info: Hello World!"
channel.basic_publish(exchange&＃61;&＃39;logs&＃39;, routing_key&＃61;&＃39;&＃39;, body&＃61;message)
print(" [x] Sent %r" % message)
connection.close()


receive.py

#coding: utf-8
#!/usr/bin/env python
import pikacredentials &＃61; pika.PlainCredentials(&＃39;admin&＃39;,&＃39;123123&＃39;)
connection &＃61; pika.BlockingConnection(pika.ConnectionParameters(&＃39;192.168.49.135&＃39;,5673,&＃39;/&＃39;,credentials))
channel &＃61; connection.channel()channel.exchange_declare(exchange&＃61;&＃39;logs&＃39;, exchange_type&＃61;&＃39;fanout&＃39;)result &＃61; channel.queue_declare(&＃39;&＃39;, exclusive&＃61;True) #创建临时队列&＃xff0c;消费者关闭后队列将被删除
queue_name &＃61; result.method.queuechannel.queue_bind(exchange&＃61;&＃39;logs&＃39;, queue&＃61;queue_name)print(&＃39; [*] Waiting for logs. To exit press CTRL&＃43;C&＃39;)def callback(ch, method, properties, body):print(" [x] %r" % body)channel.basic_consume(queue&＃61;queue_name, on_message_callback&＃61;callback, auto_ack&＃61;True)channel.start_consuming()

direct

send.py

#!/usr/bin/env python
import pika
import syscredentials &＃61; pika.PlainCredentials(&＃39;admin&＃39;,&＃39;123123&＃39;)
connection &＃61; pika.BlockingConnection(pika.ConnectionParameters(&＃39;192.168.49.135&＃39;,5673,&＃39;/&＃39;,credentials))
channel &＃61; connection.channel()channel.exchange_declare(exchange&＃61;&＃39;direct_logs&＃39;, exchange_type&＃61;&＃39;direct&＃39;)severity &＃61; sys.argv[1] if len(sys.argv) > 1 else &＃39;info&＃39;
message &＃61; &＃39; &＃39;.join(sys.argv[2:]) or &＃39;Hello World!&＃39;
channel.basic_publish(exchange&＃61;&＃39;direct_logs&＃39;, routing_key&＃61;severity, body&＃61;message)
print(" [x] Sent %r:%r" % (severity, message))
connection.close()

receive.py

#!/usr/bin/env python
import pika
import syscredentials &＃61; pika.PlainCredentials(&＃39;admin&＃39;,&＃39;123123&＃39;)
connection &＃61; pika.BlockingConnection(pika.ConnectionParameters(&＃39;192.168.49.135&＃39;,5673,&＃39;/&＃39;,credentials))
channel &＃61; connection.channel()channel.exchange_declare(exchange&＃61;&＃39;direct_logs&＃39;, exchange_type&＃61;&＃39;direct&＃39;)result &＃61; channel.queue_declare(&＃39;&＃39;, exclusive&＃61;True)
queue_name &＃61; result.method.queueseverities &＃61; 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&＃61;&＃39;direct_logs&＃39;, queue&＃61;queue_name, routing_key&＃61;severity)print(&＃39; [*] Waiting for logs. To exit press CTRL&＃43;C&＃39;)def callback(ch, method, properties, body):print(" [x] %r:%r" % (method.routing_key, body))channel.basic_consume(queue&＃61;queue_name, on_message_callback&＃61;callback, auto_ack&＃61;True)channel.start_consuming()


使用

[root&＃64;node1 5]# python send.py error "Run. Run. Or it will explode."[root&＃64;node1 5]# python receive.py warning error
[root&＃64;node1 5]# python receive.py error

topic

send.py

#!/usr/bin/env python
import pika
import syscredentials &＃61; pika.PlainCredentials(&＃39;admin&＃39;,&＃39;123123&＃39;)
connection &＃61; pika.BlockingConnection(pika.ConnectionParameters(&＃39;192.168.49.135&＃39;,5673,&＃39;/&＃39;,credentials))
channel &＃61; connection.channel()channel.exchange_declare(exchange&＃61;&＃39;topic_logs&＃39;, exchange_type&＃61;&＃39;topic&＃39;)routing_key &＃61; sys.argv[1] if len(sys.argv) > 2 else &＃39;anonymous.info&＃39;
message &＃61; &＃39; &＃39;.join(sys.argv[2:]) or &＃39;Hello World!&＃39;
channel.basic_publish(exchange&＃61;&＃39;topic_logs&＃39;, routing_key&＃61;routing_key, body&＃61;message)
print(" [x] Sent %r:%r" % (routing_key, message))
connection.close()


receive.py

#!/usr/bin/env python
import pika
import syscredentials &＃61; pika.PlainCredentials(&＃39;admin&＃39;,&＃39;123123&＃39;)
connection &＃61; pika.BlockingConnection(pika.ConnectionParameters(&＃39;192.168.49.135&＃39;,5673,&＃39;/&＃39;,credentials))
channel &＃61; connection.channel()channel.exchange_declare(exchange&＃61;&＃39;topic_logs&＃39;, exchange_type&＃61;&＃39;topic&＃39;)result &＃61; channel.queue_declare(&＃39;&＃39;, exclusive&＃61;True)
queue_name &＃61; result.method.queuebinding_keys &＃61; 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&＃61;&＃39;topic_logs&＃39;, queue&＃61;queue_name, routing_key&＃61;binding_key)print(&＃39; [*] Waiting for logs. To exit press CTRL&＃43;C&＃39;)def callback(ch, method, properties, body):print(" [x] %r:%r" % (method.routing_key, body))channel.basic_consume(queue&＃61;queue_name, on_message_callback&＃61;callback, auto_ack&＃61;True)channel.start_consuming()


参考
https://www.itsvse.com/thread-4667-1-1.html RabbitMQ BasicQos消费者并行处理限制
https://www.cnblogs.com/ccorz/p/5710098.html Python之操作RabbitMQ
https://www.cnblogs.com/wt11/p/5970297.html Python操作rabbitmq 实践笔记
https://www.cnblogs.com/kerwinC/p/5967584.html python RabbitMQ队列使用
https://www.rabbitmq.com/documentation.html 官网