欢迎访问IT教程网

python操作rabbitmq

Python admin 745℃ 0评论

RabbitMQ是一个在AMQP基础上完整的,可复用的企业消息系统。 MQ全称为Message Queue, 消息队列(MQ)是一种应用程序对应用程序的通信方法 安装模块 pip install pika 基于Queue实现生产者消费者模型

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import Queue
import threading
message = Queue.Queue(10)
def producer(i):
   while True:
   message.put(i)

def consumer(i):
   while True:
   msg = message.get()

for i in range(12):
   t = threading.Thread(target=producer, args=(i,))
   t.start()

for i in range(10):
   t = threading.Thread(target=consumer, args=(i,))
   t.start()

对于RabbitMQ来说,生产和消费不再针对内存里的一个Queue对象,而是某台服务器上的RabbitMQ Server实现的消息队列

#!/usr/bin/env python
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()
#!/usr/bin/env python

import pika
# ########################## 消费者 ##########################
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello')
def callback(ch, method, properties, body):
      print( [x] Received %r % body)
channel.basic_consume(callback,
                      queue='hello',
                      no_ack=True)

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

发布订阅 发布者

#!/usr/bin/env python
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: Hello World!
channel.basic_publish(exchange='logs',
 routing_key='',
 body=message)
print( [x] Sent %r % message)
connection.close()

订阅者

#!/usr/bin/env python
import pika

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

channel.exchange_declare(exchange='logs',
 type='fanout')

result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

channel.queue_bind(exchange='logs',
 queue=queue_name)

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

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

channel.basic_consume(callback,
 queue=queue_name,
 no_ack=True)

channel.start_consuming()

发送消息时明确指定某个队列并向其中发送消息,RabbitMQ还支持根据关键字发送,即:队列绑定关键字,发送者将数据根据关键字发送到消息exchange,exchange根据 关键字 判定应该将数据发送至指定队列

#!/usr/bin/env python
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()
消费者
#!/usr/bin/env python
import pika
import sys

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

channel.exchange_declare(exchange='direct_logs',
 type='direct')

severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(exchange='direct_logs',
 routing_key=severity,
 body=message)
print( [x] Sent %r:%r % (severity, message))
connection.close()

生产者

*在topic类型下,可以让队列绑定几个模糊的关键字,之后发送者将数据发送到exchange,exchange将传入”路由值“和 ”关键字“进行匹配,匹配成功,则将数据发送到指定队列。
- # 表示可以匹配 0 个 或 多个 单词
-
表示只能匹配 一个 单词**

#!/usr/bin/env python
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()

消费者
#!/usr/bin/env python
import pika
import sys

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

channel.exchange_declare(exchange='topic_logs',
 type='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()

生产者

设置密码

credentials = pika.PlainCredentials(username,password)

connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.14.47',credentials=credentials))
# 超时时间
conn.add_timeout(5, lambda: channel.stop_consuming())

本文固定链接:IT教程网 » python操作rabbitmq
本站内容除特别标注外均为原创,欢迎转载,但请保留出处!

喜欢 (0)
发表我的评论
取消评论
表情

Hi,您需要填写昵称和邮箱!

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址