rabbitmq 的安装与python集成

docker安装rabbitmq

docker pull rabbitmq
docker run -d --hostname rabbitmq --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq
docker exec -it rabbitmq bash
cd /usr/sbin
rabbitmq-plugins enable rabbitmq_management

Exchange 一共有三种工作模式：fanout, direct, topicd
模式一：fanout
这种模式下，传递到 exchange 的消息将会转发到所有与其绑定的 queue 上。
不需要指定 routing_key ，即使指定了也是无效。
需要提前将 exchange 和 queue 绑定，一个 exchange 可以绑定多个 queue，一个queue可以绑定多个exchange。
需要先启动 订阅者，此模式下的队列是 consumer 随机生成的，发布者 仅仅发布消息到 exchange ，由 exchange 转发消息至 queue。
模式二：direct
这种工作模式的原理是 消息发送至 exchange，exchange 根据路由键（routing_key）转发到相对应的 queue 上。
 可以使用默认 exchange =' ' ，也可以自定义 exchange
这种模式下不需要将 exchange 和 任何进行绑定，当然绑定也是可以的。可以将 exchange 和 queue ，routing_key 和 queue 进行绑定
传递或接受消息时 需要指定 routing_key
需要先启动订阅者，此模式下的队列是 consumer 随机生成的，发布者仅仅发布消息到 exchange ，由 exchange 转发消息至 queue。
模式三：topicd
这种模式和第二种模式差不多，exchange 也是通过 路由键 routing_key 来转发消息到指定的 queue 。 不同点是routing_key 使用正则表达式支持模糊匹配，但匹配规则又与常规的正则表达式不同，比如‘’#‘’是匹配全部，“*”是匹配一个词。
无交换机模式

#server
import pika
connection=pika.BlockingConnection(pika.ConnectionParameters(host="192.168.42.75",port=5672))
channel=connection.channel()
channel.queue_declare(queue="test",durable=True) #durable代表是否持久化
print("public queue test and body")
while 1:
    line=input()
    channel.basic_publish(exchange='',routing_key='test',body=line) # #注意当未定义exchange时，routing_key需和queue的值保持一致

connection.close()

#client
import pika
connection=pika.BlockingConnection(pika.ConnectionParameters(host="192.168.42.75",port=5672))
channel=connection.channel()
channel.queue_declare(queue="test",durable=True) #durable代表是否持久化
def callback(ch,method,prop,body):
    print("recv: %s" % body)

channel.basic_consume(on_message_callback=callback,queue='test',auto_ack=True) #auto_ack 收到后是否删除
channel.start_consuming()

交换机模式下-Fanout ，此模式下不需要routing_key，直接绑定队列

#server
import pika

connection=pika.BlockingConnection(pika.ConnectionParameters(host="192.168.42.75",port=5672))
channel=connection.channel()
channel.exchange_declare(exchange="ex1",exchange_type="fanout") #fanout,direct,topic
print("请输入数据")
while 1:
    line = input()
    channel.basic_publish(exchange="ex1",routing_key='',body=line)
connection.close()

#client
import pika

connection=pika.BlockingConnection(pika.ConnectionParameters(host="192.168.42.75",port=5672))
channel=connection.channel()
channel.exchange_declare(exchange="ex1",exchange_type="fanout") #fanout,direct,topic
queue=channel.queue_declare('',exclusive=True)
queue=queue.method.queue
channel.queue_bind(exchange='ex1',queue=queue)

def callback(ch,method,prop,body):
    print(body)

channel.basic_consume(queue='eq1',auto_ack=True,on_message_callback=callback)
channel.start_consuming()

交换机模式下-direct关键词绑定

#server
import pika
conn=pika.BlockingConnection(pika.ConnectionParameters(host="192.168.42.75",port=5672))
channel=conn.channel()
channel.exchange_declare(exchange="ex2",exchange_type="direct")
while 1:
    line=input()
    channel.basic_publish(exchange="ex2",routing_key="info2",body=line)
conn.close()

#client
import pika
conn=pika.BlockingConnection(pika.ConnectionParameters(host="192.168.42.75",port=5672))
channel=conn.channel()
channel.exchange_declare(exchange="ex2",exchange_type="direct")
queue=channel.queue_declare('',exclusive=True)
queue=queue.method.queue
channel.queue_bind(exchange="ex2",queue=queue,routing_key="info2")
def callback(ch,method,prop,body):
    print(body)
channel.basic_consume(queue=queue,auto_ack=False,on_message_callback=callback)
channel.start_consuming()

交换机模式下-Topic，此模式下可以使用模糊匹配的关键词，实现大小类关联绑定

#server
import pika
conn=pika.BlockingConnection(pika.ConnectionParameters(host="192.168.42.75",port=5672))
channel=conn.channel()
channel.exchange_declare(exchange="ex3",exchange_type="topic")
while 1:
    line=input()
    channel.basic_publish(exchange="ex3",routing_key="info.error",body=line)
conn.close()

#client
import pika
conn=pika.BlockingConnection(pika.ConnectionParameters(host="192.168.42.75",port=5672))
channel=conn.channel()
channel.exchange_declare(exchange="ex3",exchange_type="topic")
queue=channel.queue_declare('',exclusive=True)
queue=queue.method.queue
channel.queue_bind(exchange="ex3",queue=queue,routing_key="info.#")
def callback(ch,method,prop,body):
    print(body)
channel.basic_consume(queue=queue,auto_ack=False,on_message_callback=callback)
channel.start_consuming()