RabbitMQ介绍与python操作RabbitMQ

Source

RabbitMQ介绍

RabbitMQ基本使用

ack机制

持久化

闲置消费

RabbitMQ介绍

MQ：消息队列，就是基础数据结构中的先进先出的一种数据结构。像生活中买东西，需要排队，先排的人先买消费，就是典型的先进先出。

Linux安装

# 安装erlang
yum -y install erlang
# 安装RabbitMQ
yum -y install rabbitmq-server

windows安装

【Windows安装RabbitMQ详细教程】_慕之寒的博客-CSDN博客_rabbitmq安装windows

docker安装

# 拉取镜像
docker pull rabbitmq:management
# 启动，设置用户名为admin，密码为admin
docker run -di --name Myrabbitmq -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -p 15672:15672 -p 5672:5672 rabbitmq:management
# 访问服务器的15672端口

RabbitMQ基本使用

安装模块：

pip install pika

消息进队列：

import pika

# 没有用户名密码情况
# connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.0.0.200',port=5672))

# 有密码的情况
credentials = pika.PlainCredentials("admin", "admin")
connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.0.0.200', port=5672, credentials=credentials))
channel = connection.channel()

# 声明一个消息队列叫hello
channel.queue_declare(queue='hello')
# 往消息队列中放一条消息
# routing_key必须等于队列的名字，body是发送的消息
channel.basic_publish(exchange='', routing_key='hello', body='你好')
print("send success")
connection.close()

队列出消息：

import pika

def main():
    credentials = pika.PlainCredentials("admin", "admin")
    connection = pika.BlockingConnection(
        pika.ConnectionParameters(host='10.0.0.200', port=5672, credentials=credentials)
    )
    channel = connection.channel()

    # 创建一个hello的对列
    channel.queue_declare(queue='hello')

    # 回调函数，队列中有消息就会触发
    def callback(ch, method, properties, body):
        print("来自队列：%r" % str(body, encoding='utf8'))

    channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=True)
    print('等待中...')
    channel.start_consuming()

if __name__ == '__main__':
    main()

ack机制

消费者收到消息后，要回复给消息队列，消息队列收到回复，该消息就删除，如果收不到，就一直放在消息队列中，再起一个消费者，还会消费这个消息。

自动

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

auto_ack参数为True指的是收到消息后自动回复给消息队列。

手动

# 回调函数，队列中有消息就会触发
def callback(ch, method, properties, body):
    print("来自队列：%r" % str(body, encoding='utf8'))
    # 数据使用完了后在回复给队列
    ch.basic_ack(delivery_tag=method.delivery_tag)

持久化

消息是放在内存中的，如果RabbitMQ服务挂掉了，消息就没了，通过持久化，保证队列和消息都不丢失。

队列持久化要是一个新队列

# 声明一个新消息队列叫Q，做持久化
channel.queue_declare(queue='Q', durable=True)
# 消息也做持久化
channel.basic_publish(
    exchange='',
    routing_key='hello',
    body='你好',
    properties=pika.BasicProperties(
        delivery_mode=2,  # 消息也做持久化
    )
)

生产者：

import pika

# 没有用户名密码情况
# connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.0.0.200',port=5672))

# 有密码的情况
credentials = pika.PlainCredentials("admin", "admin")
connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.0.0.200', port=5672, credentials=credentials))
channel = connection.channel()

# 声明一个新消息队列叫Q，做持久化
channel.queue_declare(queue='Q', durable=True)
# 往消息队列中放一条消息
# routing_key必须等于队列的名字，body是发送的消息
channel.basic_publish(
    exchange='',
    routing_key='hello',
    body='你好',
    properties=pika.BasicProperties(
        delivery_mode=2,  # 消息也做持久化
    )
)
print("send success")
connection.close()

闲置消费

正常情况如果有多个消费者，是按照顺序第一个消息给第一个消费者，第二个消息给第二个消费者。

但是可能第一个消息的消费者处理消息很耗时，一直没结束，就可以让第二个消费者优先获得闲置的消息

channel.basic_qos(prefetch_count=1) # 谁闲置谁获取，没必要按照顺序一个一个来

消费者：

import pika

def main():
    credentials = pika.PlainCredentials("admin", "admin")
    connection = pika.BlockingConnection(
        pika.ConnectionParameters(host='10.0.0.200', port=5672, credentials=credentials)
    )
    channel = connection.channel()

    # 创建一个hello的对列
    channel.queue_declare(queue='hello')

    # 回调函数，队列中有消息就会触发
    def callback(ch, method, properties, body):
        print("来自队列：%r" % str(body, encoding='utf8'))
        # 数据使用完了后在回复给队列
        ch.basic_ack(delivery_tag=method.delivery_tag)

    channel.basic_qos(prefetch_count=1)  # 谁闲置谁获取，没必要按照顺序一个一个来
    channel.basic_consume(queue='hello', on_message_callback=callback)
    print('等待中...')
    channel.start_consuming()

if __name__ == '__main__':
    main()