RabbitMQ

目录

• RabbitMQ
  • 模式
    • 简单模式
  • 参数使用
    • 应答参数
    • 持久化参数
    • 分发参数
  • 交换机模式(routingkey和exchange同时满足)
    • 发布订阅
    • 关键字
    • 通配符

RabbitMQ

好处： 解耦，异步，流量削峰

阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

简单命令

rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl start_app
rabbitmqctl list_queues

默认端口号 15672

模式

简单模式（最广泛）

参数

交换机模式

​ -- 发布订阅

​ -- 关键字模式

-- 模糊匹配模式

简单模式

• 连接rabbitmq
• 创建队列
• 向指定的队列插入数据

**生产者 **basic_publish

import pika
 
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
 
channel.queue_declare(queue='hello')
 
channel.basic_publish(exchange='',  # 简单模式 交换机为空
                      routing_key='hello',  # 指定队列
                      body='Hello World!')  # 向指定队列插入内容
 
print(" [x] Sent 'Hello World!'")

**消费者 **basic_consume

import pika
 
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
 
channel.queue_declare(queue='hello')
 
 
def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
 
 
channel.basic_consume(queue='hello',
                      auto_ack=True , #自动应答
                      on_message_callback=callback)
 
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
 

参数使用

应答参数

模拟消费者出问题

生产者生产数据给消费者，消费者取到数据则队列无数据，若消费者出bug ，再次启动，则取不到数据了

auto_ack=False  #改默认应答为手动应答
ch.basic_ack(delivery_tag=method.delivery_tag)   # 给信号

原理

生产者生产数据放入队列，消费者改手动应答，消费者取数据。队列还会保留一份数据，当消费者发出信号后ch.basic_ack(delivery_tag=method.delivery_tag) ，队列则删除数据

手动应答牺牲效率

注重效率则默认应答

持久化参数

模拟队列（rabbitmq)出问题

#声明queue
channel.queue_declare(queue='hello2', durable=True)  # durable=True 声明可持久化的队列
                                                    # 但是放数据的时候还得指定是否持久化
                                                    properties=pika.BasicProperties(
                                                              delivery_mode=2,  #持久化参数2
                                                      )
 
channel.basic_publish(exchange='',
                      routing_key='hSSello2',
                      body='Hello World!',
                      properties=pika.BasicProperties(
                          delivery_mode=2,  # make message persistent
                          )
                      )                              # 但是放数据的时候还得指定是否持久化
                                                    #properties=pika.BasicProperties(
                                                      #        delivery_mode=2,  # make                                                         #    message persistent
                                                        #      )

分发参数

改为手动应应答并且加上 channel.basic_qos(prefetch_count=1) # 消费者 加上这一句话

默认轮询分发，一人一个

轮询分发（round-robin）不管谁忙，都不会多给消息，总是你一个我一个。想要做到公平分发（fair dispatch），必须关闭自动应答ack，改成手动应答。使用basicQos(perfetch=1)限制每次只发送不超过1条消息到同一个消费者，消费者必须手动反馈告知队列，才会发送下一个。

 channel.basic_qos(prefetch_count=1)  # 消费者 加上这一句话

import pika

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

channel.queue_declare(queue='hello')


def callback(ch, method, properties, body):
    import time
    time.sleep(4)
    print(" [x] Received %r" % body)
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_qos(prefetch_count=1)  ########
channel.basic_consume(queue='hello',
                      auto_ack=False,
                      on_message_callback=callback)


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

交换机模式(routingkey和exchange同时满足)

基于交换机通信（容器） 由生产者创建，向交换机插入数据

消费者创建队列 队列绑定交换机

发布订阅

# 生产者
import pika
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()
 
channel.exchange_declare(exchange='logs',          ## 交换机模式 名字任意
                         exchange_type='fanout') ## fanout 发布订阅模式
 
message = "info: 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') ## fanout 发布订阅模式
 
# 消费者创建队列
result = channel.queue_declare("",exclusive=True) # 随机名字  "" , 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(queue=queue_name,
                      auto_ack=True,
                      on_message_callback=callback)
 
channel.start_consuming()

关键字

生产者放一个关键字 消费者放一个关键字

匹配成功后 则给消费者

# 生产者
import pika
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()
 
channel.exchange_declare(exchange='logs',          ## 交换机模式 名字任意
                         exchange_type='direct') ## direct 关键字模式
 
message = "info: Hello World!"
channel.basic_publish(exchange='logs',
                      routing_key='xxx', # 绑定关键字
                      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='direct') ## direct 关键字模式
 
# 消费者创建队列
result = channel.queue_declare("",exclusive=True) # 随机名字  "" , exclusive=True
queue_name = result.method.queue  # 拿到随机名字
 
# 绑定到交换机上
channel.queue_bind(exchange='logs',
                   queue=queue_name,routing_key='xxx') # routing_key 绑定关键字
[ 可以绑定多个关键字                 
channel.queue_bind(exchange='logs',
                   queue=queue_name,routing_key='xxx') # channel.queue_bind(exchange='logs',
                   queue=queue_name,routing_key='xxx') #
 
                   ]
 
print(' [*] Waiting for logs. To exit press CTRL+C')
 
def callback(ch, method, properties, body):
    print(" [x] %r" % body)
 
 
channel.basic_consume(queue=queue_name,
                      auto_ack=True,
                      on_message_callback=callback)
 
channel.start_consuming()

通配符

同关键字相比，模糊匹配

#  一个多个单词
* 一个单词
 
uss.#
#.news
#.weather

 
# 生产者
import pika
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()
 
channel.exchange_declare(exchange='logs3',
                         exchange_type='topic')
 
message = "info: Hello ERU!"
channel.basic_publish(exchange='logs3',
                      routing_key='europe.weather',
                      body=message)
print(" [x] Sent %r" % message)
connection.close()
 
# 消费者
 
import pika
import sys
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost'))
channel = connection.channel()
 
channel.exchange_declare(exchange='logs3',
                         exchange_type='topic')
 
result = channel.queue_declare("",exclusive=True)
queue_name = result.method.queue
 
 
 
channel.queue_bind(exchange='logs3',
                   queue=queue_name,
                   routing_key="#.news")
 
print(' [*] Waiting for logs. To exit press CTRL+C')
 
def callback(ch, method, properties, body):
    print(" [x] %r" % body)
 
 
channel.basic_consume(queue=queue_name,
                      auto_ack=True,
                      on_message_callback=callback)
 
channel.start_consuming()