python RabbitMQ

RabbitMQ 消息队列

python中的Q：threading Queue :是线程直接交换

                        进程Queue :是父进程与子进程或同属于一个父进程下的多个子进程交换。

这能python自己使用。

为了和不同语言交换使用，需要使用RabbitMQ(中间管道)

RabbitMQ是流行的开源消息队列系统，用erlang语言开发，所以在安装RabbitMQ之前要安装erlang.

erlang下载：http://www.erlang.org/downloads  我的是64位

RabbitMQ下载：https://pan.baidu.com/s/1cqpG0u

安装参考：https://blog.csdn.net/lu1005287365/article/details/52315786

1.1RabbitMQ关键词说明：

Broker：消息队列服务器实体。
Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列。
Queue：消息队列载体，每个消息都会被投入到一个或多个队列。
Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来。
Routing Key：路由关键字，exchange根据这个关键字进行消息投递。
vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离。
producer：消息生产者，就是投递消息的程序。
consumer：消息消费者，就是接受消息的程序。
channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务。

 

1.2消息队列运行机制：

（1）客户端连接到消息队列服务器，打开一个channel。
（2）客户端声明一个exchange，并设置相关属性。
（3）客户端声明一个queue，并设置相关属性。
（4）客户端使用routing key，在exchange和queue之间建立好绑定关系。
（5）客户端投递消息到exchange。

（6）exchange接收到消息后，就根据消息的key和已经设置的binding，将消息投递到一个或多个队列里。

注：在声明一个队列后，如果将其持久化，则下次不需要进行声明，因为该队列已经在rabbitMQ中了！！！

         例如下面的例子中都为首次声明一个队列！！！

 

1.3exchange类型：

1.Direct交换机

特点：依据key进行投递

例如绑定时设置了routing key为”hello”，那么客户端提交的消息，只有设置了key为”hello”的才会投递到队列。

2.Topic交换机

特点：对key模式匹配后进行投递，符号”#”匹配一个或多个词，符号”*”匹配一个词

例如”abc.#”匹配”abc.def.ghi”，”abc.*”只匹配”abc.def”。

3.Fanout交换机

特点：不需要key，采取广播模式，一个消息进来时，投递到与该交换机绑定的所有队列

参考：https://www.cnblogs.com/jfl-xx/p/7324285.html

RabbitMQ支持各种语言，使用某种语言链接需要下载相应的模块，python需要pika、celery、Haigha.

http://www.cnblogs.com/alex3714/articles/5248247.html

生成者：

#!/usr/bin/env python
import pika
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
               'localhost'))
channel = connection.channel()
 
#声明queue
channel.queue_declare(queue='hello')
 
#n RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange.
channel.basic_publish(exchange='',
                      routing_key='hello',
                      body='Hello World!')
print(" [x] Sent 'Hello World!'")
connection.close()

　消费者：

#_*_coding:utf-8_*_
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='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()

消息提供者代码

import pika
import time
connection = pika.BlockingConnection(pika.ConnectionParameters(
    'localhost'))
channel = connection.channel() #声明管道
 
# 声明queue
channel.queue_declare(queue='task_queue')
 
# n RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange.
import sys
 
message = ' '.join(sys.argv[1:]) or "Hello World! %s" % time.time()
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() 

消费者代码

#_*_coding:utf-8_*_
 
import pika, time 
connection = pika.BlockingConnection(pika.ConnectionParameters(
    'localhost'))
channel = connection.channel()#声明管道 
def callback(ch, method, properties, body):  #回调函数，
    print(" [x] Received %r" % body)
    time.sleep(20)
    print(" [x] Done")
    print("method.delivery_tag",method.delivery_tag)
    ch.basic_ack(delivery_tag=method.delivery_tag) 
 
channel.basic_consume(callback, #消息来了就会触发callback函数
                      queue='task_queue',
                      no_ack=True  #如果为true的话 ，无论又误处理完斗不给服务器发消息 ，去掉的话，消费者循环消费。
 
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()    

通过命令rabbitmqct1.bat list_queues 在命令行下查看当前又多少队列及队列中的数据个数。

防止重启后消失丢失，需要添加消息持久化，在生产和消费端需要添加上 channel.queue_declare(queue='hello', durable=True)#队列持久化，消失不持久化，

消失持久化，在生产端的channel.basic_publish中添加

properties=pika.BasicProperties(

                          delivery_mode=2,  # make message persistent
                      )

消息公平分发

如果Rabbit只管按顺序把消息发到各个消费者身上，不考虑消费者负载的话，很可能出现，一个机器配置不高的消费者那里堆积了很多消息处理不完，同时配置高的消费者却一直很轻松。为解决此问题，可以在各个消费者端，配置perfetch=1,意思就是告诉RabbitMQ在我这个消费者当前消息还没处理完的时候就不要再给我发新消息了。

channel.basic_qos(prefetch_count=1)

带消息持久化+公平分发的完整代码

生产者端

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

PublishSubscribe消息发布

如果你的消息想被所有的Queue收到，类似广播的效果，这时候就要用到exchange了。

fanout: 所有bind到此exchange的queue都可以接收消息
direct: 通过routingKey和exchange决定的那个唯一的queue可以接收消息
topic:所有符合routingKey(此时可以是一个表达式)的routingKey所bind的queue可以接收消息

生产端:

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()

消费端：

#_*_coding:utf-8_*_
__author__ = 'Alex Li'
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名字,rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除
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()

有选择的接收消息

队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据 关键字 判定应该将数据发送至指定队列。

生产端：

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()

消费端：

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]
" % 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()

更细致的消息过滤

参考：https://www.cnblogs.com/alex3714/articles/5248247.html