python使用rabbitmq rabbitmq(中间消息代理)在python中的使用详解

import pika
connection = pika.BlockingConnection(pika.ConnectionParameters("localhost")) # 建立程序与rabbitmq的连接
channel = connection.channel()
channel.queue_declare(queue='hello') # 定义hello队列
channel.basic_publish(exchange='',
      routing_key='hello', # 告诉rabbitmq将消息发送到hello队列中
      body='Hello world!') # 发送消息的内容
print(" [x] Sent 'Hello World!'")
connection.close() # 关闭与rabbitmq的连接

import pika
import time
connection = pika.BlockingConnection(pika.ConnectionParameters("localhost")) # 建立程序与rabbitmq的连接
channel = connection.channel()
# 在接收端定义队列，参数与发送端的相同
channel.queue_declare(queue='hello')
def callback(ch, method, properties, body):
 """
 收到消息调用callback处理消息
 :param ch:
 :param method:
 :param properties:
 :param body:
 :return:
 """
 print(" [x] received %r" % body)
 # time.sleep(30)
 print("Done....")
channel.basic_consume(callback,
      queue='hello', # 告诉rabbitmq此程序从hello队列中接收消息
      no_ack=True)

# channel.basic_consume(callback,
#      queue='hello')
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming() # 开始接收，未收到消息阻塞

import pika
connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)  #使队列持久化
message = "Hello World"
channel.basic_publish(exchange='',
      routing_key='task_queue',
      body=message,
      properties=pika.BasicProperties(
       delivery_mode=2,  #使消息持久化
      ))
print(" [x] Sent %r" % message)
connection.close()

import pika
import time

connection = pika.BlockingConnection(pika.ConnectionParameters("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(2)
 print(" [x] Done")
 # 因为没有设置no_ask=True, 所以需要告诉rabbitmq消息已经处理完毕，rabbitmq将消息移出队列。
 ch.basic_ack(delivery_tag=method.delivery_tag)

#同一时间worker只接收一条消息，等这条消息处理完在接收下一条
channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback,
      queue='task_queue')


channel.start_consuming()

import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
# 申明一个exchange,两个参数分别为exchange的名字和类型；当exchang='fanout'时，所有绑定到此exchange的消费者队列都将收到消息
channel.exchange_declare(exchange='logs',
       exchange_type='fanout')
# 消息可以在命令行启动脚本时以参数的形式传入
# message = ' '.join(sys.argv[1:]) or "info: Hello World!"
message = 'Hello World!'
channel.basic_publish(exchange='logs',
      routing_key='',
      body=message)
print(" [x] Sent %r" % message)
connection.close()

import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='logs',
       exchange_type='fanout')
# 随机生成一个queue，此queue唯一，且在连接端开后自动销毁
result = channel.queue_declare(exclusive=True)
# 得到随机生成消费者队列的名字
queue_name = result.method.queue
# 将消费者队列与exchange绑定
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] received %r" % body)

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

channel.start_consuming()

import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs',
       exchange_type='direct')
# 命令行启动时，以参数的的形式传入发送消息的级别，未传怎默认设置未info
# severity = sys.argv[1] if len(sys.argv) > 2 else 'info'
# 命令行启动时，以参数的的形式传入发送消息的内容，未传怎默认设置Hello World!
# message = ' '.join(sys.argv[2:]) or 'Hello World!'
# 演示使用，实际运用应用上面的方式设置消息级别
severity = 'info' #作为例子直接将消息的级别设置为info
# severity = 'warning'
message = 'Hello World'

#使用exchang的direct模式时，routing_key的值为消息的级别
channel.basic_publish(exchange='direct_logs',
      routing_key=severity,
      body=message)
print(" [x] Sent %r:%r" % (severity, message))
connection.close()

#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs',
       exchange_type='direct')
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
# 命令行启动时以参数的形式传入要接收哪个级别的消息，可以传入多个级别
# severities = sys.argv[1:]
# 演示使用，实际运用应该用上面的方式指明消息级别
# 作为演示，直接设置两个接收级别，info 和 warning
severities = ['info', 'warning']
if not severities:
 """如果要接收消息的级别不存在则提示用户输入级别并退出程序"""
 sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])
 sys.exit(1)
for severity in severities:
 """依次为每个消息级别绑定queue"""
 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()

import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs',
       exchange_type='topic')
# 以命令行的方式启动发送端，以参数的形式传入发送消息类型的关键字
routing_key = sys.argv[1] if len(sys.argv[1]) > 2 else 'anonymous.info' 
# routing_key = 'anonymous.info'
# routing_key = 'abc.orange.abc'
# routing_key = 'abc.abc.rabbit'
# routing_key = 'lazy.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()

#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs',
       exchange_type='topic')
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
binding_keys = sys.argv[1:]
# binding_keys = '#'  #接收所有的消息
# binding_keys = ['*.info']  #接收所有以".info"结尾的消息
# binding_keys = ['*.orange.*'] #接收所有含有".orange."的消息
# binding_keys = ['*.*.rabbit', 'lazy.*'] #接收所有含有两个扩展名且结尾是".rabbit"和所有以"lazy."开头的消息
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()

import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='rpc_queue')
def fib(n):
 """
 计算斐波那契数列中第n个数的值
 :param n:
 :return:
 """
 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='',
      routing_key=props.reply_to, # 使用客户端传来的队列向客户端发送消息的处理结果
      properties=pika.BasicProperties(
       correlation_id = props.correlation_id), # 指明处理消息的id 用于客户端确认
      body=str(response))
 ch.basic_ack(delivery_tag = method.delivery_tag) # 未申明no_ack = True, 消息处理完毕需向rabbitmq确认
channel.basic_qos(prefetch_count=1) # 每次只处理一条消息
channel.basic_consume(on_request, queue='rpc_queue')
print(" [x] Awaiting RPC requests")
channel.start_consuming() # 开始接收消息，未收到消息处于阻塞状态