rabbitmq

rabbitmq的安装（ubuntu）：

echo 'deb http://www.rabbitmq.com/debian/ testing main' |      
           sudo tee /etc/apt/sources.list.d/rabbitmq.list

wget -O- https://www.rabbitmq.com/rabbitmq-signing-key-public.asc |
   sudo apt-key add -

sudo apt-get update
sudo apt-get install rabbitmq-server

#启动
sudo /etc/init.d/rabbitmq start

rabbitmq三种模式：

一. Direct Exchage:

1. 可以不绑定exchange，消息传递时需要一个'routeKey'

2. 消息会被发送到RouteKey中指定的队列，如果不存在则抛弃消息

二. Fanout Exchange

1. 不需要routekey

2. 需要将exchange与queue绑定，一个exchange可以绑定多个queue,一个queue也可以绑定到多个exchange

3. 如果接受到消息的exchange没有与任何queue绑定，则消息会被抛弃

三. Topic Exchange

1. 每个队列都有其关心的主题，所有的消息都带有routekey, 消息会被转发到关注主题与routekey模糊匹配的队列

2. ‘#’表示0个或多个关键字， ’*‘ 表示一个关键字

例如：

#.a 匹配a.a, aa.a, aaa.a等

*.a 匹配a.a, b.a, c.a等

最简单的队列通信：

send.py

#!/usr/bin/env python3

import pika

#连接rabbitmq
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

#生成一个管道，在管道中跑各种队列
channal = connection.channel()

#声明queue， queue默认会在rabbitmq重启后丢失；需要注意的是queue一旦声明就不允许修改了，
#比如从非持久化修改成持久化
channal.queue_declare(queue='hello')      #创建队列hello

#下面这句是将queue持久化，rabbitmq重启后不会丢失，但queue中未被消费的数据会丢失
#channal.queue_declare(queue='hello', durable=True)

#rabbitmq不能直接发送数据到队列里，需要一个Exchange， Exchange为空则使用默认Exchange.
#routing_key是队列名称，body是数据内容
channal.basic_publish(exchange='', routing_key='hello', body='Hello World!')

print('Sent "Hello Wrold!"')

#关闭连接
connection.close()

recv.py

#!/usr/bin/env python3

import pika

#连接rabbitmq
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

#生成一个管道，在管道里跑各种队列
channal = connection.channel()

#这里又生成一个队列，是说如果发送端定义了队列，这里就忽略；
#如果接收端启动的时候，发送端还没有启动，它就创建一个队列，代码不会出错
channal.queue_declare(queue='hello')

#这里的参数必须这样定义
def callback(ch, method, properties, body):
    print('Received %r' % body)
    #如果是持久化队列就需要加上面面这句，保证消息持久化
    #ch.basic_ack(delivery_tag=method.delivery_tag)

#rabbitmq如果有多个消费者，默认情况下，消息会依次分发给各个消费者，但是消费端往往
#处理能力不同，这种模式就会造成处理能力低的消费都信息堆积，处理能力高的却空闲着。
#为了解决这个问题，可以在各个消费者端，配置prefetch， 意思是告诉rabbitmq在我这个消费者当前消费还没有处理完的时候不要再给我发消息了。
channal.basic_qos(prefetch_count=1)

#no_ack的意思是不发送接收回执， 设置成False能保证客户端正确接收
channal.basic_consume(callback, queue='hello', no_ack=False)

#开始接收
channal.start_consuming()

direct模式，绑定Exchange

发布端：

import pika
import sys

#连接rabbitmq
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
#创建管道
channel = connection.channel()

#创建direct Exchange， 名称direct_logs
channel.exchange_declare(exchange='direct_logs', type='direct')

#定义routing_key，设置默认值为info
severity = sys.argv[1] if len(sys.argv) > 1 else 'info'

#要发送的消息， 默认为hello world
message = ' '.join(sys.argv[2:]) or 'hello world!'

#exchange和routing_key共同决定发送到哪个队列
channel.basic_publish(exchange='direct_logs', routing_key=serverity, body=message)

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

#绑定exchange， 这样循环是因为可以接收好几种日志类型，比如执行 python3 接收端.py info warning error
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('%r:%r' % (method.routing_key, body))

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

channel.start_consuming()

发布端运行:

$ python3 发布端.py
Sent 'info':'hello world!'

$ python3 发布端.py info warning 1111
Sent 'info':'warning 1111'

$ python3 发布端.py error 22222
Sent 'error':'22222'

接收端运行:

$ python3 接收端.py info warning error
waiting for logs To exit press CTRL+C
'info':b'hello world!'
'info':b'warning 1111'
'error':b'22222'